案例说明
- cloud-stream-rabbitmq-provider8801:作为生产者进行发消息模块
- cloud-stream-rabbitmq-provider8802:作为消费者接收消息模块
- cloud-stream-rabbitmq-provider8803:作为消费者接收消息模块
生产者
引入依赖:
<?xml version="1.0" encoding="UTF-8"?>
<project xmlns="http://maven.apache.org/POM/4.0.0"
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
<parent>
<artifactId>cloud2020</artifactId>
<groupId>pers.zhang.springcloud</groupId>
<version>1.0-SNAPSHOT</version>
</parent>
<modelVersion>4.0.0</modelVersion>
<artifactId>cloud-stream-rabbitmq-provider8801</artifactId>
<dependencies>
<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.cloud</groupId>
<artifactId>spring-cloud-starter-netflix-eureka-client</artifactId>
</dependency>
<dependency>
<groupId>org.springframework.cloud</groupId>
<artifactId>spring-cloud-starter-stream-rabbit</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>
</project>
配置:application.yml
server:
port: 8801
spring:
application:
name: cloud-stream-provider
cloud:
stream:
binders: #在此处配置要绑定的rabbitmq的服务信息
defaultRabbit: #表示定义的命名,用于binding整合
type: rabbit #消息组件类型
environment: #设置rabbitmq的相关环境配置
spring:
tabbitmq:
host: localhost
port: 5672
username: guest
password: guest
bindings: #服务的整合处理
output: #这个名字是一个通用的名称
destination: studyExchange #表示要使用的Exchange名称定义
content-type: application/json #设置消息类型,本次为json,文本则为 text/plain
binder: defaultRabbit #设置要绑定的消息服务的具体设置
eureka:
client:
service-url:
defaultZone: http://localhost:7001/eureka
instance:
lease-renewal-interval-in-seconds: 2 #心跳时间间隔(默认30秒)
lease-expiration-duration-in-seconds: 5 #超时间隔(默认90秒)
instance-id: send-8801.com #在信息列表时显示主机名称
prefer-ip-address: true #访问路径变为ip地址
消息接口及实现:
public interface IMessageProvider {
String send();
}
@EnableBinding(Source.class) //定义消息的推送管道
public class MessageProviderImpl implements IMessageProvider {
@Resource
MessageChannel output; //消息发送管道
@Override
public String send() {
String serial = UUID.randomUUID().toString();
output.send(MessageBuilder.withPayload(serial).build());
System.out.println("*****serial: " + serial);
return null;
}
}
Controller:
@RestController
public class SendMessageController {
@Resource
IMessageProvider messageProvider;
@GetMapping(value = "/sendMessage")
public String sendMessage() {
return messageProvider.send();
}
}
启动类:
@SpringBootApplication
public class StreamMQMain8801 {
public static void main(String[] args) {
SpringApplication.run(StreamMQMain8801.class, args);
}
}
测试:启动eureka,启动rabbitMQ,启动8801
访问:localhost:8801/sendMessage
后台打印:
消费者
引入依赖:
<?xml version="1.0" encoding="UTF-8"?>
<project xmlns="http://maven.apache.org/POM/4.0.0"
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
<parent>
<artifactId>cloud2020</artifactId>
<groupId>pers.zhang.springcloud</groupId>
<version>1.0-SNAPSHOT</version>
</parent>
<modelVersion>4.0.0</modelVersion>
<artifactId>cloud-stream-rabbitmq-consumer8802</artifactId>
<dependencies>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-web</artifactId>
</dependency>
<dependency>
<groupId>org.springframework.cloud</groupId>
<artifactId>spring-cloud-starter-netflix-eureka-client</artifactId>
</dependency>
<dependency>
<groupId>org.springframework.cloud</groupId>
<artifactId>spring-cloud-starter-stream-rabbit</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>
</project>
配置:application.yml
server:
port: 8802
spring:
application:
name: cloud-stream-consumer
cloud:
stream:
binders: #在此处配置要绑定的rabbitmq的服务信息
defaultRabbit: #表示定义的命名,用于binding整合
type: rabbit #消息组件类型
environment: #设置rabbitmq的相关环境配置
spring:
tabbitmq:
host: localhost
port: 5672
username: guest
password: guest
bindings: #服务的整合处理
input: #这个名字是一个通用的名称
destination: studyExchange #表示要使用的Exchange名称定义
content-type: application/json #设置消息类型,本次为json,文本则为 text/plain
binder: defaultRabbit #设置要绑定的消息服务的具体设置
eureka:
client:
service-url:
defaultZone: http://localhost:7001/eureka
instance:
lease-renewal-interval-in-seconds: 2 #心跳时间间隔(默认30秒)
lease-expiration-duration-in-seconds: 5 #超时间隔(默认90秒)
instance-id: receive-8802.com #在信息列表时显示主机名称
prefer-ip-address: true #访问路径变为ip地址
Listener:
@Component
@EnableBinding(Sink.class)
public class ReceiveMessageListenerController {
@Value("${server.port}")
String serverPort;
@StreamListener(Sink.INPUT)
public void input(Message<String> message) {
System.out.println("消费者1号,------>接收到消息:" + message.getPayload() + "\tport:" + serverPort);
}
}
启动类:
@SpringBootApplication
public class StreamMQMain8802 {
public static void main(String[] args) {
SpringApplication.run(StreamMQMain8802.class, args);
}
}
测试:启动eureka,启动RabbitMQ,启动8801生产者和8802消费者:
访问:localhost:8801/sendMessage
8802后台打印:
分组消费与持久化
依照8802,clone一个8803并启动:
访问:localhost:8801/sendMessage
此时,8801和8802都接收到了消息:
存在重复消费问题!
比如在如下场景中,订单系统我们做集群部署,都会从RabbitMQ中获取订单信息,那如果一个订单同时被两个服务获取到,那么就会造成数据错误,需要避免这种情况
这时就可以使用Stream中的消息分组来解决,在Stream中处于同一个group中的多个消费者是竞争关系,就能够保证消息只会被其中一个应用消费一次。不同组是可以全面消费的(重复消费)。同一个组内会发生竞争关系,只有其中一个可以消费。
原理:
微服务应用放置同一个group中,就能够保证消息只会被其中一个应用消费一次。不同的组是可以消费的,同一个组内会发生竞争关系,只有其中一个可以消费。
1. 使8802和8803变成不同的组:group: atguiguA、atguiguB
8802修改yml:
bindings: #服务的整合处理
input: #这个名字是一个通用的名称
destination: studyExchange #表示要使用的Exchange名称定义
content-type: application/json #设置消息类型,本次为json,文本则为 text/plain
binder: defaultRabbit #设置要绑定的消息服务的具体设置
group: atguiguA #自定义分组
8803修改yml:
bindings: #服务的整合处理
input: #这个名字是一个通用的名称
destination: studyExchange #表示要使用的Exchange名称定义
content-type: application/json #设置消息类型,本次为json,文本则为 text/plain
binder: defaultRabbit #设置要绑定的消息服务的具体设置
group: atguiguB #自定义分组
测试:
发送消息:
8802和8803均可以收到(不同组可以重复消费):
2. 使8802和8803实现轮询分组,每次只有一个消费者8801模块发送的消息只能被其中一个接收到这样就避免了重复消费:
修改8803的group 为 atguiguA, 变成相同的组。
8801发送2条消息:
8802接收到其中一条:
8803接收到另一条:
持久化
-
停止8802和8803微服务,并删除调8802的分组group:atguiguA
-
发送4条消息到RabbitMQ
-
先启动8802,无分组属性配置,后台没有打印消息
-
再启动8803,有分组属性配置,打印全部4条消息
结论:
配置了group属性的模块支持消息持久化