1. 概述
老话说的好：出错不怕，怕的是出了错，却不去改正。如果屡次出错，无法改对，就先记下了，然后找援军解决。
言归正传，今天来聊一下 Stream 组件的 出错重试 和 死信队列。
RabbitMQ 镜像模式集群的搭建，可参见我的另一篇文章《RabbitMQ 3.9.7 镜像模式集群的搭建》（https://www.jianshu.com/p/187325419e95）
在早期的 SpringCloud 版本中常使用 @Input、@Output、@EnableBinding 和 @StreamListener 注解开发生产者与消费者。
官方原文：Deprecated as of 3.1 in favor of functional programming model。
SpringCloud 2020.0.4 版本中，已经不推荐这么开发了，因此这里我们也使用新的写法（函数式编程方式） 开发。
闲话不多说，直接上代码。
2. 消息出错重试
2.1 主要依赖
<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-stream-rabbit</artifactId>
</dependency>
2.2 消息实体类
@Setter
@Getter
public class MyMessage implements java.io.Serializable {
// 消息体
private String payload;
}
2.3 生产者
// 消息出错重试
@GetMapping("/retry")
public String sendRetryMessage(@RequestParam("body") String body) {
MyMessage myMessage = new MyMessage();
myMessage.setPayload(body);
// 生产消息
// 第一个参数是绑定名称，格式为：自定义的绑定名称-out-0，myRetry是自定义的绑定名称，out代表生产者，0是固定写法
// 自定义的绑定名称必须与消费方法的方法名保持一致
// 第二个参数是发送的消息实体
streamBridge.send("myRetry-out-0", myMessage);
return "SUCCESS";
}
2.4 消费者
// 消息出错重试
@Bean
public Consumer<MyMessage> myRetry() { // 方法名必须与生产消息时自定义的绑定名称一致
return message -> {
log.info("接收消息：{}", message.getPayload());
throw new RuntimeException("消费报错");
};
}
2.5 application.yml 配置
spring:
application:
name: my-stream-new
rabbitmq: # RabbitMQ 配置
addresses: 192.168.1.12:5672
username: guest
password: guest
virtual-host: /
connection-timeout: 16000
cloud:
function:
# 定义消费者，多个用分号分隔，当存在大于1个的消费者时，不定义不会生效
definition: myRetry
stream:
bindings:
# 消息出错重试
myRetry-in-0:
destination: my-retry-topic
# 配置重试次数（本机重试）
# 次数等于 1 ，相当于不重试
consumer:
max-attempts: 3
myRetry-out-0:
destination: my-retry-topic
2.6 验证消息出错重试
发送消息接口：
Get http://localhost:49000/stream/retry?body=出错重试消息
自动生成的 Exchange
image
自动生成的 Queue
image
消费情况
image
3. 死信队列
3.1 生产者
// 死信队列
@GetMapping("/dlq")
public String sendDlqMessage(@RequestParam("body") String body) {
MyMessage myMessage = new MyMessage();
myMessage.setPayload(body);
// 生产消息
// 第一个参数是绑定名称，格式为：自定义的绑定名称-out-0，myDlq是自定义的绑定名称，out代表生产者，0是固定写法
// 自定义的绑定名称必须与消费方法的方法名保持一致
// 第二个参数是发送的消息实体
streamBridge.send("myDlq-out-0", myMessage);
return "SUCCESS";
}
3.2 消费者
// 死信队列
@Bean
public Consumer<MyMessage> myDlq() { // 方法名必须与生产消息时自定义的绑定名称一致
return message -> {
log.info("接收消息：{}", message.getPayload());
throw new RuntimeException("消费报错");
};
}
3.3 application.yml 配置
spring:
application:
name: my-stream-new
rabbitmq: # RabbitMQ 配置
addresses: 192.168.1.12:5672
username: guest
password: guest
virtual-host: /
connection-timeout: 16000
cloud:
function:
# 定义消费者，多个用分号分隔，当存在大于1个的消费者时，不定义不会生效
definition: myRetry;myDlq
stream:
bindings:
# 消息出错重试
myRetry-in-0:
destination: my-retry-topic
# 配置重试次数（本机重试）
# 次数等于 1 ，相当于不重试
consumer:
max-attempts: 3
myRetry-out-0:
destination: my-retry-topic
# 死信队列
myDlq-in-0:
destination: my-dlq-topic
group: dlq-group
# 配置重试次数（本机重试）
# 次数等于 1 ，相当于不重试
consumer:
max-attempts: 3
myDlq-out-0:
destination: my-dlq-topic
rabbit:
bindings:
# 死信队列
myDlq-in-0:
consumer:
autoBindDlq: true # 自动绑定死信队列，会自动创建一个默认的死信队列
myDlq-out-0:
producer:
autoBindDlq: true # 自动绑定死信队列，会自动创建一个默认的死信队列
3.4 验证死信队列
发送消息接口：
GET http://localhost:49000/stream/dlq?body=死信队列
自动生成的 Exchange
image
自动生成的 Queue
image
消费情况
image
死信队列情况
image
4. 综述
今天聊了一下 SpringCloud Stream 组件 消息出错重试 与 死信队列 的实现 ，希望可以对大家的工作有所帮助。
欢迎帮忙点赞、评论、转发、加关注 ：）
关注追风人聊Java，每天更新Java干货。
