分布式中间件：RabbitMQ死信队列和延迟队列

分布式 中间件：RabbitMQ死信队列和延迟队列

引言

在分布式系统的开发中，消息队列是一种常用的通信机制，它可以帮助我们实现系统之间的解耦、异步处理和流量削峰等功能。RabbitMQ 作为一款功能强大的消息队列中间件，提供了许多高级特性，其中死信队列和延迟队列在实际应用中非常实用。本文将详细介绍 RabbitMQ 的死信队列和延迟队列，并结合具体的代码示例进行讲解。

RabbitMQ 配置类

首先，我们来看一个 RabbitMQ 的配置类，这个配置类定义了一些基本的 RabbitMQ 配置，包括单一消费者实例配置、多个消费者实例配置以及自定义的 RabbitMQ 发送消息组件。

@Configuration
public class RabbitmqConfig {@Autowiredprivate Environment environment; // 注入rabbitmq环境@Autowiredprivate CachingConnectionFactory connectionFactory; // 注入rabbitmq连接工厂@Autowiredprivate SimpleRabbitListenerContainerFactoryConfigurer factoryConfigurer; // 注入rabbitmq配置// 单一消费者实例配置@Bean("singleListenerContainer")public SimpleRabbitListenerContainerFactory singleListenerContainer() {// 创建一个监听容器工厂SimpleRabbitListenerContainerFactory factory = new SimpleRabbitListenerContainerFactory();// 设置连接工厂factory.setConnectionFactory(connectionFactory);// 设置消息转换器factory.setMessageConverter(new Jackson2JsonMessageConverter());// 设置并发消费factory.setConcurrentConsumers(1);// 设置最大并发消费factory.setMaxConcurrentConsumers(1);// 设置最大单条消费消息factory.setPrefetchCount(1);return factory;}// 多个消费者实例@Bean("multiListenerContainer")public SimpleRabbitListenerContainerFactory multiListenerContainer() {// 创建一个监听容器工厂SimpleRabbitListenerContainerFactory factory = new SimpleRabbitListenerContainerFactory();// 设置连接工厂factoryConfigurer.configure(factory, connectionFactory);// 设置消息转换器factory.setMessageConverter(new Jackson2JsonMessageConverter());// 设置手动提交factory.setAcknowledgeMode(AcknowledgeMode.NONE);// 设置并发消费factory.setConcurrentConsumers(10);// 设置最大并发消费factory.setMaxConcurrentConsumers(20);// 设置最大单条消费消息factory.setPrefetchCount(10);return factory;}// 自定义RabbitMQ发送消息组件@Beanpublic RabbitTemplate rabbitTemplate() {// 设置消息发送确认connectionFactory.setPublisherConfirms(true);// 设置消息发送返回connectionFactory.setPublisherReturns(true);// 创建rabbitTemplateRabbitTemplate rabbitTemplate = new RabbitTemplate(connectionFactory);// 设置消息发送确认回调rabbitTemplate.setMandatory(true);// 设置消息发送确认回调rabbitTemplate.setConfirmCallback((correlationData, ack, cause) -> {if (ack) {System.out.println("消息发送成功");} else {System.out.println("消息发送失败:" + cause + correlationData.toString());}});// 设置消息发送返回回调rabbitTemplate.setReturnCallback((message, replyCode, replyText, exchange, routingKey) -> {System.out.println("消息丢失：exchange:" + exchange + ",route:" + routingKey + ",replyCode:" + replyCode + ",replyText:" + replyText);});// 设置消息转换器rabbitTemplate.setMessageConverter(new Jackson2JsonMessageConverter());return rabbitTemplate;}
}

代码解释

单一消费者实例配置（singleListenerContainer）：
- 创建了一个 SimpleRabbitListenerContainerFactory 实例，用于处理单一消费者的消息监听。
- 设置了连接工厂、消息转换器、并发消费者数量、最大并发消费者数量和最大单条消费消息数量。
多个消费者实例配置（multiListenerContainer）：
- 同样创建了一个 SimpleRabbitListenerContainerFactory 实例，用于处理多个消费者的消息监听。
- 配置了手动提交模式，设置了更大的并发消费者数量和最大并发消费者数量。
自定义 RabbitMQ 发送消息组件（rabbitTemplate）：
- 设置了消息发送确认和返回机制。
- 定义了消息发送确认回调和返回回调，用于处理消息发送成功和失败的情况。
- 设置了消息转换器，使用 Jackson2JsonMessageConverter 进行消息的序列化和反序列化。

死信队列

什么是死信队列

死信队列（Dead Letter Queue）是一种特殊的队列，当消息满足某些条件时，会被发送到死信队列中。这些条件包括：

消息被拒绝（basic.reject 或 basic.nack）并且 requeue 参数设置为 false。
消息过期（TTL）。
队列达到最大长度。

死信队列的应用场景

消息重试：当消息处理失败时，可以将消息发送到死信队列，经过一定的处理后再重新发送到原队列进行重试。
记录失败消息：将失败的消息存储在死信队列中，方便后续的排查和处理。

代码示例

以下是一个简单的死信队列配置示例：

@Configuration
public class DeadLetterConfig {// 正常交换机public static final String NORMAL_EXCHANGE = "normal.exchange";// 正常队列public static final String NORMAL_QUEUE = "normal.queue";// 死信交换机public static final String DEAD_LETTER_EXCHANGE = "dead.letter.exchange";// 死信队列public static final String DEAD_LETTER_QUEUE = "dead.letter.queue";// 路由键public static final String ROUTING_KEY = "normal.routing.key";// 声明正常交换机@BeanDirectExchange normalExchange() {return new DirectExchange(NORMAL_EXCHANGE);}// 声明死信交换机@BeanDirectExchange deadLetterExchange() {return new DirectExchange(DEAD_LETTER_EXCHANGE);}// 声明正常队列并绑定死信交换机@BeanQueue normalQueue() {Map<String, Object> args = new HashMap<>();// 设置死信交换机args.put("x-dead-letter-exchange", DEAD_LETTER_EXCHANGE);// 设置死信路由键args.put("x-dead-letter-routing-key", ROUTING_KEY);return new Queue(NORMAL_QUEUE, true, false, false, args);}// 声明死信队列@BeanQueue deadLetterQueue() {return new Queue(DEAD_LETTER_QUEUE, true);}// 绑定正常队列和正常交换机@BeanBinding normalBinding() {return BindingBuilder.bind(normalQueue()).to(normalExchange()).with(ROUTING_KEY);}// 绑定死信队列和死信交换机@BeanBinding deadLetterBinding() {return BindingBuilder.bind(deadLetterQueue()).to(deadLetterExchange()).with(ROUTING_KEY);}
}

代码解释

交换机和队列的声明：
- 声明了正常交换机、正常队列、死信交换机和死信队列。
- 在正常队列的声明中，通过 args 参数设置了死信交换机和死信路由键。
绑定关系：
- 将正常队列绑定到正常交换机，将死信队列绑定到死信交换机。

延迟队列

什么是延迟队列

延迟队列是一种可以让消息在指定的时间后才被消费的队列。在 RabbitMQ 中，可以通过消息的 TTL（Time To Live）和死信队列来实现延迟队列的功能。

延迟队列的应用场景

订单超时处理：当用户下单后，如果在一定时间内没有支付，可以使用延迟队列来处理订单超时的情况。
缓存刷新：定期刷新缓存，可以使用延迟队列来实现定时任务。

代码示例

以下是一个简单的延迟队列配置示例：

@Configuration
public class DelayQueueConfig {// 延迟交换机public static final String DELAY_EXCHANGE = "delay.exchange";// 延迟队列public static final String DELAY_QUEUE = "delay.queue";// 死信交换机public static final String DEAD_LETTER_EXCHANGE = "dead.letter.exchange";// 死信队列public static final String DEAD_LETTER_QUEUE = "dead.letter.queue";// 路由键public static final String ROUTING_KEY = "delay.routing.key";// 声明延迟交换机@BeanDirectExchange delayExchange() {return new DirectExchange(DELAY_EXCHANGE);}// 声明死信交换机@BeanDirectExchange deadLetterExchange() {return new DirectExchange(DEAD_LETTER_EXCHANGE);}// 声明延迟队列并绑定死信交换机@BeanQueue delayQueue() {Map<String, Object> args = new HashMap<>();// 设置死信交换机args.put("x-dead-letter-exchange", DEAD_LETTER_EXCHANGE);// 设置死信路由键args.put("x-dead-letter-routing-key", ROUTING_KEY);// 设置消息的 TTLargs.put("x-message-ttl", 5000); // 5 秒return new Queue(DELAY_QUEUE, true, false, false, args);}// 声明死信队列@BeanQueue deadLetterQueue() {return new Queue(DEAD_LETTER_QUEUE, true);}// 绑定延迟队列和延迟交换机@BeanBinding delayBinding() {return BindingBuilder.bind(delayQueue()).to(delayExchange()).with(ROUTING_KEY);}// 绑定死信队列和死信交换机@BeanBinding deadLetterBinding() {return BindingBuilder.bind(deadLetterQueue()).to(deadLetterExchange()).with(ROUTING_KEY);}
}