SpringBoot+Nacos+Kafka微服务流编排的简单实现
作者:热黄油啤酒 发布时间:2023-03-21 10:34:53
目录
前言
准备工作
Nacos安装及使用入门
准备三个SpringBoot服务,引入Nacos及Kafka
业务解读
Nacos配置
创建配置
读取配置
监听配置改变
总结
前言
最近一直在做微服务开发,涉及了一些数据处理模块的开发,每个处理业务都会开发独立的微服务,便于后面拓展和流编排,学习了SpringCloud Data Flow等框架,感觉这个框架对于我们来说太重了,维护起来也比较麻烦,于是根据流编排的思想,基于我们目前的技术栈实现简单的流编排功能。
简单的说,我们希望自己的流编排就是微服务可插拔,微服务数据入口及输出可不停机修改。
准备工作
Nacos安装及使用入门
自己学习的话推荐使用docker安装,命令如下
拉取镜像 docker pull nacos/nacos-server
创建服务 docker run --env MODE=standalone --name nacos -d -p 8848:8848 nacos/nacos-server
然后在浏览器输入 ip:8848/nacos 账号nacos 密码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服务监听前置服务发送的数据流,输入的topic为前置数据服务输出topic
node2监听node1处理后的数据,所以node2监听的topic为node1输出的topic,node3同理,最终node3处理完成后将数据发送到数据流终点
我们现在要调整流程 移除node2-server,我们只需要把node1-sink改变成node2-sink即可,这样我们这几个服务就可以灵活的嵌入的不同项目的数据流处理业务中,做到即插即用(当然,数据格式这些业务层面的都是需要约定好的)
动态可调还可以保证服务某一节点出现问题时候,即时改变数据流向,比如发送到数暂存服务,避免Kafka中积累太多数据,吞吐不平衡
Nacos配置
创建配置
通常流编排里面每个服务都有一个输入及输出,分别为input及sink,所以每个服务我们需要配置两个topic,分别是input-topic output-topic,我们就在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;
}
}
监听配置改变
服务的输入需要在服务启动时候创建消费者,在topic发生改变时候重新创建消费者,移除旧topic的消费者,输出是业务驱动的,无需监听改变,在每次发送时候读取到的都是最新配置的topic,因为在上面的配置类中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();
}
}
}
}
来源:https://juejin.cn/post/6997704312835047438
猜你喜欢
- 一、什么是热部署?热部署,就是在应用正在运行的时候升级软件,却不需要重新启动应用。二、什么是SpringBoot热部署?SpringBoot
- 排查@CacheEvict注解失效我简单看了一下《Spring实战》中的demo,然后就应用到业务代码中了,本以为如此简单的事情,竟然在代码
- 项目上线之后,如果日志打印的很模糊或者业务逻辑比较复杂,有时候无法定位具体的错误原因,因此可以通过IDEA远程代理进行Debug。线上的代码
- public static String getCharset(File file) { &n
- String 对象是不可改变的。每次使用 System.String 类中的方法之一时,都要在内存中创建一个新的字符串对象,这就需要为该新对
- 写在前面之前想尝试把JWT和Shiro结合到一起,但是在网上查了些博客,也没太有看懂,所以就自己重新研究了一下Shiro的工作机制,然后自己
- 前言Spring框架对Bean进行装配提供了很灵活的方式,下面归纳一下主要的方式:在XML中进行显示配置在Java中进行显示配置隐式的bea
- Maven是个很好用的管理工具,不经能够管理jar,还能实现打包。这里讲解Maven 本地打包,服务器打包,可以全部交给jenkins去完成
- springboot读取配置文件到静态工具类通常我们读取配置文件可以用@Value注解和@Configuration,@Configurat
- 里氏替换原则(LSP)定义:在任何父类出现的地方都可以用它的子类类替换,且不影响功能。解释说明:其实LSP是对开闭原则的一个扩展,在OO思想
- 本文实例为大家分享了java实现简单石头剪刀布游戏的具体代码,供大家参考,具体内容如下问题描述Alice, Bob和Cindy一起玩猜拳的游
- 一:简述如果我们想要生成一个随机数,通常会使用Random类。但是在并 * 况下Random生成随机数的性能并不是很理想,今天给大家介绍一下J
- 打包与运行在项目开发完成之后,可以直接用IDEA将其打包成JAR包运行,也可以打包成WAR包运行以便在多服务器、多配置环境下运行。双击cle
- 文章来源:aspcn 作者:孙雯重复和并发服务器这个应用程序被当作一个重复的服务器.因为它只有在处理完一个进程以后才会接受另一个连接.更多的
- 初学C++的朋友经常在类中看到public,protected,private以及它们在继承中表示的一些访问范围,很容易搞糊涂。今天本文就来
- 本文实例讲述了Java获得当前时间前指定几个小时具体时间的方法。分享给大家供大家参考,具体如下:package getBeforeHourD
- 本文实例讲述了Java String类简单用法。分享给大家供大家参考,具体如下:一 String类的实例化方式1 代码public clas
- 第一个SpringMvc HelloWorld无启动类springboot才有启动类前端操作访问项目名进入默认主页,显示一个链接点击链接后请
- 最近接触到个新项目,发现它用了一个比较有意思的框架,可以说实现了我刚入行时候的梦想,所以这里马不停蹄的和大家分享下。在我刚开始工作接触的项目
- 一、前言:垃圾回收:在未来的JDK中可能G1会为ZGC所取代先问自己几个问题:什么是垃圾?垃圾就是堆内存中(范指)没有任何指针指向的对象实体