一、背景
以前我们在spring boot中构建一个消息驱动的微服务应用，通常会使用rabbitMQ或是kafka来做消息中间件，应用中均需代码实现具体消息中间件的通信细节。此时如果再更换一个新的消息中间件，这会我们又需新增这些通信代码，写起来会比较繁琐，而stream出现就是为了简化这一过程。
二、简介
它是一个构建消息驱动的微服务应用的框架。通过一些抽象出来的基础概念，来简化消息中间件的使用。我们可以看下官网上的处理模型图：
- 关键概念
Inputs 接收消息的通道
Output 发送消息的通道
Binder 可理解为一个抽象的中间件，应用通过在spring cloud stream中所注入的inputs,outputs通道来跟外界消息通信，而这些通道又是通过具体中间件的Binder实现来连接到消息队列的服务器上。有了Binder，甚至可以不改一行代码，就切换中间件的类型。目前Binder实现支持的具体中间件类型为：rabbitMQ 和 kfaka这俩
当然MQ中的消费组group 和 分区 partion的概念他也有，跟kafka里面的概念是一样的。
Group：消费组，一个消息到达一个消费组后，只能被这个消费组的其中一个实例消费掉；
Partion：消息分区，一个非常大的topic可以分布到多个broker（即服务器）上
三、使用步骤
以下以rabbitMQ为具体中间件作为示例：
1. 在pom.xml中添加依赖
<dependency>
<groupid >org.springframework.cloud</groupid>
<artifactid>spring-cloud-starter-stream-rabbit</artifactid>
</dependency>
2.自定义通道的创建
自定义通道的创建有两种方式，一种是提前在代码里定义好的，另一种是在运行时通过读取完通道名来创建的。
2.1 方式一：提前定义好的通道
- 定义生产者
我们定义一个生产者类SampleSource，这类要完成2件事： - 2.1.1 自定义发送通道
- 2.1.2 完成发送消息的功能
@EnableBinding(SampleSource.MultiOutputSource.class)
public class SampleSource {
//自定义发送通道
public interface MultiOutputSource {
String OUTPUT1 = "output1";
String OUTPUT2 = "output2";
@Output(OUTPUT1)
MessageChannel output1();
@Output(OUTPUT2)
MessageChannel output2();
}
}
注意：要加上@EnableBinding 绑定通道才能够发出消息到mq的服务器。
-
方式二：动态创建通道
@EnableBinding
@Controller
public class SourceWithDynamicDestination {
@Autowired
private BinderAwareChannelResolver resolver;
@RequestMapping(path = "/{target}", method = POST, consumes = "*/*")
@ResponseStatus(HttpStatus.ACCEPTED)
public void handleRequest(@RequestBody String body, @PathVariable("target") target,
@RequestHeader(HttpHeaders.CONTENT_TYPE) Object contentType) {
sendMessage(body, target, contentType);
}
private void sendMessage(String body, String target, Object contentType) {
resolver.resolveDestination(target).send(MessageBuilder.createMessage(body,
new MessageHeaders(Collections.singletonMap(MessageHeaders.CONTENT_TYPE, contentType))));
}
}
这是官网的示例代码，可以看到关键代码是这两句
@Autowired
private BinderAwareChannelResolver resolver;
//中间省略代码...
resolver.resolveDestination(target).send(MessageBuilder.createMessage(body,
new MessageHeaders(Collections.singletonMap(MessageHeaders.CONTENT_TYPE, contentType))));
resolveDestination里面的处理就是，先查找传入的target通道名，看下有没创建过，如果没有就会默认创建一个。
- 生产者实现发送消息的函数
在SampleSource这类里添加sendMessage函数
public class SampleSource{
@Bean
@InboundChannelAdapter(value = MultiOutputSource.OUTPUT1, poller = @Poller(fixedDelay = "1000", maxMessagesPerPoll = "1"))
public synchronized MessageSource<String> messageSource1() {
return new MessageSource<String>() {
public Message<String> receive() {
String message = "FromSource1";
System.out.println("******************");
System.out.println("From Source1");
System.out.println("******************");
System.out.println("Sending value: " + message);
return new GenericMessage(message);
}
};
}
@Bean
@InboundChannelAdapter(value = MultiOutputSource.OUTPUT2, poller = @Poller(fixedDelay = "1000", maxMessagesPerPoll = "1"))
public synchronized MessageSource<String> timerMessageSource() {
return new MessageSource<String>() {
public Message<String> receive() {
String message = "FromSource2";
System.out.println("******************");
System.out.println("From Source2");
System.out.println("******************");
System.out.println("Sending value: " + message);
return new GenericMessage(message);
}
};
}
}
对应的配置application.yml：
spring:
cloud.stream.bindings:
output1:
contentType: application/json #约定消息的内容编码格式
output2:
contentType: application/json #约定消息的内容编码格式
rabbitmq:
host: 127.0.0.1
port: 5672
username: sa
password: 123456
3. 消费者类的实现
消费者类要做的事也是相似的：
3.1.自定义接收通道
3.2消费消息的功能实现
@EnableBinding(SampleSink.MultiInputSink.class)
public class SampleSink {
@StreamListener(MultiInputSink.INPUT1)
public synchronized void receive1(String message) {
System.out.println("******************");
System.out.println("At Sink1");
System.out.println("******************");
System.out.println("Received message " + message);
}
@StreamListener(MultiInputSink.INPUT2)
public synchronized void receive2(String message) {
System.out.println("******************");
System.out.println("At Sink2");
System.out.println("******************");
System.out.println("Received message " + message);
}
public interface MultiInputSink {
String INPUT1 = "input1";
String INPUT2 = "input2";
@Input(INPUT1)
SubscribableChannel input1();
@Input(INPUT2)
SubscribableChannel input2();
}
}
对应的配置 application.yml (当然用application.propertities)
spring:
cloud.stream:
bindings:
input1:
group: inputGroup #加上group是为了持久化
input2:
group: inputGroup2
rabbit:
host: 127.0.0.1
port: 5672
username: sa
password: 123456
4.死信队列设置
问题列表：
消息在什么条件下进入死信队列？发送失败后，如何设置重试次数 or TTL？
进入死信队列之后，若又需要该消息重新回到原队列进行处理，该怎么办
转入死信队列的条件：
- 消息被拒绝（basic.reject/ basic.nack）并且requeue=false
- 消息TTL过期（参考：RabbitMQ之TTL（Time-To-Live 过期时间））
- 队列达到最大长度
配置死信队列及消息消费失败重试次数(application.yml)：
-
配置消息消费重试次数
两种方式
1） 如果允许重试一定次数：如上图配置所示，设置max_attempt ，大于1即可
2）如果不允许重发，消费失败了就进入死信队列，在配置中添加requeueRejected设为true
spring:
cloud.stream:
bindings:
input1:
group: inputGroup1
rabbit:
bindings:
input1:
consumer:
autoBindDlq: true #启用死信队列，默认会生成一个DLX EXCHANGE，当消息重复消费失败后
dlqDeadLetterExchange: input-deadLetter.DLX #如果该列声明，那么deadLetterExchange也要声明，这个保持一致
deadLetterExchange: input-deadLetter.DLX #与dlqDeadLetterExchange保持一致
requeueRejected: true
host: 127.0.0.1
port: 5672
username: sa
password: 123456
- 在代码中将某消息转入死信队列，另可见官网示例
5、消息的格式：
（1）消息头，包含了以下字段：
翻看源码，MessageHeader是个Map结构，左边是字段名，右边是字段内容。可以在创建MessageHeader的时候传入已经初始化的Map，注意我们可以在这指定body的contentType。contentType能填什么内容，查找下表即可（官网上找到的）：
（2）至于消息体(payLoad)，它支持自定义结构，格式自定。
6、如何保证消息的可靠性？
一般是通过具体的消息中间件来保证。
配置组就可以来保证消息可靠性。见官网描述：消息的持久性
设置持久性的属性：durableSubscription
