SpringBoot+Nacos+Kafka简单实现微服务流编排

本文涉及的产品
注册配置 MSE Nacos/ZooKeeper,182元/月
MSE Nacos/ZooKeeper 企业版试用,1600元额度,限量50份
云原生网关 MSE Higress,422元/月
简介: SpringBoot+Nacos+Kafka简单实现微服务流编排
  • 前言
  • 准备工作
  • Nacos 安装及使用入门
  • 准备三个 SpringBoot 服务,引入 Nacos 及 Kafka
  • 业务解读
  • Nacos 配置
  • 总结

前言

最近一直在做微服务开发,涉及了一些数据处理模块的开发,每个处理业务都会开发独立的微服务,便于后面拓展和流编排。

学习了 SpringCloud Data Flow 等框架,感觉这个框架对于我们来说太重了,维护起来也比较麻烦,于是根据流编排的思想,基于我们目前的技术栈实现简单的流编排功能。

简单的说,我们希望自己的流编排就是微服务可插拔,微服务数据入口及输出可不停机修改。

基于 Spring Boot + MyBatis Plus + Vue & Element 实现的后台管理系统 + 用户小程序,支持 RBAC 动态权限、多租户、数据权限、工作流、三方登录、支付、短信、商城等功能。

项目地址:https://github.com/YunaiV/ruoyi-vue-pro

准备工作

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。

微信图片_20220907105127.png

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
> 基于微服务的思想,构建在 B2C 电商场景下的项目实战。核心技术栈,是 Spring Boot + Dubbo 。未来,会重构成 Spring Cloud Alibaba 。
>
> 项目地址:<https://github.com/YunaiV/onemall>
# 部署的nacos服务
nacos:
  config:
    server-addr: nacos-server:8848

建议配置本机 host 就可以填写 xxx-server 不用填写服务 ip。

业务解读

我们现在需要对三个服务进行编排,保障每个服务可以插拔,也可以调整服务的位置。

微信图片_20220907105532.png

示意图如上:

  • 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 配置项,配置如下:

微信图片_20220907105609.png

完成其中一个服务的配置,其它服务参考下图配置即可:

微信图片_20220907105631.png

②读取配置

代码如下:

@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 实现自己的动态调整方案,可以帮助自己更好的理解流编码思想及原理。

在实际业务中,还有许多业务问题需要去突破,我们这样处理更多是因为服务可插拔,便于流处理微服务在项目灵活搭配。

因为我现在工作是在传统公司,由于一些原因很难去推动新框架的使用,经常会用一些现有技术栈组合搞一些 操作,供大家参考,希望大家多多指教。


相关文章
|
6月前
|
安全 Java Apache
微服务——SpringBoot使用归纳——Spring Boot中集成 Shiro——Shiro 身份和权限认证
本文介绍了 Apache Shiro 的身份认证与权限认证机制。在身份认证部分,分析了 Shiro 的认证流程,包括应用程序调用 `Subject.login(token)` 方法、SecurityManager 接管认证以及通过 Realm 进行具体的安全验证。权限认证部分阐述了权限(permission)、角色(role)和用户(user)三者的关系,其中用户可拥有多个角色,角色则对应不同的权限组合,例如普通用户仅能查看或添加信息,而管理员可执行所有操作。
258 0
|
6月前
|
安全 Java 数据安全/隐私保护
微服务——SpringBoot使用归纳——Spring Boot中集成 Shiro——Shiro 三大核心组件
本课程介绍如何在Spring Boot中集成Shiro框架,主要讲解Shiro的认证与授权功能。Shiro是一个简单易用的Java安全框架,用于认证、授权、加密和会话管理等。其核心组件包括Subject(认证主体)、SecurityManager(安全管理员)和Realm(域)。Subject负责身份认证,包含Principals(身份)和Credentials(凭证);SecurityManager是架构核心,协调内部组件运作;Realm则是连接Shiro与应用数据的桥梁,用于访问用户账户及权限信息。通过学习,您将掌握Shiro的基本原理及其在项目中的应用。
213 0
|
6月前
|
NoSQL Java 关系型数据库
微服务——SpringBoot使用归纳——Spring Boot 中集成Redis——Redis 介绍
本文介绍在 Spring Boot 中集成 Redis 的方法。Redis 是一种支持多种数据结构的非关系型数据库(NoSQL),具备高并发、高性能和灵活扩展的特点,适用于缓存、实时数据分析等场景。其数据以键值对形式存储,支持字符串、哈希、列表、集合等类型。通过将 Redis 与 Mysql 集群结合使用,可实现数据同步,提升系统稳定性。例如,在网站架构中优先从 Redis 获取数据,故障时回退至 Mysql,确保服务不中断。
232 0
微服务——SpringBoot使用归纳——Spring Boot 中集成Redis——Redis 介绍
|
6月前
|
Java 数据安全/隐私保护 微服务
微服务——SpringBoot使用归纳——Spring Boot中使用监听器——Spring Boot中自定义事件监听
本文介绍了在Spring Boot中实现自定义事件监听的完整流程。首先通过继承`ApplicationEvent`创建自定义事件,例如包含用户数据的`MyEvent`。接着,实现`ApplicationListener`接口构建监听器,用于捕获并处理事件。最后,在服务层通过`ApplicationContext`发布事件,触发监听器执行相应逻辑。文章结合微服务场景,展示了如何在微服务A处理完逻辑后通知微服务B,具有很强的实战意义。
314 0
|
6月前
|
缓存 Java 数据库
微服务——SpringBoot使用归纳——Spring Boot中使用监听器——监听器介绍和使用
本文介绍了在Spring Boot中使用监听器的方法。首先讲解了Web监听器的概念,即通过监听特定事件(如ServletContext、HttpSession和ServletRequest的创建与销毁)实现监控和处理逻辑。接着详细说明了三种实际应用场景:1) 监听Servlet上下文对象以初始化缓存数据;2) 监听HTTP会话Session对象统计在线用户数;3) 监听客户端请求的Servlet Request对象获取访问信息。每种场景均配有代码示例,帮助开发者理解并应用监听器功能。
329 0
|
6月前
|
Java 关系型数据库 数据库
微服务——SpringBoot使用归纳——Spring Boot事务配置管理——常见问题总结
本文总结了Spring Boot中使用事务的常见问题,虽然通过`@Transactional`注解可以轻松实现事务管理,但在实际项目中仍有许多潜在坑点。文章详细分析了三个典型问题:1) 异常未被捕获导致事务未回滚,需明确指定`rollbackFor`属性;2) 异常被try-catch“吃掉”,应避免在事务方法中直接处理异常;3) 事务范围与锁范围不一致引发并发问题,建议调整锁策略以覆盖事务范围。这些问题看似简单,但一旦发生,排查难度较大,因此开发时需格外留意。最后,文章提供了课程源代码下载地址,供读者实践参考。
116 0
|
6月前
|
Java 关系型数据库 数据库
微服务——SpringBoot使用归纳——Spring Boot事务配置管理——Spring Boot 事务配置
本文介绍了 Spring Boot 中的事务配置与使用方法。首先需要导入 MySQL 依赖,Spring Boot 会自动注入 `DataSourceTransactionManager`,无需额外配置即可通过 `@Transactional` 注解实现事务管理。接着通过创建一个用户插入功能的示例,展示了如何在 Service 层手动抛出异常以测试事务回滚机制。测试结果表明,数据库中未新增记录,证明事务已成功回滚。此过程简单高效,适合日常开发需求。
732 0
|
6月前
|
Java 数据库 微服务
微服务——SpringBoot使用归纳——Spring Boot事务配置管理——事务相关
本文介绍Spring Boot事务配置管理,阐述事务在企业应用开发中的重要性。事务确保数据操作可靠,任一异常均可回滚至初始状态,如转账、购票等场景需全流程执行成功才算完成。同时,事务管理在Spring Boot的service层广泛应用,但根据实际需求也可能存在无需事务的情况,例如独立数据插入操作。
109 0
|
4月前
|
人工智能 数据可视化 JavaScript
颠覆开发效率!国内首个微服务编排框架Juggle开源啦!
Juggle是国内首个开源的微服务编排框架,专注于解决企业微服务进程中接口重复开发、系统对接复杂等问题。它提供零代码、低代码和AI增强功能,通过可视化拖拽快速组装简单API为复杂接口,支持多协议、多语言脚本和流程多版本管理。相比国外框架如Conductor,Juggle更贴合国内需求,具备高效开发、企业级可靠性及信创适配等优势,助力企业实现敏捷创新与数字化转型。
颠覆开发效率!国内首个微服务编排框架Juggle开源啦!