SpringCloud Stream消息总线
一、消息驱动概述
官网: https://spring.io/projects/spring-cloud-stream#overview
文档: https://cloud.spring.io/spring-cloud-static/spring-cloud-stream/3.0.1.RELEASE/reference/html/
SpringCloud Stream中文知道手册:https://m.wang1314.com/doc/webapp/topic/20971999.html
1、消息驱动是什么
在实际应用中有很多消息中间件,比如现在企业里常用的有ActiveMQ、RabbitMQ、RocketMQ、Kafka等,学习所有这些消息中间件无疑需要大量时间经历成本,那有没有一种技术,使我们不再需要关注具体的消息中间件的细节,而只需要用一种适配绑定的方式,自动的在各种消息中间件内切换呢?消息驱动就是这样的技术,它能屏蔽底层消息中间件的差异,降低切换成本,统一消息的编程模型。
什么是SpringCloud Stream
官方定义 SpringCloud Stream是一个构件消息驱动微服务的框架。
通过 inputs 和 outputs 来与 Spring Cloud Stream中的binder(绑定器)对象交互。
通过配置来binding(绑定),而 Spring Cloud Stream 的binder对象负责与消息中间件交互。
所以,我们只需要搞清楚如何与 Spring Cloud Stream交互就可以方便使用消息驱动的方式。
通话Spring Integration来连接消息代理中间件以实现消息事件驱动。
Spring Cloud Stream 为一些供应商的消息中间件产品通过了个性化的自动化配置,引用了发布-订阅、消费组、分区的三个核心概念。
但是截至到目前时间,SpringCloud Stream目前仅支持RabbitMQ和Kafka。
2、设计思想
标准MQ
- 生产者/消费者之间靠消息媒介传递信息内容(Message)
- 消息必须走特定的通道(消息通道MessageChannel)
- 消息通道里的消息如何被消费呢,谁负责收发处理(消息通道MessageChannel的子接口SubscribableChannel,由MessageHandler消息处理器所订阅)
为什么用Cloud Stream
比方说我们使用到了RibbitMQ和Kafka,由于这两个消息中间件的架构上的不同,
像RibbitMQ有exchange,kafka有Topic和Paritions分区,
在没有绑定器这个概念的情况下,我们的SpringBoot应用要直接与中间件进行信息交互的时候,
由于个消息中间件构建的初衷不同,它们的实现细节上会有较大的差异性
通过定义绑定器作为中间层,完美实现了应用程序与消息中间件细节之间的隔离。
通过向应用程序暴露统一的Channel通道,使得应用程序不需要再考虑各种不同的消息中间件实现。
通过定义绑定器作为中间层,实现了应用程序与消息中间件细节之间隔离。
- input对应于消费者
- output对应于生产者
可以做到代码层面对中间件的无感知,甚至于动态的切换中间件(rabbitmq切换为kafka),使得微服务开发的高速解耦,服务可以关注更多自己的业务流程
通过定义绑定器作为中间层,实现了应用程序与消息中间件细节之间的隔离。
Stream中的消息通信方式遵循了发布-订阅模式
Topic主题进行广播:
在RabbitMQ就是Exchange
在Kafka中就是Topic
3、Spring Cloud Stream标准流程
- Binder (连接中间件,屏蔽差异)
- Channel(通道,是队列Queue的一种抽象,在消息通讯系统中就是实现存储和转发的媒介,通过Channel对队列进行配置)
- Source和Sink(简单的可理解为参照对象是Spring Cloud Stream自身,从Stream发布消息就是输出,接受消息就是输入)
4、编码API和常用注解
组成 | 说明 |
---|---|
Middleware | 中间件,目前只支持RabbitMQ和Kaf |
Binder | Binder是应用与消息中间件之间的封装,目前实行了RabbitMQ和Kafka的Binder,通过Binder可以很方便的连接中间件,可以动态的改变消息类型(对应于Kafka的topic,RabbitMQ的exchange),这些都可以通过配置文件来实现 |
@Input | 注解标识输入通道,通过该输入通道接收到的消息进入应用程序 |
@Output | 注解标识输出通道,发布的消息将通过该通道离开应用程序 |
@StreamListner | 监听队列,用于消费者的队列的消息接收 |
@EnableBinding | 使信道Channel和交换机/主题(Exchange/Topic)绑定在一起 |
二、案列说明
- 本地安装RabbitMQ环境OK
- 工程中新建三个子模块
模块名称 | 模块功能 |
---|---|
stream-rabbitmq-provider-8801 | 生产者,发送消息模块 |
stream-rabbitmq-consumer-8802 | 消费者,接受消息模块 |
stream-rabbitmq-consumer-8803 | 消费者,接受消息模块 |
三、消息驱动之生产者
新建Module:stream-rabbitmq-provider-8801作为消息的生产者用来发送消息
pom文件:
<dependencies>
<!--stream-rabbit-->
<dependency>
<groupId>org.springframework.cloud</groupId>
<artifactId>spring-cloud-starter-stream-rabbit</artifactId>
</dependency>
<!-- eureka-client -->
<dependency>
<groupId>org.springframework.cloud</groupId>
<artifactId>spring-cloud-starter-netflix-eureka-client</artifactId>
</dependency>
<!-- 引入自定义的API通用包 实体类 -->
<dependency>
<groupId>com.zzp.springcloud</groupId>
<artifactId>cloud-api-commons</artifactId>
<version>${project.version}</version>
</dependency>
<!--web actuator-->
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-web</artifactId>
</dependency>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-actuator</artifactId>
</dependency>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-devtools</artifactId>
<scope>runtime</scope>
<optional>true</optional>
</dependency>
<dependency>
<groupId>org.projectlombok</groupId>
<artifactId>lombok</artifactId>
<optional>true</optional>
</dependency>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-test</artifactId>
<scope>test</scope>
</dependency>
</dependencies>
application.yml配置文件:
server:
port: 8801
spring:
application:
name: stream-provider
cloud:
stream:
binders: # 在此处配置要绑定的rabbitmq的服务信息;
defaultRabbit: # 表示定义的名称,用于于binding整合
type: rabbit # 消息组件类型
environment: # 设置rabbitmq的相关的环境配置
spring:
rabbitmq:
host: localhost
port: 5672
username: guest
password: guest
bindings: # 服务的整合处理
output: # 这个名字是一个通道的名称
destination: zzpStudyExchange # 表示要使用的Exchange名称定义
content-type: application/json # 设置消息类型,本次为json,文本则设置“text/plain”
binder: defaultRabbit # 设置要绑定的消息服务的具体设置
eureka:
client: # 客户端进行Eureka注册的配置
service-url:
defaultZone: http://eureka7001.com:7001/eureka
instance:
lease-renewal-interval-in-seconds: 2 # 设置心跳的时间间隔(默认是30秒)
lease-expiration-duration-in-seconds: 5 # 如果现在超过了5秒的间隔(默认是90秒)
instance-id: send-8801.com # 在信息列表时显示主机名称
prefer-ip-address: true # 访问的路径变为IP地址
启动类:
@SpringBootApplication
public class StreamMQProviderMain8801{
public static void main(String[] args) {
SpringApplication.run(StreamMQProviderMain8801.class, args);
}
}
业务类:
package com.zzp.springcloud.service;
/**
* @author zzp
* @create
*/
public interface IMessageProvider {
//发送消息
String send();
}
这里不需要添加@Service注解,这个业务是需要跟MQ交互的,不再是传统的业务。添加@EnableBinding注解 用来绑定消息的发送管道,消息生产者绑定的消息推送管道为org.springframework.cloud.stream.messaging.Source:(源)
实现类:
package com.zzp.springcloud.service.impl;
import com.zzp.springcloud.service.IMessageProvider;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.cloud.stream.annotation.EnableBinding;
import org.springframework.cloud.stream.messaging.Source;
import org.springframework.messaging.MessageChannel;
import org.springframework.messaging.support.MessageBuilder;
import java.util.UUID;
@Slf4j
@EnableBinding(Source.class) //定义消息的推送管道
public class IMessageProviderImpl implements IMessageProvider {
@Autowired
private MessageChannel output; // 消息发送管道
@Override
public String send() {
String serial = UUID.randomUUID().toString().replace("-","");
output.send(MessageBuilder.withPayload(serial).build());
log.info(">>>>>>>>>>>>serial: " + serial);
return serial;
}
}
MessageBuilder.withPayload(serial).build()消息构建器–参考文档
控制类:
@RestController
public class SendMessageController {
@Autowired
private IMessageProvider messageProvider;
@GetMapping("/sendMessage")
public String sendMessage() {
return messageProvider.send();
}
}
启动7001,8801测试:http://localhost:8801/sendMessage
查看日记:
查看RabbitMQ控制面板中的交换机上多出了一个名为zzpStudyExchange的交换机,这就是我们yml配置的定义的Exchange名称
在RabbitMQ的控制面板中我们也看到了确实发送了消息
四、消息驱动之消费者
新建Module:stream-rabbitmq-consumer-8802/8803作为消息的生产者用来接收消息
pom文件:
<dependencies>
<!--stream-rabbit-->
<dependency>
<groupId>org.springframework.cloud</groupId>
<artifactId>spring-cloud-starter-stream-rabbit</artifactId>
</dependency>
<!-- eureka-client -->
<dependency>
<groupId>org.springframework.cloud</groupId>
<artifactId>spring-cloud-starter-netflix-eureka-client</artifactId>
</dependency>
<!-- 引入自定义的API通用包 实体类 -->
<dependency>
<groupId>com.zzp.springcloud</groupId>
<artifactId>cloud-api-commons</artifactId>
<version>${project.version}</version>
</dependency>
<!--web actuator-->
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-web</artifactId>
</dependency>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-actuator</artifactId>
</dependency>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-devtools</artifactId>
<scope>runtime</scope>
<optional>true</optional>
</dependency>
<dependency>
<groupId>org.projectlombok</groupId>
<artifactId>lombok</artifactId>
<optional>true</optional>
</dependency>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-test</artifactId>
<scope>test</scope>
</dependency>
</dependencies>
yml配置文件:
server:
port: 8802
spring:
application:
name: stream-consumer
cloud:
stream:
binders: # 在此处配置要绑定的rabbitmq的服务信息;
defaultRabbit: # 表示定义的名称,用于于binding整合
type: rabbit # 消息组件类型
environment: # 设置rabbitmq的相关的环境配置
spring:
rabbitmq:
host: localhost
port: 5672
username: guest
password: guest
bindings: # 服务的整合处理
input: # 这个名字是一个通道的名称 输入
destination: zzpStudyExchange # 表示要使用的Exchange名称定义
content-type: application/json # 设置消息类型,本次为json,文本则设置“text/plain”
binder: defaultRabbit # 设置要绑定的消息服务的具体设置
eureka:
client: # 客户端进行Eureka注册的配置
service-url:
defaultZone: http://eureka7001.com:7001/eureka
instance:
lease-renewal-interval-in-seconds: 2 # 设置心跳的时间间隔(默认是30秒)
lease-expiration-duration-in-seconds: 5 # 如果现在超过了5秒的间隔(默认是90秒)
instance-id: receiver-8802.com # 在信息列表时显示主机名称
prefer-ip-address: true # 访问的路径变为IP地址
其中:input表示消息的消费者
8801的yml文件中:output表示消息的生产者
bindings: # 服务的整合处理
output: # 这个名字是一个通道的名称
启动类:
@SpringBootApplication
public class StreamMQConsumer8802 {
public static void main(String[] args) {
SpringApplication.run(StreamMQConsumer8802.class, args);
}
}
业务类:
编写消息消费者的业务类,消费者也是消费mq中的消息,添加@EnableBinding注解用来绑定消息的推送管道,消息消费者绑定的消息推送管道为import org.springframework.cloud.stream.messaging.Sink,在接收消息的方法中需要使用注解@StreamListner来监听其绑定的消息推送管道:
package com.zzp.springcloud.listener;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.cloud.stream.annotation.EnableBinding;
import org.springframework.cloud.stream.annotation.StreamListener;
import org.springframework.cloud.stream.messaging.Sink;
import org.springframework.messaging.Message;
import org.springframework.stereotype.Component;
/**
* @author zzp
* @create
*/
@Component
@EnableBinding(Sink.class) // 定义消息的接受
@Slf4j
public class ReceiverMessageListener {
@Value("${server.port}")
private String serverPort;
@StreamListener(Sink.INPUT)
public void input(Message<String> message) {
log.info("serverPort: "+serverPort+ " 消费者1号,接受:"+message.getPayload());
}
}
启动8802/8803,然后在8801生成者发送消息,查看8802/8803消费者微服务的日记:
查看RabbitMQ的控制面板中exchanges
出现波峰,说明这个交互机有消费
交换机有两台机器(默认分组)
五、分组消费与持久化
1、重复消费问题
当生产者发送消息后,此时的我们的消费者都接受了消息并进行了消费,也就是说同一条消息被多个消息消费者所消费
生产者推送三条消息:
两台消费者都收到三条消息:
上述的问题就是消息的重复消费问题,那么这个问题为什么如此重要呢?其实重复消费这个问题本身不可怕,可怕的是没考虑到重复消费之后,怎么保证幂等性。(幂等性,通俗的说,就一个数据,或者一个请求,重复很多次,需要确保对应的数据是不会改变的,不能出错)。分布式微服务应用为了实现高可用和负载均衡,实际上同一功能的服务都会部署多个具体的服务实例。举个例子,假设有一个系统,有一条消息要求往数据库里插入一条数据,要是这个消息重复消费两次,结果就是向数据库里插入了两条数据,这样数据就错了,就违背了幂等性原则,但是要是该消息消费到第二次的时候,可以判断一下该消息已经消费过了,然后直接将该消息丢弃,这就实现了只插入一条数据,一条消息重复出现了两次,但是只有第一次真正被消费了,数据库里也就只插入了一条数据,这就保证了系统的幂等性。
这时我们可以使用Stream中的消息分组来解决
注意:在Stream总处于同一个group(组)中的多个消费者是竞争关系,就能够保证消息只会被其中一个应用消费一次。
不同组是可以全面消费的(重复消费),
同一组内会发生竞争关系,只有其中一个可以消费。
在RabbitMQ中,查看zzpStudyExchange交换机,可以发现8802和8803两个消息消费者处于不同的组,所以8801消息生产者发送的消息可以被这两个消费者重复消费:
2、分组解决重复消费问题
面在RabbitMQ控制面板中我们看到的组流水号是系统随机分配的,这样无疑不好控制,所以我们应该自定义配置分组,将8802/8803两个消息消费者微服务分为同一个组,以此来解决消息的重复消费问题,先来演示如何自定义分组。
8802/8803消费者微服务的yml配置文件中分别添加组名属性(不同的组名):
group: zzpA/B # 分组名称
先在RabbitMQ中控制面板中,查看zzpStudyExchange交换机;
此时由于8802/8803位于两个不同分组下,所以没有竞争关系,消息生产者发送消息后,仍然可以重复消费。
下面我们将这两个消息消费方微服务分到相同的消费组中,这样每次就只有一个消费者,消息生产者发送的消息只能被8802或8803其中一个接受到,这样就避免了重复消费,将8802和8803的分组名都改为zzpA,再次重启两个消息消费方微服务,此时我们可以看到在分组A下已经有了两个消费者
生产者推送四条消息:
消费者服务接收消息:我们发现8802/8803分别消费了各2条不同的消息,而没有出现重复消费的问题:
3、持久化
我们把8803消费微服务的yml文件配置分组名称去掉
停掉8802和8803消费者微服务:
在8801生产者服务推送四条消息:
启动8803消费者服务:
8803启动后,日记没有打印消费日记,说明8803并没有去mq上拿未曾消费的消息
启动8802消费者服务:
8802服务消费了从8801生产者服务推送四条消息
Stream的group分组属性,在我们的消息重复消费和消息持久化消费、避免消息丢失,是一个非常重要的属性。
原文地址:https://blog.csdn.net/weixin_42469070/article/details/114375274
版权声明:本文内容由互联网用户自发贡献,该文观点与技术仅代表作者本人。本站仅提供信息存储空间服务,不拥有所有权,不承担相关法律责任。如发现本站有涉嫌侵权/违法违规的内容, 请发送邮件至 dio@foxmail.com 举报,一经查实,本站将立刻删除。