SpringBoot+Nacos+Kafka简单实现微服务流编排
Java后端技术
共 10863字,需浏览 22分钟
· 2022-06-25
往期热门文章:
1、几行代码,搞定 SpringBoot 接口恶意刷新和暴力请求! 2、程序员坐牢了,会被安排写代码吗? 3、同事写了一个责任链模式,Bug无数! 4、40 个 SpringBoot 常用注解:让生产力爆表! 5、八股文天花板!(PDF高清下载) 文章来源:https://c1n.cn/RWt2e
目录
前言 准备工作 总结
前言
准备工作
| Nacos 安装及使用入门
自己学习的话推荐使用 docker 安装,命令如下:
拉取镜像:
docker pull nacos/nacos-server
创建服务:
docker run --env MODE=standalone --name nacos -d -p 8848:8848 nacos/nacos-server
| 准备三个 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
| 业务解读
我们现在需要对三个服务进行编排,保障每个服务可以插拔,也可以调整服务的位置。
示意图如上:
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();
}
}
}
}
总结
流编排的思路整体来说就是数据流方向可调,我们以此为需求,根据一些主流框架提供的 api 实现自己的动态调整方案,可以帮助自己更好的理解流编码思想及原理。
在实际业务中,还有许多业务问题需要去突破,我们这样处理更多是因为服务可插拔,便于流处理微服务在项目灵活搭配。
因为我现在工作是在传统公司,由于一些原因很难去推动新框架的使用,经常会用一些现有技术栈组合搞一些 sao 操作,供大家参考,希望大家多多指教。
往期热门文章:
1、MySQL 暴跌! 2、超越 Xshell!号称下一代 Terminal 终端神器,用完爱不释手! 3、IDEA 官宣全新默认 UI,太震撼了!! 4、让你直呼「卧槽」的 GitHub 项目! 5、Kafka又笨又重,为啥不选Redis? 6、50多个高频免费 API 接口分享 7、IDEA公司再发新神器!超越 VS Code 骚操作! 8、我怀疑这是 IDEA 的 BUG,但是我翻遍全网没找到证据! 9、Spring MVC 中的 Controller 是线程安全的吗? 10、Gitee 倒下了???
评论
偷偷告诉你如何一台电脑开多个微信!
大家好,我是轩辕。前几天在粉丝群里,有人问我是怎么在一台电脑上同时登录两个微信的?正好之前写过一篇文章,分析过原理,分享给没看过的小伙伴学习一下。手机端多开微信估计很多人都知道,像华为、小米等手机系统都对此做了支持,不过在运行Windows系统的电脑上怎么启动两个微信呢?其实很简单,你只需要写一个批
编程技术宇宙
0
15种时间序列预测方法总结(包含多种方法代码实现)
向AI转型的程序员都关注了这个号👇👇👇在这篇文章中,我们将深入探讨时间序列预测的基本概念和方法。我们将首先介绍单元预测和多元预测的概念,然后详细介绍各种深度学习和传统机器学习方法如何应用于时间序列预测,包括循环神经网络(RNN)、一维卷积神经网络(1D-CNN)、Transformer、自回归模型(
机器学习AI算法工程
0
好未来测开一面,挺简单!(0428面试原题解析)
大家好,我是二哥呀。今天继续给大家分享春招面试题《好未来测开一面原题》,附详细答案,我会用通俗易懂+手绘图的方式,让天下所有的面渣都能逆袭 😁二哥的 Java 面试指南内容较长,建议正在冲刺 24 届春招和 25 届暑期实习、秋招的同学先收藏起来,面试的时候大概率会碰到,1、二哥的 Linux 速查
沉默王二
0
SpringBoot 实现图片防盗链功能
程序员的成长之路互联网/程序员/技术/资料共享 关注阅读本文大概需要 4 分钟。来自:blog.csdn.net/weixin_46157208/article/details/138051737前言出于安全考虑,我们需要后端返回的图片只允许在某个网站内展示,不想被爬虫拿到图片地
程序员的成长之路
0
Java8 Steam流太难用了?那你可以试试 JDFrame
来源:juejin.cn/post/7356652717392740404👉 欢迎加入小哈的星球 ,你将获得: 专属的项目实战 / Java 学习路线 / 一对一提问 / 学习打卡 / 赠书福利全栈前后端分离博客项目 2.0 版本完结啦, 演示链接
小哈学Java
1
springboot第70集:字节跳动后端三面经,一文让你走出微服务迷雾架构周刊
创建一个使用Kubernetes (K8s) 和 Jenkins 来自动化 GitLab 前端项目打包的CI/CD流水线,需要配置多个组件。下面,我将概述一个基本的设置步骤和示例脚本,以帮助你理解如何使用这些工具整合一个自动化流程。前提条件确保你已经有:Kubernetes 集群:用于部署 Jenk
程序源代码
0
一站式解决方案:基于 Arthas 实现服务发现和权限控制
来源:juejin.cn/post/7281849496983994383👉 欢迎加入小哈的星球 ,你将获得: 专属的项目实战 / Java 学习路线 / 一对一提问 / 学习打卡 / 赠书福利全栈前后端分离博客项目 2.0 版本完结啦, 演示链接
小哈学Java
0
用 Shader 实现旗帜飘扬动画效果
我觉得对于刚入门 3D 编程的朋友来说,如果能够完成代码创建模型数据->创建材质->编写Shader动画这一系列,想必会有满满的成就感。今天就用 Cocos Creator 的 utils.MeshUtils.createMesh 接口,带大家感受一下这个流程。这个流程不仅可以用于新手学
COCOS
2