RabbitMQ实现消费端异常处理
当消费侧出现异常时，若未进行处理，可能会导致消息再次被消费，然后再次出现异常，进入循环
官方文档：https://www.rabbitmq.com/confirms.html
前言
思考：因为在开发项目时，RabbitMQ的消费端出现了异常（工具类操作文件时，未找到文件路径）。由于在此之前并未对该异常进行预判，导致异常出现后，消费端仍然对MQ的消息进行消费，但是出现异常后无法对MQ进行回复，所以造成后果消费端一直消费该条信息，进入死循环！
另需注意：若在消费者上加了声明式事物，基于事务的传播行为，默认为加入当前事务，这样以来，如果内部报错，即便在外层加了try-catch，事务也会进行回滚，因为默认是在最后进行ack的，消息就不会被ack，从而再次被消费。
从而引发了自己的思考：1. 开发时难免会出现异常，这种异常如果事先未预判，那么在程序运行中，消费端该怎么避免以上出现的死循环；2. 如果事先预判到异常，对其进行了抛出或捕获，消费端又该如何表现？
异常
第一种方法，可以对可能发生异常的部分try、catch；只要事先将问题catch住，就证明消费端已经将该问题消费掉，然后该消息就不存在于队列中，不会造成无限报错的情况。这里，你可以在catch中写一些业务，把这个出现异常的“消息”记录到数据库或者怎么怎么处理，反正是相当于被消费掉了。
第二种方法，因为RabbitMQ 默认的异常策略是不断重试，除非抛出了fatal（致命）类型的异常，这种异常类型如下
异常类
MessageConversionException
MessageConversionException
MethodArgumentNotValidException
MethodArgumentTypeMismatchException
NoSuchMethodException
ClassCastException
所以可以在catch语句块中抛出fatal 类型的异常（虽然不推荐，但也是种方式）
第三种方法，”消费者重试“模式。基本配置同一，只是在catch中显式的抛异常。这样其实就和没有catch差不多，就相当于未知状况下出现了异常。catch是为了解决业务问题，在这里处理自己需要的业务。catch中的throw有什么用呢？
throw配合着application.yml中的“开启消费者重试”模式：若异常发生，重试n次（n为yml中的 max-attempts），之后消息就自动进入死信队列（或者如果没配置死信队列，消息被扔掉）。
具体如下，消费者的mq配置类中设置了死信队列（参数只有死信交换机和路由，没有TTL）。
@Configuration
public class RabbitmqConfig {
public static final String EXCHANGE_TOPICS_INFORM = "exchange_topics_inform";
private static final String DEAD_INFO_EXCHANGE = "x-dead-letter-exchange";
public static final String QUEUE_INFORM_EMAIL = "queue_inform_email";
public static final String ROUTE_KEY_EMAIL = "inform.#.email.#";
public static final String DEAD_INFO_QUEUE = "dead_info_queue";
public static final String DEAD_ROUTE_KEY = "dead_info_dev";
//声明交换机 Topic交换机
@Bean
public Exchange exchangeTopicsInform() {
return ExchangeBuilder.topicExchange(EXCHANGE_TOPICS_INFORM).durable(true).build();
}
//声明QUEUE_INFORM_EMAIL队列，配置死信队列需要的参数
@Bean
public Queue queueInformEmail() {
// Map<String, Object> map = new HashMap<>();
// key固定，value根据业务
// map.put("x-dead-letter-exchange", DEAD_INFO_EXCHANGE);
// map.put("x-dead-letter-routing-key", DEAD_ROUTE_KEY);
// return new Queue(QUEUE_INFORM_EMAIL, true, false, false, map);
return QueueBuilder
.durable(QUEUE_INFORM_EMAIL)
// .withArguments(map)
// 可以配置个时间，到时间后自动转发到死信队列 ms
.withArgument("x-message-ttl", 10000)
// 满足要求后转发的死信交换机及路由键
.withArgument("x-dead-letter-exchange", DEAD_INFO_EXCHANGE)
.withArgument("x-dead-letter-routing-key", DEAD_ROUTE_KEY)
.build();
}
//ROUTE_KEY_EMAIL队列绑定交换机，指定routingKey
@Bean
public Binding bindingQueueInformEmail(Queue queueInformEmail, Exchange exchangeTopicsInform) {
return BindingBuilder.bind(queueInformEmail).to(exchangeTopicsInform).with(ROUTE_KEY_EMAIL).noargs();
}
//以下为死信队列
// 交换机
@Bean
public Exchange deadInfoExchange() {
return ExchangeBuilder.directExchange(DEAD_INFO_EXCHANGE).durable(true).build();
}
@Bean
public Queue deadInfoQueue() {
return QueueBuilder.durable(DEAD_INFO_QUEUE).build();
}
@Bean
public Binding deadInfoQueueBind(Queue deadInfoQueue, Exchange deadInfoExchange) {
return BindingBuilder.bind(deadInfoQueue).to(deadInfoExchange).with(DEAD_ROUTE_KEY).noargs();
}
}消费者端不做任何异常处理，模拟开发时并不知道会出现异常的情况。（注释掉的，catch里的throw和这个是一样的效果）
@RabbitListener(queues = "queue_inform_email")
public void receiveMediaProcessTask(String msg, Channel channel, Message message){
System.out.println("Listen===========" + msg);
int i = 1;
int b = i/0;
System.out.println("解决了");
// try {
// System.out.println("Listen===========" + msg);
// int i = 1;
// int b = i/0;
// }catch (Exception e){
// System.out.println("解决了");
// throw new RuntimeException("还是不行");
// }
}但是配置文件中开启“消费者尝试”，并配置最大尝试数。
rabbitmq:
host: 127.0.0.1
port: 5672
username: guest
password: guest
virtual-host: /
listener:
simple:
concurrency: 1 # Minimum number of consumers.
max-concurrency: 20 # Maximum number of consumers.
prefetch: 50
default-requeue-rejected: true #意思是，消息被拒后（即未消费），重新（true）放入队列
retry:
enabled: true #是否开启消费者重试（为false时关闭消费者重试，这时消费端代码异常会一直重复收到消息）
max-attempts: 3
initial-interval: 5000ms这样，消费端发现了异常，尝试了规定次数后，这条“问题消息”就会被解决（如果设置了死信队列，就被送到了死信队列；否则直接扔掉）。是开启了“消费者重试尝试”的功劳。如果不开启该模式，那么会无限的循环下去。和 “default-requeue-rejected: true”参数没有任何关系，“消费者重试”模式会覆盖掉default-requeue-rejected(默认为true)。所以，只要是开了该模式，异常就可以被解决。如果只设置 default-requeue-rejected: true（消费者重试未开启，应答方式为默认），那么会无限报错！
第四种，只设置 default-requeue-rejected: false（消费者重试未开启，应答方式为默认），异常只出现一次，然后该“问题消息”被解决（如果设置了死信队列，就被送到了死信队列；否则直接扔掉）。
第五种，在队列中设置了TTL参数！！！那么异常会无脑的跑一会，当消息到了一定时间就会过期，自动进入死信队列。这是TTL的功劳。
@Bean(QUEUE_INFORM_EMAIL)
public Queue QUEUE_INFORM_EMAIL(){
Map<String,Object> map = new HashMap<>();
//设置TTL
map.put("x-message-ttl",10000);
map.put("x-dead-letter-exchange",DEAD_EXCHANGE);
map.put("x-dead-letter-routing-key","dev");
return new Queue(QUEUE_INFORM_EMAIL,true,false,false,map);
}目前为止，都是自动（acknowledge-mode默认auto）应答mq，不需要手动应答。
第六种，yml配置文件手动应答,见最后一行的配置。
spring:
application:
name: test-rabbitmq-producer
rabbitmq:
host: 127.0.0.1
port: 5672
username: guest
password: guest
virtual-host: /
listener:
simple:
concurrency: 1 # Minimum number of consumers.
max-concurrency: 20 # Maximum number of consumers.
prefetch: 50
acknowledge-mode: manual #关键 消费方手动ack这时，消费端的监听需要如下这样，参照死信队列的概念，channel.basicReject的requeue参数必须设为false。
@RabbitListener(queues = "queue_inform_email")
public void receiveMediaProcessTask(String msg, Channel channel, Message message) throws IOException {
try {
System.out.println("Listen===========" + msg);
int i = 1;
int b = i/0;
}catch (Exception e){
//手动应答，采取拒绝，第二位参数requeue，必须设置为false
channel.basicReject(message.getMessageProperties().getDeliveryTag(),false);
System.out.println("解决了");
//下面的抛异常就随意了，因为上面已经把当前的消息扔到队列外，所以不会无限执行该条消息，也就是说，抛异常只会抛一次，并不会无限下去。
throw new RuntimeException("还是不行？？");
}
}如果把requeue的值设为true，那就白玩了，“问题消息”又被你放到了当前队列，下一次消费方又执行这条“问题消息”。可以看出，第六种方案的推行并不依赖于“消费端重试”和TTL，仅仅依照死信队列的定义：利用basicReject拒绝，并把requeue设置为false.
注意：如果是，不管是否设置“消费者重试”模式，配置了default-requeue-rejected: false，且手动应答，异常只会出现一次，但是不会进入死信队列。消息以unack形式存在队列中。
综上所述，我们可以发现消费端异常的几种方案的特点：
- TTL可以设置消息的过期时间，不管你是不是无脑抛异常，只要过期，就进入死信队列；
- “消费者重试”模式，只要你抛异常抛到了我的底线（次数达标），那我就把你送走，可能是直接扔了，也可能是扔到死信队列；
- try、catch，只要你能提前预判，捕获到相应异常，那就平平安安，没有一点波澜；
- 手动回应，需要提前知道哪里会出错，就在哪里拒绝，而且requeue设成false；还要在哪里不拒绝（普通的消息回应），对mq做出相应正确的反馈
其实从这些特点可以看出，死信的定义就是最好的答案。
- 死信的产生：
- 消息被拒绝(basic.reject / basic.nack)，并且requeue = false
- 消息TTL过期
- 队列达到最大长度
附，配置属性说明
# base
spring.rabbitmq.host: 服务Host
spring.rabbitmq.port: 服务端口
spring.rabbitmq.username: 登陆用户名
spring.rabbitmq.password: 登陆密码
spring.rabbitmq.virtual-host: 连接到rabbitMQ的vhost
spring.rabbitmq.addresses: 指定client连接到的server的地址，多个以逗号分隔(优先取addresses，然后再取host)
spring.rabbitmq.requested-heartbeat: 指定心跳超时，单位秒，0为不指定；默认60s
spring.rabbitmq.publisher-confirms: 是否启用【发布确认】
spring.rabbitmq.publisher-returns: 是否启用【发布返回】
spring.rabbitmq.connection-timeout: 连接超时，单位毫秒，0表示无穷大，不超时
spring.rabbitmq.parsed-addresses:
# ssl
spring.rabbitmq.ssl.enabled: 是否支持ssl
spring.rabbitmq.ssl.key-store: 指定持有SSL certificate的key store的路径
spring.rabbitmq.ssl.key-store-password: 指定访问key store的密码
spring.rabbitmq.ssl.trust-store: 指定持有SSL certificates的Trust store
spring.rabbitmq.ssl.trust-store-password: 指定访问trust store的密码
spring.rabbitmq.ssl.algorithm: ssl使用的算法，例如，TLSv1.1
# cache
spring.rabbitmq.cache.channel.size: 缓存中保持的channel数量
spring.rabbitmq.cache.channel.checkout-timeout: 当缓存数量被设置时，从缓存中获取一个channel的超时时间，单位毫秒；如果为0，则总是创建一个新channel
spring.rabbitmq.cache.connection.size: 缓存的连接数，只有是CONNECTION模式时生效
spring.rabbitmq.cache.connection.mode: 连接工厂缓存模式：CHANNEL 和 CONNECTION
# listener
spring.rabbitmq.listener.simple.auto-startup: 是否启动时自动启动容器
spring.rabbitmq.listener.simple.acknowledge-mode: 表示消息确认方式，其有三种配置方式，分别是none、manual和auto；默认auto
spring.rabbitmq.listener.simple.concurrency: 最小的消费者数量
spring.rabbitmq.listener.simple.max-concurrency: 最大的消费者数量
spring.rabbitmq.listener.simple.prefetch: 指定一个请求能处理多少个消息，如果有事务的话，必须大于等于transaction数量.
spring.rabbitmq.listener.simple.transaction-size: 指定一个事务处理的消息数量，最好是小于等于prefetch的数量.
spring.rabbitmq.listener.simple.default-requeue-rejected: 决定被拒绝的消息是否重新入队；默认是true（与参数acknowledge-mode有关系）
spring.rabbitmq.listener.simple.idle-event-interval: 多少长时间发布空闲容器时间，单位毫秒
spring.rabbitmq.listener.simple.retry.enabled: 监听重试是否可用
spring.rabbitmq.listener.simple.retry.max-attempts: 最大重试次数
spring.rabbitmq.listener.simple.retry.initial-interval: 第一次和第二次尝试发布或传递消息之间的间隔
spring.rabbitmq.listener.simple.retry.multiplier: 应用于上一重试间隔的乘数
spring.rabbitmq.listener.simple.retry.max-interval: 最大重试时间间隔
spring.rabbitmq.listener.simple.retry.stateless: 重试是有状态or无状态
# template
spring.rabbitmq.template.mandatory: 启用强制信息；默认false
spring.rabbitmq.template.receive-timeout: receive() 操作的超时时间
spring.rabbitmq.template.reply-timeout: sendAndReceive() 操作的超时时间
spring.rabbitmq.template.retry.enabled: 发送重试是否可用
spring.rabbitmq.template.retry.max-attempts: 最大重试次数
spring.rabbitmq.template.retry.initial-interval: 第一次和第二次尝试发布或传递消息之间的间隔
spring.rabbitmq.template.retry.multiplier: 应用于上一重试间隔的乘数
spring.rabbitmq.template.retry.max-interval: 最大重试时间间隔