RabbitMQ高级特性--重试特性

1.重试配置

spring:
  rabbitmq:
    addresses: amqp://study:study@你的服务器IP:15673/你的虚拟机名
    listener:
      simple:
        acknowledge-mode: auto #消息接收确认
      retry:
        enabled: true # 开启消费者失败重试
        initial-interval: 5000ms # 初始失败等待时⻓为5秒
        max-attempts: 5 # 最⼤重试次数（包括自身消费的一次）

2.配置交换机&队列

//重试机制
public static final String RETRY_QUEUE = "retry_queue";
public static final String RETRY_EXCHANGE_NAME = "retry_exchange";


/重试机制 发布订阅模式
//1. 交换机
@Bean("retryExchange")
public Exchange retryExchange() {
 return
ExchangeBuilder.fanoutExchange(Constant.RETRY_EXCHANGE_NAME).durable(true).build();
}
//2. 队列
@Bean("retryQueue")
public Queue retryQueue() {
 return QueueBuilder.durable(Constant.RETRY_QUEUE).build();
}
//3. 队列和交换机绑定 Binding
@Bean("retryBinding")
public Binding retryBinding(@Qualifier("retryExchange") FanoutExchange 
exchange, @Qualifier("retryQueue") Queue queue) {
 return BindingBuilder.bind(queue).to(exchange);
}

3.发送消息

@RequestMapping("/retry")
public String retry(){
 rabbitTemplate.convertAndSend(Constant.RETRY_EXCHANGE_NAME, "", "retry 
test...");
 return "发送成功!";
}

4.消费消息

@Component
public class RetryQueueListener {
 //指定监听队列的名称
 @RabbitListener(queues = Constant.RETRY_QUEUE)
 public void ListenerQueue(Message message) throws Exception {
 System.out.printf("接收到消息: %s, deliveryTag: %d%n", new
String(message.getBody(),"UTF-8"), 
message.getMessageProperties().getDeliveryTag());
 //模拟处理失败
 try {
 int num = 3/0;
 System.out.println("处理完成");
}catch (Exception e){
 System.out.println("处理失败");
}
 }
}

5. 运行程序观察结果

接收到消息: consumer ack test..., deliveryTag: 1
处理失败

6. 手动确认

@RabbitListener(queues = Constant.RETRY_QUEUE)
public void ListenerQueue(Message message, Channel channel) throws Exception {
 long deliveryTag = message.getMessageProperties().getDeliveryTag();
 try {
 System.out.printf("接收到消息: %s, deliveryTag: %d%n", new
String(message.getBody(),"UTF-8"), 
message.getMessageProperties().getDeliveryTag());
 //模拟处理失败
 int num = 3/0;
 System.out.println("处理完成");
 //3. ⼿动签收
 channel.basicAck(deliveryTag, true);
 }catch (Exception e){
 //4. 异常了就拒绝签收
 Thread.sleep(1000);
 //第三个参数requeue, 是否重新发送, 如果为true, 则会重新发送,,若为false, 则直接
丢弃
 channel.basicNack(deliveryTag, true,true);
 }
}

运行结果：

接收到消息: retry test..., deliveryTag: 1
接收到消息: retry test..., deliveryTag: 2
接收到消息: retry test..., deliveryTag: 3
接收到消息: retry test..., deliveryTag: 4
接收到消息: retry test..., deliveryTag: 5
接收到消息: retry test..., deliveryTag: 6
接收到消息: retry test..., deliveryTag: 7
接收到消息: retry test..., deliveryTag: 8
接收到消息: retry test..., deliveryTag: 9
接收到消息: retry test..., deliveryTag: 10
接收到消息: retry test..., deliveryTag: 11

 注意：