springboot整合rabbitmaq延时队列

Spring Boot 整合 RabbitMQ 延时队列是一种常见场景，主要用于处理那些需要延迟执行的任务。在Spring Boot中，你可以使用`RabbitmqTemplate`和`AmqpAdmin`来集成RabbitMQ，并利用`delayed-message-exchange`和`x-delayed-message-routing-key`特性实现延时队列。

以下是基本步骤：

1. **配置RabbitMQ**: 首先，在Spring Boot的配置文件中（如application.yml）添加RabbitMQ的相关配置，包括连接地址、交换器、队列等。

spring:
  rabbit:
    host: localhost
    port: 5672
    username: guest
    password: guest
    virtual-host: /

2. **声明延时队列**: 使用`AmqpAdmin`创建一个延时队列，它需要一个特殊的交换器（通常是`x-delayed-message`类型）。

@Autowired
private AmqpAdmin admin;

public void configureQueue(String queueName) {
    ExchangeOptions exchangeOptions = new ExchangeOptions()
        .type(ExchangeTypes.X_DELAYED_MESSAGE)
        .durable(true);
    admin.declareExchange("delayed_queue", exchangeOptions);

    Map<String, Object> bindingArgs = new HashMap<>();
    bindingArgs.put("x-delayed-type", "direct");
    admin.declareQueue(queueName, true, true, bindingArgs);
}

3. **发送延时消息**: 在发送消息时，指定`x-delay`属性为延迟时间（单位通常为毫秒），并使用`x-delayed-message-routing-key`作为路由键。

RabbitTemplate rabbitTemplate = new RabbitTemplate(rabbitConnectionFactory);
Map<String, Object> messageProperties = new HashMap<>();
messageProperties.put(RabbitMQMessageHeaders.DELIVERY_MODE, DeliveryMode.PERSISTENT);
messageProperties.put(RabbitMQMessageHeaders.EXPIRE_TIME, System.currentTimeMillis() + (delayInMilliseconds * 1000)); // delayInMilliseconds是你希望的延迟时间
rabbitTemplate.convertAndSend("delayed_queue", "your.routing.key", "your.message", messageProperties);

4. **消费延时消息**: 使用普通的消费者或者延时消费者（`@RabbitListener`或自定义`MessageListenerContainer`）来消费队列中的消息。对于延时消费，你可以监听特定的路由键，或者使用`@RabbitListener`的`container`参数指定一个带有延时策略的`ConcurrentMessageListenerContainer`。

@Component
public class DelayedConsumer {
    @RabbitListener(queues = "your.queue.name", containerFactory = "delayedListenerContainer")
    public void consume(String message) {
        // 处理消息
    }

    @Bean(name = "delayedListenerContainer")
    public ConcurrentMessageListenerContainer delayedListenerContainer() {
        ConcurrentMessageListenerContainer container = new ConcurrentMessageListenerContainer(rabbitConnectionFactory);
        container.setDeadLetterExchange("dead_letter_exchange");
        return container;
    }
}