zoukankan      html  css  js  c++  java
  • RabbitMQ 延时消息队列

    一、简述

    二、示例demo

    、简述

    延时消息在日常随处可见:

    1、订单创建10min之后不发起支付,自动取消。

    2、30min定时推送一次邮件信息。

    最常用到方式为定时任务轮训,数据量小的时候使用没什么问题 而当有千万甚至上亿的数据量时就会出现数据读取的瓶颈,此时全表扫面进行处理一定是下下策。但是也有比较讨巧的方式,分享公司内部订单拆分的例子:

    由于线上每天订单量50万+的增长量,单表早已无法吃撑这个增长的速度。采取的方式为订单归档:线上热数据保留2-3天的数据,其余都归档进入历史订单表中,这样热数据在200万以内。
    订单超过10min不支付即取消的功能,可以采取简单的扫表形式而不会出现数据读取性能的问题。

     这样的方式很简单,但需要跟业务进行沟通妥协,本文会讲另一种方式即RabbitMQ延迟队列。RabbitMQ实际并没有直接实现延时队列,但可利用RabbitMQ提供的属性来模拟延时队列,甚至已经有的配套的插件rabbitmq_delayed_message_exchange 下面先介绍使用到的RabbitMQ的属性。

    1、消息的Time To Live (TTL) 

    x-message-ttl:消息过期时间,超过过期时间之后即变为死信(Dead-letter)不会再被消费者消费。

    设置消息TTL有两种方式:

    • 创建队列时指定x-message-ttl,此时队列所有的消息具有统一过期时间。
    • 发送消息为每个消息设置 expiration,此时消息之间过期时间不同。 

    如果两者都设置,过期时间取两者最小。如果设置TTL为0即表示除非立马能发送到队列,否则直接丢弃该消息。利用TTL为0的特性再结合死信转发器可以替代RabbitMQ 3.0的immediate参数。

    2、队列的TTL

    x-expires: RabbitMQ会确保时间达到后将队列删除,但是并不保障这个动作有多及时。队列过期代表着处于未使用状态,即

    • 队列无任何消费者
    • 队列没有被重新声明
    • 队列在过期未调用Basic.Get命令获取消息

    3、x-dead-letter-exchange(RabbitMQ文档):死信转发器(转发器类型)当消息达到过期时间未被消费则会由该exchange按照配置的x-dead-letter-routing-key转发到指定队列,最后被消费者消费,如果未配置x-dead-letter-routing-key则会按照原队列的key进行转发。

    4、队列的消息在以下几种情况会变成死信(Dead-letter)

    • 设置的x-message-ttl或者expiration到期,即消息过期
    • 消息被消费者拒绝(调用Basic.Reject / Basic.Nack)且 requeue参数设置为false
    • 队列达到最大长度

    二、示例demo

    • 单个延迟队列

    RabbitMQ延时队列逻辑:

      

    1、exchange_delay_begin:缓冲队列exchange交换器,用于将消息转发至缓存消息队列 queue_delay_begin 。

    2、exchange_delay_done:死信(dead-letter)队列exchange交换器,用于将队列 queue_delay_begin 转发到死信队列。

    3、queue_delay_begin:缓冲消息队列,等待消息过期。

    4、queue_delay_done:死信消息队列,消费者能够真正消费信息。

     spring-rabbitmq.xml :

    <?xml version="1.0" encoding="UTF-8"?>
    <beans xmlns="http://www.springframework.org/schema/beans"
           xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xmlns:tx="http://www.springframework.org/schema/tx"
           xmlns:util="http://www.springframework.org/schema/util" xmlns:context="http://www.springframework.org/schema/context"
           xmlns:rabbit="http://www.springframework.org/schema/rabbit"
           xsi:schemaLocation="http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans-3.1.xsd
                                http://www.springframework.org/schema/context
                                http://www.springframework.org/schema/context/spring-context-3.1.xsd
                                http://www.springframework.org/schema/tx
                                http://www.springframework.org/schema/tx/spring-tx.xsd
                                http://www.springframework.org/schema/aop
                                http://www.springframework.org/schema/aop/spring-aop.xsd
                                http://www.springframework.org/schema/util http://www.springframework.org/schema/util/spring-util-3.1.xsd
                                http://www.springframework.org/schema/rabbit http://www.springframework.org/schema/rabbit/spring-rabbit-1.6.xsd">
        <!--配置connection-factory,指定连接rabbit server参数 -->
        <rabbit:connection-factory id="connectionFactory"  username="guest" password="guest" host="127.0.0.1" port="5672" publisher-confirms="true"/>
        <!-- 延时队列 -->
        <rabbit:direct-exchange id="exchange_delay_begin" name="exchange_delay_begin"  durable="false" auto-delete="false" >
            <rabbit:bindings>
                <rabbit:binding queue="queue_delay_begin" key="delay" />
            </rabbit:bindings>
        </rabbit:direct-exchange>
    
        <rabbit:queue name="queue_delay_begin" durable="false">
            <rabbit:queue-arguments>
                <!--  队列过期时间 -->
                <entry key="x-message-ttl" value="30000" value-type="java.lang.Long" />
                <entry key="x-dead-letter-exchange" value="exchange_delay_done" />
                <entry key="x-dead-letter-routing-key" value="delay" />
            </rabbit:queue-arguments>
        </rabbit:queue>
    
        <rabbit:direct-exchange id="exchange_delay_done" name="exchange_delay_done"  durable="false" auto-delete="false" >
            <rabbit:bindings>
                <rabbit:binding queue="queue_delay_done" key="delay" />
                <!--  binding key 相同为 【delay】exchange转发消息到多个队列 -->
                <!--<rabbit:binding queue="queue_delay_done_two" key="delay" />-->
            </rabbit:bindings>
        </rabbit:direct-exchange>
    
        <rabbit:queue name="queue_delay_done" durable="false"/>
        <rabbit:template id="delayMsgTwoTemplate" connection-factory="connectionFactory" />
        <bean id="messageConsumer" class="com.nancy.rabbitmq.demo.MessageConsumer"></bean>
    
        <!-- 消息接收者 -->
        <rabbit:listener-container connection-factory="connectionFactory" channel-transacted="false" >
            <rabbit:listener queues="queue_delay_done" ref="messageConsumer" />
        </rabbit:listener-container>
    </beans>

     DelayMessageProducer.java

    @Service
    public class DelayMessageProducer {
        @Resource(name="delayMsgTwoTemplate")
        private AmqpTemplate delayMsgTwoTemplate;
        public void delayMsgTwo(String exchange, String routingKey, Object msg) {
            delayMsgTwoTemplate.convertAndSend(exchange, routingKey, msg, new MessagePostProcessor() {
                @Override
                public Message postProcessMessage(Message message) throws AmqpException {
                    message.getMessageProperties().setExpiration(String.valueOf(10000));
                    return message;
                }
            });
        }
    }

     MessageConsumer.java

    public class MessageConsumer implements MessageListener {
        @Override
        public void onMessage(Message message) {
            System.out.println("consumer receive message 22------->:{}"+ message);
        }
    }

     application.xml 

    <?xml version="1.0" encoding="UTF-8"?>
    <beans xmlns="http://www.springframework.org/schema/beans"
           xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xmlns:context="http://www.springframework.org/schema/context"
           xsi:schemaLocation="http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans-3.1.xsd
        http://www.springframework.org/schema/context http://www.springframework.org/schema/context/spring-context-3.1.xsd">
        <import resource="spring-rabbitmq.xml" />
        <!-- 扫描指定package注释的注册为Spring Beans -->
        <context:component-scan base-package="com.nancy.rabbitmq" />
        <!-- 激活annotation功能 -->
        <context:annotation-config />
        <!-- 激活annotation功能 -->
        <context:spring-configured />
    </beans>

    DelayQueueTest.java

    public class DelayQueueTest {
        private ApplicationContext context = null;
        @org.junit.Before
        public void setUp() throws Exception {
            context = new ClassPathXmlApplicationContext("rabbitmq/application.xml");
        }
        @Test
        public void delayQueueTest() throws Exception {
            DelayMessageProducer messageProducer = context.getBean(DelayMessageProducer.class);
            int a = 10;
            while (a > 0) {
                System.out.println("send "+ a);
                messageProducer.delayMsgTwo("exchange_delay_begin","delay", "hello world delay2 :" + a--);
                try {
                    Thread.sleep(10);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
            System.out.println("sended ");
            Thread.sleep(1000*60);
        }
    }

     运行结果: 发送消息 10s之后, 消费监听到消息 消费。

    send 10
    send 9
    send 8
    send 7
    send 6
    send 5
    send 4
    send 3
    send 2
    send 1
    sended 
    consumer receive message 22------->:{}(Body:'hello world delay2 :10' MessageProperties [headers={x-first-death-exchange=exchange_delay_begin, x-death=[{reason=expired, original-expiration=10000, count=1, exchange=exchange_delay_begin, time=Sun Jul 01 18:33:19 CST 2018, routing-keys=[delay], queue=queue_delay_begin}], x-first-death-reason=expired, x-first-death-queue=queue_delay_begin}, timestamp=null, messageId=null, userId=null, receivedUserId=null, appId=null, clusterId=null, type=null, correlationId=null, correlationIdString=null, replyTo=null, contentType=text/plain, contentEncoding=UTF-8, contentLength=0, deliveryMode=null, receivedDeliveryMode=PERSISTENT, expiration=null, priority=0, redelivered=false, receivedExchange=exchange_delay_done, receivedRoutingKey=delay, receivedDelay=1000, deliveryTag=1, messageCount=0, consumerTag=amq.ctag-oOJzeYHEJYrLFL4HbHWZcA, consumerQueue=queue_delay_done])
    consumer receive message 22------->:{}(Body:'hello world delay2 :9' MessageProperties [headers={x-first-death-exchange=exchange_delay_begin, x-death=[{reason=expired, original-expiration=10000, count=1, exchange=exchange_delay_begin, time=Sun Jul 01 18:33:19 CST 2018, routing-keys=[delay], queue=queue_delay_begin}], x-first-death-reason=expired, x-first-death-queue=queue_delay_begin}, timestamp=null, messageId=null, userId=null, receivedUserId=null, appId=null, clusterId=null, type=null, correlationId=null, correlationIdString=null, replyTo=null, contentType=text/plain, contentEncoding=UTF-8, contentLength=0, deliveryMode=null, receivedDeliveryMode=PERSISTENT, expiration=null, priority=0, redelivered=false, receivedExchange=exchange_delay_done, receivedRoutingKey=delay, receivedDelay=1000, deliveryTag=2, messageCount=0, consumerTag=amq.ctag-oOJzeYHEJYrLFL4HbHWZcA, consumerQueue=queue_delay_done])
    consumer receive message 22------->:{}(Body:'hello world delay2 :8' MessageProperties [headers={x-first-death-exchange=exchange_delay_begin, x-death=[{reason=expired, original-expiration=10000, count=1, exchange=exchange_delay_begin, time=Sun Jul 01 18:33:19 CST 2018, routing-keys=[delay], queue=queue_delay_begin}], x-first-death-reason=expired, x-first-death-queue=queue_delay_begin}, timestamp=null, messageId=null, userId=null, receivedUserId=null, appId=null, clusterId=null, type=null, correlationId=null, correlationIdString=null, replyTo=null, contentType=text/plain, contentEncoding=UTF-8, contentLength=0, deliveryMode=null, receivedDeliveryMode=PERSISTENT, expiration=null, priority=0, redelivered=false, receivedExchange=exchange_delay_done, receivedRoutingKey=delay, receivedDelay=1000, deliveryTag=3, messageCount=0, consumerTag=amq.ctag-oOJzeYHEJYrLFL4HbHWZcA, consumerQueue=queue_delay_done])
    consumer receive message 22------->:{}(Body:'hello world delay2 :7' MessageProperties [headers={x-first-death-exchange=exchange_delay_begin, x-death=[{reason=expired, original-expiration=10000, count=1, exchange=exchange_delay_begin, time=Sun Jul 01 18:33:19 CST 2018, routing-keys=[delay], queue=queue_delay_begin}], x-first-death-reason=expired, x-first-death-queue=queue_delay_begin}, timestamp=null, messageId=null, userId=null, receivedUserId=null, appId=null, clusterId=null, type=null, correlationId=null, correlationIdString=null, replyTo=null, contentType=text/plain, contentEncoding=UTF-8, contentLength=0, deliveryMode=null, receivedDeliveryMode=PERSISTENT, expiration=null, priority=0, redelivered=false, receivedExchange=exchange_delay_done, receivedRoutingKey=delay, receivedDelay=1000, deliveryTag=4, messageCount=0, consumerTag=amq.ctag-oOJzeYHEJYrLFL4HbHWZcA, consumerQueue=queue_delay_done])
    consumer receive message 22------->:{}(Body:'hello world delay2 :6' MessageProperties [headers={x-first-death-exchange=exchange_delay_begin, x-death=[{reason=expired, original-expiration=10000, count=1, exchange=exchange_delay_begin, time=Sun Jul 01 18:33:19 CST 2018, routing-keys=[delay], queue=queue_delay_begin}], x-first-death-reason=expired, x-first-death-queue=queue_delay_begin}, timestamp=null, messageId=null, userId=null, receivedUserId=null, appId=null, clusterId=null, type=null, correlationId=null, correlationIdString=null, replyTo=null, contentType=text/plain, contentEncoding=UTF-8, contentLength=0, deliveryMode=null, receivedDeliveryMode=PERSISTENT, expiration=null, priority=0, redelivered=false, receivedExchange=exchange_delay_done, receivedRoutingKey=delay, receivedDelay=1000, deliveryTag=5, messageCount=0, consumerTag=amq.ctag-oOJzeYHEJYrLFL4HbHWZcA, consumerQueue=queue_delay_done])
    consumer receive message 22------->:{}(Body:'hello world delay2 :5' MessageProperties [headers={x-first-death-exchange=exchange_delay_begin, x-death=[{reason=expired, original-expiration=10000, count=1, exchange=exchange_delay_begin, time=Sun Jul 01 18:33:19 CST 2018, routing-keys=[delay], queue=queue_delay_begin}], x-first-death-reason=expired, x-first-death-queue=queue_delay_begin}, timestamp=null, messageId=null, userId=null, receivedUserId=null, appId=null, clusterId=null, type=null, correlationId=null, correlationIdString=null, replyTo=null, contentType=text/plain, contentEncoding=UTF-8, contentLength=0, deliveryMode=null, receivedDeliveryMode=PERSISTENT, expiration=null, priority=0, redelivered=false, receivedExchange=exchange_delay_done, receivedRoutingKey=delay, receivedDelay=1000, deliveryTag=6, messageCount=0, consumerTag=amq.ctag-oOJzeYHEJYrLFL4HbHWZcA, consumerQueue=queue_delay_done])
    consumer receive message 22------->:{}(Body:'hello world delay2 :4' MessageProperties [headers={x-first-death-exchange=exchange_delay_begin, x-death=[{reason=expired, original-expiration=10000, count=1, exchange=exchange_delay_begin, time=Sun Jul 01 18:33:19 CST 2018, routing-keys=[delay], queue=queue_delay_begin}], x-first-death-reason=expired, x-first-death-queue=queue_delay_begin}, timestamp=null, messageId=null, userId=null, receivedUserId=null, appId=null, clusterId=null, type=null, correlationId=null, correlationIdString=null, replyTo=null, contentType=text/plain, contentEncoding=UTF-8, contentLength=0, deliveryMode=null, receivedDeliveryMode=PERSISTENT, expiration=null, priority=0, redelivered=false, receivedExchange=exchange_delay_done, receivedRoutingKey=delay, receivedDelay=1000, deliveryTag=7, messageCount=0, consumerTag=amq.ctag-oOJzeYHEJYrLFL4HbHWZcA, consumerQueue=queue_delay_done])
    consumer receive message 22------->:{}(Body:'hello world delay2 :3' MessageProperties [headers={x-first-death-exchange=exchange_delay_begin, x-death=[{reason=expired, original-expiration=10000, count=1, exchange=exchange_delay_begin, time=Sun Jul 01 18:33:19 CST 2018, routing-keys=[delay], queue=queue_delay_begin}], x-first-death-reason=expired, x-first-death-queue=queue_delay_begin}, timestamp=null, messageId=null, userId=null, receivedUserId=null, appId=null, clusterId=null, type=null, correlationId=null, correlationIdString=null, replyTo=null, contentType=text/plain, contentEncoding=UTF-8, contentLength=0, deliveryMode=null, receivedDeliveryMode=PERSISTENT, expiration=null, priority=0, redelivered=false, receivedExchange=exchange_delay_done, receivedRoutingKey=delay, receivedDelay=1000, deliveryTag=8, messageCount=0, consumerTag=amq.ctag-oOJzeYHEJYrLFL4HbHWZcA, consumerQueue=queue_delay_done])
    consumer receive message 22------->:{}(Body:'hello world delay2 :2' MessageProperties [headers={x-first-death-exchange=exchange_delay_begin, x-death=[{reason=expired, original-expiration=10000, count=1, exchange=exchange_delay_begin, time=Sun Jul 01 18:33:19 CST 2018, routing-keys=[delay], queue=queue_delay_begin}], x-first-death-reason=expired, x-first-death-queue=queue_delay_begin}, timestamp=null, messageId=null, userId=null, receivedUserId=null, appId=null, clusterId=null, type=null, correlationId=null, correlationIdString=null, replyTo=null, contentType=text/plain, contentEncoding=UTF-8, contentLength=0, deliveryMode=null, receivedDeliveryMode=PERSISTENT, expiration=null, priority=0, redelivered=false, receivedExchange=exchange_delay_done, receivedRoutingKey=delay, receivedDelay=1000, deliveryTag=9, messageCount=0, consumerTag=amq.ctag-oOJzeYHEJYrLFL4HbHWZcA, consumerQueue=queue_delay_done])
    consumer receive message 22------->:{}(Body:'hello world delay2 :1' MessageProperties [headers={x-first-death-exchange=exchange_delay_begin, x-death=[{reason=expired, original-expiration=10000, count=1, exchange=exchange_delay_begin, time=Sun Jul 01 18:33:19 CST 2018, routing-keys=[delay], queue=queue_delay_begin}], x-first-death-reason=expired, x-first-death-queue=queue_delay_begin}, timestamp=null, messageId=null, userId=null, receivedUserId=null, appId=null, clusterId=null, type=null, correlationId=null, correlationIdString=null, replyTo=null, contentType=text/plain, contentEncoding=UTF-8, contentLength=0, deliveryMode=null, receivedDeliveryMode=PERSISTENT, expiration=null, priority=0, redelivered=false, receivedExchange=exchange_delay_done, receivedRoutingKey=delay, receivedDelay=1000, deliveryTag=10, messageCount=0, consumerTag=amq.ctag-oOJzeYHEJYrLFL4HbHWZcA, consumerQueue=queue_delay_done])
    View Code

      

    • 多个延迟队列

    实际的业务需求中会出现不同的时间延迟,此时可设置多个队列以达到不同的延迟效果。例如5个队列 common-queue_5s、common-queue_15s、common-queue_30s、common-queue_45s、common-queue_50s达到不同的延迟效果,整体的结构如下:

     上述bindingKey的值有所简化,但对路由结构图无影响。这里使用一个死信转发器(转发器类型)通过绑定不同的key路由到不同的死信队列。也可以死信转发器和死信队列一对一绑定 即 成对出现(dlx_exchange_5s 和 dead-letter-queue_5s)

    这里贴出部分xml部分配置:

    <?xml version="1.0" encoding="UTF-8"?>
    <beans xmlns="http://www.springframework.org/schema/beans"
        xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xmlns:rabbit="http://www.springframework.org/schema/rabbit"
        xsi:schemaLocation="http://www.springframework.org/schema/beans 
        http://www.springframework.org/schema/beans/spring-beans.xsd
        http://www.springframework.org/schema/rabbit
        http://www.springframework.org/schema/rabbit/spring-rabbit-1.4.xsd">
    
        <rabbit:connection-factory
                id="connectionFactory"
                host="${rabbit.host}"
                port="${rabbit.port}"
                username="${rabbit.username}"
                password="${rabbit.password}"
                publisher-confirms="true"
        />
    
        <rabbit:admin connection-factory="connectionFactory" ignore-declaration-exceptions="true" />
    
        <!-- 正常队列 -->
        <!-- 5s过期 -->
        <rabbit:queue name="common-queue_5s">
            <rabbit:queue-arguments>
                <entry key="x-message-ttl" value="5000" value-type="java.lang.Long" />
                <entry key="x-dead-letter-exchange" value="dlx-exchange" />
                <entry key="x-dead-letter-routing-key" value="dead-letter-queue_5s" />
            </rabbit:queue-arguments>
        </rabbit:queue>
        <!-- 15s过期 -->
        <rabbit:queue name="common-queue_15s">
            <rabbit:queue-arguments>
                <entry key="x-message-ttl" value="15000" value-type="java.lang.Long" />
                <entry key="x-dead-letter-exchange" value="dlx-exchange" />
                <entry key="x-dead-letter-routing-key" value="dead-letter-queue_15s" />
            </rabbit:queue-arguments>
        </rabbit:queue>
        <!-- 30s过期 -->
        <rabbit:queue name="common-queue_30s">
            <rabbit:queue-arguments>
                <entry key="x-message-ttl" value="30000" value-type="java.lang.Long" />
                <entry key="x-dead-letter-exchange" value="dlx-exchange" />
                <entry key="x-dead-letter-routing-key" value="dead-letter-queue_30s" />
            </rabbit:queue-arguments>
        </rabbit:queue>
        <!-- 45s过期 -->
        <rabbit:queue name="common-queue_45s">
            <rabbit:queue-arguments>
                <entry key="x-message-ttl" value="45000" value-type="java.lang.Long" />
                <entry key="x-dead-letter-exchange" value="dlx-exchange" />
                <entry key="x-dead-letter-routing-key" value="dead-letter-queue_45s" />
            </rabbit:queue-arguments>
        </rabbit:queue>
        <!-- 50s过期 -->
        <rabbit:queue name="common-queue_50s">
            <rabbit:queue-arguments>
                <entry key="x-message-ttl" value="50000" value-type="java.lang.Long" />
                <entry key="x-dead-letter-exchange" value="dlx-exchange" />
                <entry key="x-dead-letter-routing-key" value="dead-letter-queue_50s" />
            </rabbit:queue-arguments>
        </rabbit:queue>
        <!-- 正常路由 -->
        <rabbit:direct-exchange name="common-exchange" durable="false" id="common-exchange">
            <rabbit:bindings>
                <rabbit:binding queue="common-queue_5s" />
                <rabbit:binding queue="common-queue_15s" />
                <rabbit:binding queue="common-queue_30s" />
                <rabbit:binding queue="common-queue_45s" />
                <rabbit:binding queue="common-queue_50s" />
            </rabbit:bindings>
        </rabbit:direct-exchange>
    
        <!-- 死信队列 -->
        <rabbit:queue name="dead-letter-queue_5s" />
        <rabbit:queue name="dead-letter-queue_15s" />
        <rabbit:queue name="dead-letter-queue_30s" />
        <rabbit:queue name="dead-letter-queue_45s" />
        <rabbit:queue name="dead-letter-queue_50s" />
        <rabbit:direct-exchange name="dlx-exchange" durable="false" id="dlx-exchange">
            <rabbit:bindings>
                <rabbit:binding queue="dead-letter-queue_5s" />
                <rabbit:binding queue="dead-letter-queue_15s" />
                <rabbit:binding queue="dead-letter-queue_30s" />
                <rabbit:binding queue="dead-letter-queue_45s" />
                <rabbit:binding queue="dead-letter-queue_50s" />
            </rabbit:bindings>
        </rabbit:direct-exchange>
        
        <!-- 配置consumer, 监听的类和queue的对应关系 -->
        <rabbit:listener-container
            connection-factory="connectionFactory" acknowledge="manual" >
            <rabbit:listener queues="dead-letter-queue_5s" ref="receiveConfirmTestListener" />
            <rabbit:listener queues="dead-letter-queue_15s" ref="receiveConfirmTestListener" />
            <rabbit:listener queues="dead-letter-queue_30s" ref="receiveConfirmTestListener" />
            <rabbit:listener queues="dead-letter-queue_45s" ref="receiveConfirmTestListener" />
            <rabbit:listener queues="dead-letter-queue_50s" ref="receiveConfirmTestListener" />
        </rabbit:listener-container>
    
    </beans>
    View Code

     junit测试:

    @RunWith(SpringJUnit4ClassRunner.class)  
    @ContextConfiguration(locations = {"classpath:application-context.xml"})  
    public class TestDeadLetter {  
        @Autowired  
        private DeadLetterPublishService publishService;  
    
        @Test
        public void testALL() throws InterruptedException{
            String message = "currentTime:" + System.currentTimeMillis();
            System.out.println("test1---message: "+ message);
            publishService.send("common-exchange","common-queue_5s", message);
            publishService.send("common-exchange","common-queue_15s", message);
            publishService.send("common-exchange","common-queue_30s", message);
            publishService.send("common-exchange","common-queue_45s", message);
            publishService.send("common-exchange","common-queue_50s", message);
            Thread.sleep(100000);
        }
    
    } 
    TestDeadLetter.java

     最后运行结果:消息实际发送时间点 和 消息被延迟消费时间点无限接近 五个消息分别延迟大约 5s 15s 30s 45s 50s  但做不到精确一致。

    test1---message: currentTime:1566920053524
    // 。。。。
    1566920058551 - consumer--:MessageProperties [headers={spring_return_correlation=1ac8ee8e-6d61-4bbf-bac1-f21523a0d759, x-first-death-exchange=common-exchange, x-death=[{reason=expired, count=1, exchange=common-exchange, time=Tue Aug 27 23:34:18 CST 2019, routing-keys=[common-queue_5s], queue=common-queue_5s}], x-first-death-reason=expired, x-first-death-queue=common-queue_5s}, timestamp=null, messageId=null, userId=null, appId=null, clusterId=null, type=null, correlationId=null, replyTo=null, contentType=text/plain, contentEncoding=UTF-8, contentLength=0, deliveryMode=PERSISTENT, expiration=null, priority=0, redelivered=false, receivedExchange=dlx-exchange, receivedRoutingKey=dead-letter-queue_5s, deliveryTag=1, messageCount=0]:currentTime:1566920053524
    1566920068578 - consumer--:MessageProperties [headers={spring_return_correlation=1ac8ee8e-6d61-4bbf-bac1-f21523a0d759, x-first-death-exchange=common-exchange, x-death=[{reason=expired, count=1, exchange=common-exchange, time=Tue Aug 27 23:34:28 CST 2019, routing-keys=[common-queue_15s], queue=common-queue_15s}], x-first-death-reason=expired, x-first-death-queue=common-queue_15s}, timestamp=null, messageId=null, userId=null, appId=null, clusterId=null, type=null, correlationId=null, replyTo=null, contentType=text/plain, contentEncoding=UTF-8, contentLength=0, deliveryMode=PERSISTENT, expiration=null, priority=0, redelivered=false, receivedExchange=dlx-exchange, receivedRoutingKey=dead-letter-queue_15s, deliveryTag=1, messageCount=0]:currentTime:1566920053524
    1566920083550 - consumer--:MessageProperties [headers={spring_return_correlation=1ac8ee8e-6d61-4bbf-bac1-f21523a0d759, x-first-death-exchange=common-exchange, x-death=[{reason=expired, count=1, exchange=common-exchange, time=Tue Aug 27 23:34:43 CST 2019, routing-keys=[common-queue_30s], queue=common-queue_30s}], x-first-death-reason=expired, x-first-death-queue=common-queue_30s}, timestamp=null, messageId=null, userId=null, appId=null, clusterId=null, type=null, correlationId=null, replyTo=null, contentType=text/plain, contentEncoding=UTF-8, contentLength=0, deliveryMode=PERSISTENT, expiration=null, priority=0, redelivered=false, receivedExchange=dlx-exchange, receivedRoutingKey=dead-letter-queue_30s, deliveryTag=1, messageCount=0]:currentTime:1566920053524
    1566920098549 - consumer--:MessageProperties [headers={spring_return_correlation=1ac8ee8e-6d61-4bbf-bac1-f21523a0d759, x-first-death-exchange=common-exchange, x-death=[{reason=expired, count=1, exchange=common-exchange, time=Tue Aug 27 23:34:58 CST 2019, routing-keys=[common-queue_45s], queue=common-queue_45s}], x-first-death-reason=expired, x-first-death-queue=common-queue_45s}, timestamp=null, messageId=null, userId=null, appId=null, clusterId=null, type=null, correlationId=null, replyTo=null, contentType=text/plain, contentEncoding=UTF-8, contentLength=0, deliveryMode=PERSISTENT, expiration=null, priority=0, redelivered=false, receivedExchange=dlx-exchange, receivedRoutingKey=dead-letter-queue_45s, deliveryTag=1, messageCount=0]:currentTime:1566920053524
    1566920103551 - consumer--:MessageProperties [headers={spring_return_correlation=1ac8ee8e-6d61-4bbf-bac1-f21523a0d759, x-first-death-exchange=common-exchange, x-death=[{reason=expired, count=1, exchange=common-exchange, time=Tue Aug 27 23:35:03 CST 2019, routing-keys=[common-queue_50s], queue=common-queue_50s}], x-first-death-reason=expired, x-first-death-queue=common-queue_50s}, timestamp=null, messageId=null, userId=null, appId=null, clusterId=null, type=null, correlationId=null, replyTo=null, contentType=text/plain, contentEncoding=UTF-8, contentLength=0, deliveryMode=PERSISTENT, expiration=null, priority=0, redelivered=false, receivedExchange=dlx-exchange, receivedRoutingKey=dead-letter-queue_50s, deliveryTag=1, messageCount=0]:currentTime:1566920053524

      

    参考资料:

    DLX: https://www.rabbitmq.com/dlx.html

    RabbitMQ如何实现延迟队列?:https://blog.csdn.net/u013256816/article/details/55106401

    消息队列之 RabbitMQ :https://www.jianshu.com/p/79ca08116d57

    使用RabbitMQ实现延迟任务 : https://www.cnblogs.com/haoxinyue/p/6613706.html

  • 相关阅读:
    [XNA]2D图形概要(2D Graphics Overview)
    [WP7]WindowsPhone支持VS2010的开发工具出来了
    [WM]用双缓冲在CStatic上面画
    [读书]至理名言摘自你的灯还亮着吗
    无题!!
    Windows 8 Consumer Preview 中的快捷键
    aptana studio 汉化与安装 zencoding、配置
    jQuery常用焦点图,可做选项卡切换
    jQuery 导航点击变换样式
    原生JS:焦点图 左右滚动
  • 原文地址:https://www.cnblogs.com/xiaoxing/p/9250823.html
Copyright © 2011-2022 走看看