Spring Boot与RabbitMQ结合实现延迟队列
fdxjj
7年前
<p>顾名思义，延迟队列就是进入该队列的消息会被延迟消费的队列。而一般的队列，消息一旦入队了之后就会被消费者马上消费。</p> <h3>延迟队列能做什么？</h3> <p>延迟队列多用于需要延迟工作的场景。最常见的是以下两种场景：</p> <ol> <li>延迟消费。比如： <ul> <li>用户生成订单之后，需要过一段时间校验订单的支付状态，如果订单仍未支付则需要及时地关闭订单。</li> <li>用户注册成功之后，需要过一段时间比如一周后校验用户的使用情况，如果发现用户活跃度较低，则发送邮件或者短信来提醒用户使用。</li> </ul> </li> <li>延迟重试。比如消费者从队列里消费消息时失败了，但是想要延迟一段时间后自动重试。</li> </ol> <p>如果不使用延迟队列，那么我们只能通过一个轮询扫描程序去完成。这种方案既不优雅，也不方便做成统一的服务便于开发人员使用。但是使用延迟队列的话，我们就可以轻而易举地完成。</p> <h3>如何实现？</h3> <p>别急，在下文中，我们将详细介绍如何利用 <strong>Spring Boot</strong> 加 <strong>RabbitMQ</strong> 来实现延迟队列。</p> <p>本文出现的示例代码都已push到Github仓库中： <a href="/misc/goto?guid=4959755495812036009" rel="nofollow,noindex">https://github.com/Lovelcp/blog-demos/tree/master/spring-boot-rabbitmq-delay-queue</a></p> <h2>实现思路</h2> <p>在介绍具体的实现思路之前，我们先来介绍一下RabbitMQ的两个特性，一个是Time-To-Live Extensions，另一个是Dead Letter Exchanges。</p> <h3>Time-To-Live Extensions</h3> <p>RabbitMQ允许我们为消息或者队列设置TTL（time to live），也就是过期时间。TTL表明了一条消息可在队列中存活的最大时间，单位为毫秒。也就是说，当某条消息被设置了TTL或者当某条消息进入了设置了TTL的队列时，这条消息会在经过TTL秒后“死亡”，成为Dead Letter。如果既配置了消息的TTL，又配置了队列的TTL，那么较小的那个值会被取用。更多资料请查阅 <a href="/misc/goto?guid=4959755495904819048" rel="nofollow,noindex">官方文档</a> 。</p> <h3>Dead Letter Exchange</h3> <p>刚才提到了，被设置了TTL的消息在过期后会成为Dead Letter。其实在RabbitMQ中，一共有三种消息的“死亡”形式：</p> <ol> <li>消息被拒绝。通过调用basic.reject或者basic.nack并且设置的requeue参数为false。</li> <li>消息因为设置了TTL而过期。</li> <li>消息进入了一条已经达到最大长度的队列。</li> </ol> <p>如果队列设置了Dead Letter Exchange（DLX），那么这些Dead Letter就会被重新publish到Dead Letter Exchange，通过Dead Letter Exchange路由到其他队列。更多资料请查阅 <a href="/misc/goto?guid=4959755495987669934" rel="nofollow,noindex">官方文档</a> 。</p> <h3>流程图</h3> <p>聪明的你肯定已经想到了，如何将RabbitMQ的TTL和DLX特性结合在一起，实现一个延迟队列。</p> <p>针对于上述的延迟队列的两个场景，我们分别有以下两种流程图：</p> <p>延迟消费</p> <p>延迟消费是延迟队列最为常用的使用模式。如下图所示，生产者产生的消息首先会进入缓冲队列（图中红色队列）。通过RabbitMQ提供的TTL扩展，这些消息会被设置过期时间，也就是延迟消费的时间。等消息过期之后，这些消息会通过配置好的DLX转发到实际消费队列（图中蓝色队列），以此达到延迟消费的效果。</p> <p><img src="https://simg.open-open.com/show/8a089b22046f705832ec26fb26f0e287.png"></p> <p>延迟重试</p> <p>延迟重试本质上也是延迟消费的一种，但是这种模式的结构与普通的延迟消费的流程图较为不同，所以单独拎出来介绍。</p> <p>如下图所示，消费者发现该消息处理出现了异常，比如是因为网络波动引起的异常。那么如果不等待一段时间，直接就重试的话，很可能会导致在这期间内一直无法成功，造成一定的资源浪费。那么我们可以将其先放在缓冲队列中（图中红色队列），等消息经过一段的延迟时间后再次进入实际消费队列中（图中蓝色队列），此时由于已经过了“较长”的时间了，异常的一些波动通常已经恢复，这些消息可以被正常地消费。</p> <p><img src="https://simg.open-open.com/show/f2cbe9216f90a22d434389abbc92652a.png"></p> <h2>代码实现</h2> <p>接下来我们将介绍如何在Spring Boot中实现基于RabbitMQ的延迟队列。我们假设读者已经拥有了Spring Boot与RabbitMQ的基本知识。如果想快速了解Spring Boot的相关基础知识，可以参考我之前写的一篇文章。</p> <h3>初始化工程</h3> <p>首先我们在Intellij中创建一个Spring Boot工程，并且添加 spring-boot-starter-amqp 扩展。</p> <h3>配置队列</h3> <p>从上述的流程图中我们可以看到，一个延迟队列的实现，需要一个缓冲队列以及一个实际的消费队列。又由于在RabbitMQ中，我们拥有两种消息过期的配置方式，所以在代码中，我们一共配置了三条队列：</p> <ul> <li>delay_queue_per_message_ttl：TTL配置在消息上的缓冲队列。</li> <li>delay_queue_per_queue_ttl：TTL配置在队列上的缓冲队列。</li> <li>delay_process_queue：实际消费队列。</li> </ul> <p>我们通过Java Config的方式将上述的队列配置为Bean。由于我们添加了 spring-boot-starter-amqp 扩展，Spring Boot在启动时会根据我们的配置自动创建这些队列。为了方便接下来的测试，我们将delay_queue_per_message_ttl以及delay_queue_per_queue_ttl的DLX配置为同一个，且过期的消息都会通过DLX转发到delay_process_queue。</p> <p>delay_queue_per_message_ttl</p> <p>首先介绍delay_queue_per_message_ttl的配置代码：</p> <pre> <code class="language-java">@Bean QueuedelayQueuePerMessageTTL(){ return QueueBuilder.durable(DELAY_QUEUE_PER_MESSAGE_TTL_NAME) .withArgument("x-dead-letter-exchange", DELAY_EXCHANGE_NAME) // DLX，dead letter发送到的exchange .withArgument("x-dead-letter-routing-key", DELAY_PROCESS_QUEUE_NAME) // dead letter携带的routing key .build(); } </code></pre> <p>其中， x-dead-letter-exchange 声明了队列里的死信转发到的DLX名称， x-dead-letter-routing-key 声明了这些死信在转发时携带的routing-key名称。</p> <p>delay_queue_per_queue_ttl</p> <p>类似地，delay_queue_per_queue_ttl的配置代码：</p> <pre> <code class="language-java">@Bean QueuedelayQueuePerQueueTTL(){ return QueueBuilder.durable(DELAY_QUEUE_PER_QUEUE_TTL_NAME) .withArgument("x-dead-letter-exchange", DELAY_EXCHANGE_NAME) // DLX .withArgument("x-dead-letter-routing-key", DELAY_PROCESS_QUEUE_NAME) // dead letter携带的routing key .withArgument("x-message-ttl", QUEUE_EXPIRATION) // 设置队列的过期时间 .build(); } </code></pre> <p>delay_queue_per_queue_ttl队列的配置比delay_queue_per_message_ttl队列的配置多了一个 x-message-ttl ，该配置用来设置队列的过期时间。</p> <p>delay_process_queue</p> <p>delay_process_queue的配置最为简单：</p> <pre> <code class="language-java">@Bean QueuedelayProcessQueue(){ return QueueBuilder.durable(DELAY_PROCESS_QUEUE_NAME) .build(); } </code></pre> <h3>配置Exchange</h3> <p>配置DLX</p> <p>首先，我们需要配置DLX，代码如下：</p> <pre> <code class="language-java">@Bean DirectExchangedelayExchange(){ return new DirectExchange(DELAY_EXCHANGE_NAME); } </code></pre> <p>然后再将该DLX绑定到实际消费队列即delay_process_queue上。这样所有的死信都会通过DLX被转发到delay_process_queue：</p> <pre> <code class="language-java">@Bean BindingdlxBinding(Queue delayProcessQueue, DirectExchange delayExchange){ return BindingBuilder.bind(delayProcessQueue) .to(delayExchange) .with(DELAY_PROCESS_QUEUE_NAME); } </code></pre> <p>配置延迟重试所需的Exchange</p> <p>从延迟重试的流程图中我们可以看到，消息处理失败之后，我们需要将消息转发到缓冲队列，所以缓冲队列也需要绑定一个Exchange。 在本例中，我们将delay_process_per_queue_ttl作为延迟重试里的缓冲队列 。具体代码是如何配置的，这里就不赘述了，大家可以查阅我 <a href="/misc/goto?guid=4959755495812036009" rel="nofollow,noindex">Github</a> 中的代码。</p> <h3>定义消费者</h3> <p>我们创建一个最简单的消费者ProcessReceiver，这个消费者监听delay_process_queue队列，对于接受到的消息，他会：</p> <ul> <li>如果消息里的消息体不等于FAIL_MESSAGE，那么他会输出消息体。</li> <li>如果消息里的消息体恰好是FAIL_MESSAGE，那么他会模拟抛出异常，然后将该消息重定向到缓冲队列（对应延迟重试场景）。</li> </ul> <p>另外，我们还需要新建一个监听容器用于存放消费者，代码如下：</p> <pre> <code class="language-java">@Bean SimpleMessageListenerContainerprocessContainer(ConnectionFactory connectionFactory, ProcessReceiver processReceiver){ SimpleMessageListenerContainer container = new SimpleMessageListenerContainer(); container.setConnectionFactory(connectionFactory); container.setQueueNames(DELAY_PROCESS_QUEUE_NAME); // 监听delay_process_queue container.setMessageListener(new MessageListenerAdapter(processReceiver)); return container; } </code></pre> <p>至此，我们前置的配置代码已经全部编写完成，接下来我们需要编写测试用例来测试我们的延迟队列。</p> <h3>编写测试用例</h3> <p>延迟消费场景</p> <p>首先我们编写用于测试TTL设置在消息上的测试代码。</p> <p>我们借助 spring-rabbit 包下提供的RabbitTemplate类来发送消息。由于我们添加了 spring-boot-starter-amqp 扩展，Spring Boot会在初始化时自动地将RabbitTemplate当成bean加载到容器中。</p> <p>解决了消息的发送问题，那么又该如何为每个消息设置TTL呢？这里我们需要借助MessagePostProcessor。MessagePostProcessor通常用来设置消息的Header以及消息的属性。我们新建一个ExpirationMessagePostProcessor类来负责设置消息的TTL属性：</p> <pre> <code class="language-java">/** * 设置消息的失效时间 */ public class ExpirationMessagePostProcessorimplements MessagePostProcessor{ private final Long ttl; // 毫秒 public ExpirationMessagePostProcessor(Long ttl){ this.ttl = ttl; } @Override public Message postProcessMessage(Message message)throws AmqpException { message.getMessageProperties() .setExpiration(ttl.toString()); // 设置per-message的失效时间 return message; } } </code></pre> <p>然后在调用RabbitTemplate的convertAndSend方法时，传入ExpirationMessagePostPorcessor即可。我们向缓冲队列中发送3条消息，过期时间依次为1秒，2秒和3秒。具体的代码如下所示：</p> <pre> <code class="language-java">@Test public void testDelayQueuePerMessageTTL()throws InterruptedException { ProcessReceiver.latch = new CountDownLatch(3); for (int i = 1; i <= 3; i++) { long expiration = i * 1000; rabbitTemplate.convertAndSend(QueueConfig.DELAY_QUEUE_PER_MESSAGE_TTL_NAME, (Object) ("Message From delay_queue_per_message_ttl with expiration " + expiration), new ExpirationMessagePostProcessor(expiration)); } ProcessReceiver.latch.await(); } </code></pre> <p>细心的朋友一定会问，为什么要在代码中加一个CountDownLatch呢？这是因为如果没有latch阻塞住测试方法的话，测试用例会直接结束，程序退出，我们就看不到消息被延迟消费的表现了。</p> <p>那么类似地，测试TTL设置在队列上的代码如下：</p> <pre> <code class="language-java">@Test public void testDelayQueuePerQueueTTL()throws InterruptedException { ProcessReceiver.latch = new CountDownLatch(3); for (int i = 1; i <= 3; i++) { rabbitTemplate.convertAndSend(QueueConfig.DELAY_QUEUE_PER_QUEUE_TTL_NAME, "Message From delay_queue_per_queue_ttl with expiration " + QueueConfig.QUEUE_EXPIRATION); } ProcessReceiver.latch.await(); } </code></pre> <p>我们向缓冲队列中发送3条消息。理论上这3条消息会在4秒后同时过期。</p> <p>延迟重试场景</p> <p>我们同样还需测试延迟重试场景。</p> <pre> <code class="language-java">@Test public void testFailMessage()throws InterruptedException { ProcessReceiver.latch = new CountDownLatch(6); for (int i = 1; i <= 3; i++) { rabbitTemplate.convertAndSend(QueueConfig.DELAY_PROCESS_QUEUE_NAME, ProcessReceiver.FAIL_MESSAGE); } ProcessReceiver.latch.await(); } </code></pre> <p>我们向delay_process_queue发送3条会触发FAIL的消息，理论上这3条消息会在4秒后自动重试。</p> <h3>查看测试结果</h3> <p>延迟消费场景</p> <p>延迟消费的场景测试我们分为了TTL设置在消息上和TTL设置在队列上两种。首先，我们先看一下TTL设置在消息上的测试结果：</p> <p><img src="https://simg.open-open.com/show/7cb87a144c089dd4232bc340f53b94e5.gif"></p> <p>从上图中我们可以看到，ProcessReceiver分别经过1秒、2秒、3秒收到消息。测试结果表明消息不仅被延迟消费了，而且每条消息的延迟时间是可以被个性化设置的。TTL设置在消息上的延迟消费场景测试成功。</p> <p>然后，TTL设置在队列上的测试结果如下图：</p> <p><img src="https://simg.open-open.com/show/0dbc2727481a08e0c0c8337b25481b55.gif"></p> <p>从上图中我们可以看到，ProcessReceiver经过了4秒的延迟之后，同时收到了3条消息。测试结果表明消息不仅被延迟消费了，同时也证明了当TTL设置在队列上的时候，消息的过期时间是固定的。TTL设置在队列上的延迟消费场景测试成功。</p> <p>延迟重试场景</p> <p>接下来，我们再来看一下延迟重试的测试结果：</p> <p><img src="https://simg.open-open.com/show/5b1e9434f4d598530a8d72705e7bfe0f.gif"></p> <p>ProcessReceiver首先收到了3条会触发FAIL的消息，然后将其移动到缓冲队列之后，过了4秒，又收到了刚才的那3条消息。延迟重试场景测试成功。</p> <h2>总结</h2> <p>本文首先介绍了延迟队列的概念以及用途，并且通过代码详细讲解了如何通过Spring Boot和RabbitMQ实现一个延迟队列。希望本文能够对大家平时的学习和工作能有所启发和帮助。</p> <p> </p> <p> </p> <p>来自：http://www.kissyu.org/2017/11/18/Spring Boot与RabbitMQ结合实现延迟队列/</p> <p> </p>