Spring Boot与RabbitMQ结合实现延迟队列的示例

一、介绍

RabbitMQ是一个被广泛使用的消息队列中间件，而延迟队列则是RabbitMQ中常用的功能之一。本文将详细讲解Spring Boot和RabbitMQ结合实现延迟队列的具体实现方式，以及通过两个示例来说明实现的过程。

二、实现步骤

1. 添加依赖

在pom.xml文件中添加以下依赖：

<dependency>
    <groupId>org.springframework.amqp</groupId>
    <artifactId>spring-rabbit</artifactId>
    <version>2.2.4.RELEASE</version>
</dependency>

1. 配置RabbitMQ

在application.yml中添加RabbitMQ的配置：

spring:
  rabbitmq:
    host: localhost
    port: 5672
    username: guest
    password: guest

1. 创建消息接收者

创建一个消息接收者，用于接收延迟消息所触发的消息：

@Component
public class DelayedReceiver {

    @RabbitListener(queues = "${delayed.queue}")
    public void receive(String message) {
        log.info("Received message: {}", message);
    }
}

其中，@RabbitListener注解表示该方法用于监听名为${delayed.queue}的队列，一旦该队列有消息到达，则会触发receive方法并将消息内容传入。

1. 创建延迟队列

为了实现延迟队列的功能，我们需要通过RabbitMQ提供的“死信队列”的机制来实现。具体实现步骤如下：

• 创建正常的业务队列（例如news.queue）；
• 创建死信队列（例如news.dlx）；
• 将正常的业务队列绑定到死信队列上（即将news.queue的DLX属性设置为news.dlx）；
• 创建一个延迟队列（例如news.delayed.queue），并将其绑定到正常的业务队列上。

代码如下：

@Configuration
public class DelayedConfig {

  private static final String NEWS_QUEUE = "news.queue";
  private static final String NEWS_DELAYED_QUEUE = "news.delayed.queue";
  private static final String NEWS_DLX = "news.dlx";

  @Bean
  Queue newsQueue() {
    return QueueBuilder.durable(NEWS_QUEUE)
        .withArgument("x-dead-letter-exchange", "")
        .withArgument("x-dead-letter-routing-key", NEWS_DLX)
        .build();
  }

  @Bean
  Queue newsDelayedQueue() {
    return QueueBuilder.durable(NEWS_DELAYED_QUEUE)
        .withArgument("x-delayed-type", "direct")
        .withArgument("x-dead-letter-exchange", "")
        .withArgument("x-dead-letter-routing-key", NEWS_QUEUE)
        .build();
  }

  @Bean
  DirectExchange newsExchange() {
    return ExchangeBuilder.directExchange("news.exchange").durable().build();
  }

  @Bean
  Binding newsBinding(Queue newsQueue, DirectExchange newsExchange) {
    return BindingBuilder.bind(newsQueue).to(newsExchange).with("news.routing.key");
  }

  @Bean
  Binding newsDelayedBinding(Queue newsDelayedQueue, DirectExchange newsExchange) {
    return BindingBuilder.bind(newsDelayedQueue).to(newsExchange).with("news.delayed.routing.key");
  }

  @Bean
  Queue newsDlxQueue() {
    return QueueBuilder.durable(NEWS_DLX).build();
  }

  @Bean
  DirectExchange newsDlxExchange() {
    return ExchangeBuilder.directExchange(NEWS_DLX).durable().build();
  }

  @Bean
  Binding newsDlxExchangeBinding(Queue newsDlxQueue, DirectExchange newsDlxExchange) {
    return BindingBuilder.bind(newsDlxQueue).to(newsDlxExchange).with(NEWS_DLX);
  }

  @Bean
  SimpleMessageListenerContainer newsDelayedListenerContainer(ConnectionFactory connectionFactory,
                                                               DelayedReceiver delayedReceiver) {
    SimpleMessageListenerContainer container = new SimpleMessageListenerContainer();
    container.setConnectionFactory(connectionFactory);
    container.setQueueNames(NEWS_DELAYED_QUEUE);
    container.setMessageListener(new MessageListenerAdapter(delayedReceiver));
    return container;
  }
}

在以上代码中：

• newsQueue用于创建正常的业务队列；
• newsDelayedQueue用于创建延迟队列，并将其设置为“x-delayed-type”为“direct”，表示消息将通过“direct”方式进行路由；
• newsExchange和newsDlxExchange分别用于创建正常业务队列和死信队列所使用的交换机；
• newsBinding和newsDelayedBinding分别用于将正常业务队列和延迟队列绑定到交换机上；
• newsDlxQueue用于创建死信队列；
• newsDlxExchangeBinding用于将死信队列绑定到交换机上；

• newsDelayedListenerContainer用于创建一个MessageListenerContainer，用于监听延迟队列。

• 创建消息发送者

创建一个消息发送者，用于发送延迟消息。在消息发送者中，我们需要将消息发送到延迟队列中，并且设置消息的TTL属性。代码如下：

@Service
public class DelayedSender {

    @Autowired
    private RabbitTemplate rabbitTemplate;

    public void send(String message, long delayTime) {
        log.info("Sending message: {} with delay :{}", message, delayTime);
        rabbitTemplate.convertAndSend("news.exchange", "news.delayed.routing.key",
                message, new MessagePostProcessor() {
            @Override
            public Message postProcessMessage(Message message) throws AmqpException {
                message.getMessageProperties().setDelay((int) delayTime);
                return message;
            }
        });
    }
}

在以上代码中，我们使用rabbitTemplate来将消息发送到名为news.exchange的交换机上，然后指定将消息发送到名为news.delayed.routing.key的延迟队列中，并且设置消息的TTL属性为delayTime。

三、示例1

下面以发送一条延迟60秒钟的消息为例进行说明：

@Service
public class TestService {

    private static final long DELAY_TIME = 60 * 1000;

    @Autowired
    private DelayedSender delayedSender;

    public void test() {
        delayedSender.send("Test message", DELAY_TIME);
    }
}

在上述示例中，我们将消息发送到延迟时间为60秒的延迟队列中。

四、示例2

下面以发送多条延迟消息为例进行说明：

@Service
public class TestService {

    private static final long DELAY_10_SECONDS = 10 * 1000;
    private static final long DELAY_20_SECONDS = 20 * 1000;
    private static final long DELAY_30_SECONDS = 30 * 1000;

    @Autowired
    private DelayedSender delayedSender;

    public void test() {
        delayedSender.send("Message A", DELAY_10_SECONDS);
        delayedSender.send("Message B", DELAY_20_SECONDS);
        delayedSender.send("Message C", DELAY_30_SECONDS);
    }
}

在以上示例中，我们将一共发送三条延迟消息，分别延迟10秒，20秒和30秒。通过以上两个示例，我们可以看到Spring Boot与RabbitMQ结合实现延迟队列的过程及代码实现方式。