11.SpringCloud Stream (消息驱动)
1.消息驱动概述
是什么
屏蔽底层消息中间件的差异，降低切换版本，统一消息的编程模型官网https://spring.io/projects/spring-cloud-stream#overviewhttps://cloud.spring.io/spring-cloud-static/spring-cloud-stream/3.0.1.RELEASE/reference/html/ Spring Cloud Stream中文指导手册https://m.wang1314.com/doc/webapp/topic/20971999.html
设计思想
标准MQ
- 生产者/消费者之间靠消息媒介传递信息内容 Message
- 消息必须走特定的通道 消息通道MessageChannel
- 消息通道里的消息如何被消费呢，谁负责收发处理
为什么用Cloud Stream
stream凭什么可以统一底层差异
Binder
- INPUT对应于消费者
- OUTPUT对应于生产者
Stream中的消息通信方式遵循了发布-订阅模式
Topic主题进行广播- 在RabbitMQ就是Exchange
- 在kafka中就是Topic
Spring Cloud Stream标准流程套路
- Binder
- Channel
- Source和Sink
编码API和常用注解
2.案例说明
前提：RabbitMQ环境已经OK工程中新建三个子模块- cloud-stream-rabbitmq-provider8801,作为生产者进行发消息模块
- cloud-stream-rabbitmq-consumer8802,作为消息接收模块
- cloud-stream-rabbitmq-consumer8803,作为消息接收模块
3.消息驱动之生产者
新建模块 cloud-stream-rabbitmq-provider8801pom<?xml version="1.0" encoding="UTF-8"?>
springcloud2020
com.chl.springcloud
1.0-SNAPSHOT
4.0.0
cloud-stream-rabbitmq-provider8801
org.springframework.cloud
spring-cloud-starter-stream-rabbit
org.springframework.cloud
spring-cloud-starter-netflix-eureka-client
com.chl.springcloud
cloud-api-commons
1.0-SNAPSHOT
org.springframework.boot
spring-boot-starter-web
org.springframework.boot
spring-boot-starter-actuator
org.springframework.boot
spring-boot-devtools
runtime
true
org.projectlombok
lombok
true
org.springframework.boot
spring-boot-starter-test
test
server:
port: 8801
spring:
application:
name: cloud-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: studyExchange # 表示要使用的Exchange名称定义
content-type: application/json # 设置消息类型，本次为json，文本则设置“text/plain”
binder: defaultRabbit # 设置要绑定的消息服务的具体设置
eureka:
client: # 客户端进行Eureka注册的配置
service-url:
defaultZone: http://localhost: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 StreamMQMain8801 {
public static void main(String[] args) {
SpringApplication.run(StreamMQMain8801.class, args);
}
}public interface IMessageProvider {
public String send();
}@EnableBinding(Source.class) //定义消息的推送管道
public class MessageProviderImpl implements IMessageProvider {
@Autowired
private MessageChannel output; // 消息发送管道
@Override
public String send() {
String serial = UUID.randomUUID().toString();
//发消息出去 将消息用builder一下，发出去
output.send(MessageBuilder.withPayload(serial).build());
System.out.println("*****serial: "+serial);
return null;
}
}@RestController
public class SendMessageController {
@Autowired
private IMessageProvider messageProvider;
@GetMapping(value = "/sendMessage")
public String sendMessage() {
return messageProvider.send();
}
}4.消息驱动之消费者
新建 cloud-stream-rabbitmq-consumer8802pom<?xml version="1.0" encoding="UTF-8"?>
springcloud2020
com.chl.springcloud
1.0-SNAPSHOT
4.0.0
cloud-stream-rabbitmq-consumer8802
org.springframework.cloud
spring-cloud-starter-stream-rabbit
org.springframework.cloud
spring-cloud-starter-netflix-eureka-client
com.chl.springcloud
cloud-api-commons
1.0-SNAPSHOT
org.springframework.boot
spring-boot-starter-web
org.springframework.boot
spring-boot-starter-actuator
org.springframework.boot
spring-boot-devtools
runtime
true
org.projectlombok
lombok
true
org.springframework.boot
spring-boot-starter-test
test
server:
port: 8802
spring:
application:
name: cloud-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: studyExchange # 表示要使用的Exchange名称定义
content-type: application/json # 设置消息类型，本次为json，文本则设置“text/plain”
binder: defaultRabbit # 设置要绑定的消息服务的具体设置
group: chlGroup1 #所属分组
eureka:
client: # 客户端进行Eureka注册的配置
service-url:
defaultZone: http://localhost:7001/eureka
instance:
lease-renewal-interval-in-seconds: 2 # 设置心跳的时间间隔（默认是30秒）
lease-expiration-duration-in-seconds: 5 # 如果现在超过了5秒的间隔（默认是90秒）
instance-id: receive-8802.com # 在信息列表时显示主机名称
prefer-ip-address: true # 访问的路径变为IP地址@SpringBootApplication
public class StreamMQMain8802 {
public static void main(String[] args) {
SpringApplication.run(StreamMQMain8802.class, args);
}
}@Component
@EnableBinding(Sink.class) //开启监听 input
public class ReceiveMessageListenerController {
@Value("${server.port}")
private String serverPort;
@StreamListener(Sink.INPUT)//监听
public void input(Message message) {
System.out.println("消费者2号，接受："+message.getPayload()+"\t port:"+serverPort);
}
} 5.分组消费与持久化
重复消费演示
依照8802，clone出来一份运行8803新建立 cloud-stream-rabbitmq-consumer8803启动- RabbitMQ
- 服务注册 7001
- 消息生产 8801
- 消息消费 8802 ，8803
运行后两个问题
- 有重复消费问题
- 消息持久化问题
目前是8802/8803同时都收到了，存在重复消费问题http://localhost:8801/sendMessage 如何解决分组和持久化属性group
分组
原理：微服务应用放置于同一个group中，就能够保证消息只会被其中一个应用消费一次。不同的组是可以消费的，同一个组内会发生竞争关系，只有其中一个可以消费。分为同组测试
8802/8803都变成不同组，group两个不同group: chlGroup1、chlGroup28802 yml配置8803 yml配置
配置结果测试结果 还是重复消费
分为不同组测试
8802/8803都变成相同组，group两个相同- group: chlGroup1
6.持久化
通过上述，解决了重复消费问题，再看看持久化停止8802/8803 并去除掉8802的分组 8803的保留分组- group: chlGroup1