点击上方“芋道源码”,选择“设为星标”
管她前浪,还是后浪?
能浪的浪,才是好浪!
每天 10:33 更新文章,每天掉亿点点头发...
源码精品专栏
最近一直在做微服务开发,涉及了一些数据处理模块的开发,每个处理业务都会开发独立的微服务,便于后面拓展和流编排。
学习了 SpringCloud Data Flow 等框架,感觉这个框架对于我们来说太重了,维护起来也比较麻烦,于是根据流编排的思想,基于我们目前的技术栈实现简单的流编排功能。
简单的说,我们希望自己的流编排就是微服务可插拔,微服务数据入口及输出可不停机修改。
基于 Spring Boot + MyBatis Plus + Vue & Element 实现的后台管理系统 + 用户小程序,支持 RBAC 动态权限、多租户、数据权限、工作流、三方登录、支付、短信、商城等功能。 项目地址:https://github.com/YunaiV/ruoyi-vue-pro
自己学习的话推荐使用 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 能够帮助我们快速安装服务,减少再环境准备花的时间。
<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
> 基于微服务的思想,构建在 B2C 电商场景下的项目实战。核心技术栈,是 Spring Boot + Dubbo 。未来,会重构成 Spring Cloud Alibaba 。
>
> 项目地址:<https://github.com/YunaiV/onemall>
# 部署的nacos服务
nacos:
config:
server-addr: nacos-server:8848
建议配置本机 host 就可以填写 xxx-server 不用填写服务 ip。
我们现在需要对三个服务进行编排,保障每个服务可以插拔,也可以调整服务的位置。
示意图如上:
通常流编排里面每个服务都有一个输入及输出,分别为 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 操作,供大家参考,希望大家多多指教。
欢迎加入我的知识星球,一起探讨架构,交流源码。加入方式,长按下方二维码噢:
已在知识星球更新源码解析如下:
最近更新《芋道 SpringBoot 2.X 入门》系列,已经 101 余篇,覆盖了 MyBatis、Redis、MongoDB、ES、分库分表、读写分离、SpringMVC、Webflux、权限、WebSocket、Dubbo、RabbitMQ、RocketMQ、Kafka、性能测试等等内容。
提供近 3W 行代码的 SpringBoot 示例,以及超 4W 行代码的电商微服务项目。
获取方式:点“在看”,关注公众号并回复 666 领取,更多内容陆续奉上。
文章有帮助的话,在看,转发吧。谢谢支持哟 (*^__^*)