背景
对于核心业务需要保证消息必须正常消费，就必须考虑消费失败的场景，rabbitmq提供了以下三种消费失败处理机制
- 直接reject，丢弃消息（默认）
- 返回nack，消息重新入队列
- 将失败消息投递到指定的交换机
对于核心业务，第一种方法显然不可接受，第二种方法如果代码有异常导致消费一直失败就会出现不断失败重新入队列的死循环问题，较好的方案是3，待消费失败问题修复后将消息从死信队列取出发回原队列重新消费。
实现
- rabbit版本
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-amqp</artifactId>
<version>2.6.3</version>
</dependency>
- 配置死信交换机，路由，队列
- 配置延迟消息业务队列消费失败投递到死信队列
@Bean("orderCloseQueue")
public Queue orderCloseQueue() {
return QueueBuilder.durable(OrderRabbitConstants.ORDER_CLOSE_QUEUE)
.deadLetterExchange(RabbitMqConstants.DEAD_LETTER_EXCHANGE)
.deadLetterRoutingKey(RabbitMqConstants.DEAD_LETTER_ROUTING_KEY)
.build();
}
- 配置手动返回ACK
@Bean(name = {"manualContainerFactory"})
public SimpleRabbitListenerContainerFactory manualContainerFactory(@Qualifier("connectionFactory") ConnectionFactory connectionFactory) {
SimpleRabbitListenerContainerFactory factory = new SimpleRabbitListenerContainerFactory();
this.manualFactoryConfigurer.configure(factory, connectionFactory);
factory.setAcknowledgeMode(AcknowledgeMode.MANUAL);
factory.setDefaultRequeueRejected(this.enableRequeueRejected);
if (this.enableConsumers) {
factory.setConcurrentConsumers(this.concurrentConsumers);
factory.setMaxConcurrentConsumers(this.maxConcurrentConsumers);
factory.setPrefetchCount(this.prefetchCount);
}
return factory;
}
- 业务队列消息消费模拟失败
@RabbitListener(queues = OrderRabbitConstants.ORDER_CLOSE_QUEUE, containerFactory = "manualContainerFactory")
public void consumerCloseOrder(Message message, Channel channel) throws IOException {
String orderCode = new String(message.getBody(), CharsetUtil.UTF_8);
String messageId = message.getMessageProperties().getMessageId();
log.info("收到MQ messageId[{}],订单号[{}]", messageId, orderCode);
if (true) {
channel.basicReject(message.getMessageProperties().getDeliveryTag(), false);
return;
}
}
- 效果
可以看到死信队列dead.letter.queue已经正常收到死信消息 - 编写逻辑将死信消费推回原队列
for (int i = 0; i < 10_000; i++) {
Message message = rabbitTemplate.receive(RabbitMqConstants.DEAD_LETTER_QUEUE);
if (message == null) {
return String.format("完成%d条", i);
}
log.info("拉取死信消息:[{}]", message);
try {
Map<String, Object> headers = message.getMessageProperties().getHeaders();
Map<String, Object> deathMap = ((List<Map<String, Object>>) headers.get("x-death")).get(0);
String exchange = deathMap.get("exchange").toString();
String routingKey = ((List) deathMap.get("routing-keys")).get(0).toString();
rabbitTemplate.send(exchange, routingKey, message);
} catch (Exception ex) {
log.error("消费死信消息失败", ex);
rabbitTemplate.send(RabbitMqConstants.DEAD_LETTER_EXCHANGE, RabbitMqConstants.DEAD_LETTER_ROUTING_KEY, message);
return "重入队列异常";
}
}
- 重推回业务队列效果
