предисловие
В последнее время я занимаюсь разработкой микросервисов, включая разработку некоторых модулей обработки данных.Каждый обрабатывающий бизнес будет разрабатывать независимые микросервисы, что удобно для последующего расширения и оркестрации потоков.Я изучил такие фреймворки, как Spring Cloud Data Flow, и чувствую что этот фреймворк для нас.Он слишком тяжелый и более хлопотный в обслуживании,поэтому по идее оркестровки потоков мы реализуем простую функцию оркестровки потоков на основе нашего текущего стека технологий.
简单的说，我们希望自己的流编排就是微服务可插拔，微服务数据入口及输出可不停机修改。
Готов к работе
Установка Nacos и начало работы
Если вы учитесь самостоятельно, рекомендуется использовать установку docker, команда выглядит следующим образом
- Вытягивание зеркала Docker Pull NACO / NACOS-Server
- Создать службу docker run --env MODE=standalone --name nacos -d -p 8848:8848 nacos/nacos-server
Затем введите ip:8848/nacos account nacos password nacos в браузере
docker能够帮助我们快速安装服务，减少再环境准备花的时间
Подготовить три сервиса SpringBoot, представить Nacos и Kafka
<parent>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-parent</artifactId>
<version>2.1.0.RELEASE</version>
</parent>
<dependency>
<groupId>org.springframework.kafka</groupId>
<artifactId>spring-kafka</artifactId>
</dependency>
<dependency>
<groupId>com.alibaba.boot</groupId>
<artifactId>nacos-config-spring-boot-starter</artifactId>
<version>0.2.1</version>
</dependency>
конфигурационный файл
spring:
kafka:
bootstrap-servers: kafka-server:9092
producer:
acks: all
consumer:
group-id: node1-group #三个服务分别为node1 node2 node3
enable-auto-commit: false
# 部署的nacos服务
nacos:
config:
server-addr: nacos-server:8848
建议配置本机host 就可以填写xxx-server 不用填写服务ip
Деловая интерпретация
Теперь нам нужно организовать три службы, чтобы гарантировать, что каждую службу можно подключать и отключать, а также можно настроить место службы.Схема выглядит следующим образом:
- Служба node1 прослушивает поток данных, отправленный внешней службой, а входная тема является выходной темой внешней службы данных.
- Узел 2 прослушивает данные, обрабатываемые узлом 1, поэтому тема, отслеживаемая узлом 2, является темой, выводимой узлом 1, и то же самое верно для узла 3. Наконец, узел 3 отправляет данные в конец потока данных после завершения обработки.
- Теперь нам нужно настроить процесс для удаления node2-server, нам нужно только изменить node1-sink на node2-sink, чтобы наши сервисы можно было гибко встраивать в бизнес обработки потоков данных разных проектов, чтобы Plug and play ( конечно, форматы данных и эти бизнес-аспекты должны быть согласованы)
- Динамическая настройка также может гарантировать, что при возникновении проблемы с определенным узлом службы направление потока данных может быть немедленно изменено, например, отправлено в службу временного хранения, чтобы избежать накопления слишком большого количества данных в Kafka и несбалансированная пропускная способность
Конфигурация Nacos
Создать конфигурацию
Обычно каждый сервис в потоковой оркестрации имеет вход и выход, которые являются входом и приемником соответственно, поэтому нам нужно настроить две темы для каждой службы, а именно вход-тема-выход-тема, мы добавляем конфигурацию ввода и вывода в nacos
nacos配置项需要配置groupId,dataId,通常我们用服务名称作为groupId,配置项的名称作为dataId,如node1-server服务有一个input配置项，配置如下:
完成其中一个服务的配置，其它服务参考下图配置即可
чтение конфигурации
@Configuration
@NacosPropertySource(dataId = "input", groupId = "node1-server", autoRefreshed = true)
// autoRefreshed=true指的是nacos中配置发生改变后会刷新，false代表只会使用服务启动时候读取到的值
@NacosPropertySource(dataId = "sink", groupId = "node1-server", autoRefreshed = true)
public class NacosConfig {
@NacosValue(value = "${input:}", autoRefreshed = true)
private String input;
@NacosValue(value = "${sink:}", autoRefreshed = true)
private String sink;
public String getInput() {
return input;
}
public String getSink() {
return sink;
}
}
Отслеживание изменений конфигурации
Входные данные службы должны создавать потребителя при запуске службы, повторно создавать потребителя при изменении темы и удалять потребителя старой темы.Вывод ориентирован на бизнес, нет необходимости отслеживать изменения и читать каждый раз, когда он отправляется. Это последняя настроенная тема, поскольку autoRefreshed = true в приведенном выше классе конфигурации, это только обновит значение конфигурации в nacosConfig, служба должна знать об изменении конфигурации, чтобы управлять бизнесом по созданию потребления, и нуждается создать монитор конфигурации nacos
/**
* 监听Nacos配置改变，创建消费者，更新消费
*/
@Component
public class ConsumerManager {
@Value("${spring.kafka.bootstrap-servers}")
private String servers;
@Value("${spring.kafka.consumer.enable-auto-commit}")
private boolean enableAutoCommit;
@Value("${spring.kafka.consumer.group-id}")
private boolean groupId;
@Autowired
private NacosConfig nacosConfig;
@Autowired
private KafkaTemplate kafkaTemplate;
// 用于存放当前消费者使用的topic
private String topic;
// 用于执行消费者线程
private ExecutorService executorService;
/**
* 监听input
*/
@NacosConfigListener(dataId = "node1-server", groupId = "input")
public void inputListener(String input) {
// 这个监听触发的时候 实际NacosConfig中input的值已经是最新的值了 我们只是需要这个监听触发我们更新消费者的业务
String inputTopic = nacosConfig.getInput();
// 我使用nacosConfig中读取的原因是因为监听到内容是input=xxxx而不是xxxx,如果使用需要自己截取一下,nacosConfig中的内容框架会处理好，大家看一下第一张图的配置内容就明白了
// 先检查当前局部变量topic是否有值,有值代表是更新消费者，没有值只需要创建即可
if(topic != null) {
// 停止旧的消费者线程
executorService.shutdownNow();
executorService == null;
}
// 根据为新的topic创建消费者
topic = inputTopic;
ThreadFactory threadFactory = new ThreadFactoryBuilder().setNameFormat(topic + "-pool-%d").build();
executorService = new ThreadPoolExecutor(1, 1, 0L, TimeUnit.MILLISECONDS, new LinkedBlockingQueue<Runnable>(2), threadFactory);
// 执行消费业务
executorService.execute(() -> consumer(topic));
}
/**
* 创建消费者
*/
public void consumer(String topic) {
Properties properties = new Properties();
properties.put("bootstrap.servers", servers);
properties.put("enable.auto.commit", enableAutoCommit);
properties.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
properties.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
properties.put("group.id", groupId);
KafkaConsumer<String, String> consumer = new KafkaConsumer<>(properties);
consumer.subscribe(Arrays.asList(topic));
try {
while (!Thread.currentThread().isInterrupted()) {
Duration duration = Duration.ofSeconds(1L);
ConsumerRecords<String, String> records = consumer.poll(duration);
for (ConsumerRecord<String, String> record : records) {
String message = record.value();
// 执行数据处理业务 省略业务实现
String handleMessage = handle(message);
// 处理完成后发送到下一个节点
kafkaTemplate.send(nacosConfig.getSink(), handleMessage);
}
}
consumer.commitAsync();
}
} catch (Exception e) {
LOGGER.error(e.getMessage(), e);
} finally {
try {
consumer.commitSync();
} finally {
consumer.close();
}
}
}
}
Суммировать
Идея оркестровки потоков в целом заключается в том, что направление потока данных можно корректировать.Мы используем это как требование и реализуем собственную схему динамической корректировки на основе API-интерфейсов, предоставляемых некоторыми основными фреймворками, которые могут помочь нам лучше понять идею и принцип потокового кодирования.Есть еще много бизнес-задач, которые необходимо решить.Мы занимаемся этим больше, потому что сервисы подключаемые, что удобно для гибкого коллокации микросервисов потоковой обработки в проекте.Потому что я Работая сейчас в традиционной компании, трудно продвигать новые по некоторым причинам.Для использования фреймворка некоторые операции sao часто используются в сочетании с некоторыми существующими технологическими стеками для вашей справки.Я надеюсь, что вы можете дать мне больше советов .