Spring Cloud Stream整合RabbitMQ
前言
在使用消息队列组件时，如果程序需要更换消息队列组件，那么代码内部可能需要做巨大的修改，而Spring Cloud Stream则解决了这个问题，应用程序只需要通过 inputs 或者 outputs 来与 Spring Cloud Stream 中binder 交互，通过我们配置来 binding ，而 Spring Cloud Stream 的 binder 负责与消息中间件交互。所以，我们只需要搞清楚如何与 Spring Cloud Stream 交互就可以方便使用消息驱动的方式。
生产者
依赖
1
2
3
4
| <dependency>
<groupId>org.springframework.cloud</groupId>
<artifactId>spring-cloud-starter-stream-rabbit</artifactId>
</dependency>
|
配置
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
| spring:
cloud:
stream:
bindings:
output:
destination: testStreamExchange
binder: local_rabbit
binders:
local_rabbit:
type: rabbit
environment:
spring:
rabbitmq:
host: 127.0.0.1
port: 5672
username: guest
password: guest
|
消息发送类
1
2
3
4
5
6
7
8
9
10
11
12
13
| @RestController
@EnableBinding(Processor.class)
public class MessageController {
@Autowired
private Processor processor;
@GetMapping("/sendMessage")
public void sendMessage(String message){
processor.output().send(MessageBuilder.withPayload(message).build());
}
}
|
注：这里的Processor继承了Source与Sink，可以参考Processor的结构定义自己的配置类，与配置文件中自定义的output(myoutput)对应
消费者
依赖
1
2
3
4
| <dependency>
<groupId>org.springframework.cloud</groupId>
<artifactId>spring-cloud-starter-stream-rabbit</artifactId>
</dependency>
|
配置
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
| spring:
cloud:
stream:
bindings:
input:
destination: testStreamExchange
binder: local_rabbit
group: testStreamQueue
binders:
local_rabbit:
type: rabbit
environment:
spring:
rabbitmq:
host: 127.0.0.1
port: 5672
username: guest
password: guest
rabbit:
bindings:
input:
consumer:
requeue-rejected: false
acknowledge-mode: MANUAL
recovery-interval: 3000
durable-subscription: true
max-concurrency: 5
|
监听类
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
| @Component
@EnableBinding(Processor.class)
public class Listener {
private Logger log = LoggerFactory.getLogger(Listener.class);
@StreamListener(Processor.INPUT)
public void process(Message message){
log.info("{}: message is {}", this.getClass().getSimpleName(), message);
MessageHeaders headers = message.getHeaders();
Channel channel = (Channel) headers.get(AmqpHeaders.CHANNEL);
Long deliveryTag = (Long) message.getHeaders().get(AmqpHeaders.DELIVERY_TAG);
try {
assert channel != null;
channel.basicAck(deliveryTag, false);
} catch (IOException e) {
e.printStackTrace();
}
}
}
|
注：这里的Processor继承了Source与Sink，可以参考Processor的结构定义自己的配置类，与配置文件中自定义的input(myinput)对应\
关于Channel
如果每一次访问RabbitMQ都建立一个Connection，在消息量大的时候建立TCP Connection的开销将是巨大的，效率也较低。Channel是在connection内部建立的逻辑连接，如果应用程序支持多线程，通常每个thread创建单独的channel进行通讯，AMQP method包含了channel id帮助客户端和message broker识别channel，所以channel之间是完全隔离的。Channel作为轻量级的Connection极大减少了操作系统建立TCP connection的开销
我们在消费这边编写两个监听器监听同一个队列
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
| @Component
@EnableBinding(Processor.class)
public class Listener {
private Logger log = LoggerFactory.getLogger(Listener.class);
private Channel channel = null;
@StreamListener(Processor.INPUT)
public void process(Message message){
deal(message);
}
@StreamListener(Processor.INPUT)
public void processTwo(Message message){
deal(message);
}
private void deal(Message message) {
log.info("{}: message is {}", this.getClass().getSimpleName(), message);
MessageHeaders headers = message.getHeaders();
Channel channel = (Channel) headers.get(AmqpHeaders.CHANNEL);
if (this.channel == null){
this.channel = channel;
}else{
log.info("{}", channel == this.channel);
}
Long deliveryTag = (Long) message.getHeaders().get(AmqpHeaders.DELIVERY_TAG);
try {
assert channel != null;
channel.basicAck(deliveryTag, false);
} catch (IOException e) {
e.printStackTrace();
}
}
}
|
生产者发送消息（多次）
可以看到发送一次消息两个监听器同时被触发，与纯rabbitmq不同（纯rabbitmq是轮询），并且第一次判断为true，证明一个接收通道（input）只使用同一个channel，监听同一个接收通道与监听同一个队列是不同的概念。而且可以发送当两（多）个监听器监听同一个接收通道时，会导致线程死掉，重新生成channel。
