Go RabbitMQ 死信消息队列

Posted 2021-03-10 fengchuiyizh

实现原理：

   /**

 （1）创建一个正常的队列 Q1,目的是处理业务逻辑，比如发送订单消息等 ，对应交换器和绑定键  分别为  E1 和  Bingkey1

 （2）创建一个延时消息队列 Q2，设定队列的延时时间为10s，对应的交换器和绑定键分别为 E2和Bingkey2;并在该队列创建时候，设定队列的  （a）超时时间 （b） 超时后跳转的 路由E1和绑定Bingkey1，即超时后跳到     队列Q1上

 （3） 将消息先发送到 队列Q2 上，然后等着队列超时，执行逻辑

 

 

 

 

 * 主要测试一个死信队列，功能主要实现延时消费，原理是先把消息发到正常队列，

 * 正常队列有超时时间，当达到时间后自动发到死信队列，然后由消费者去消费死信队列里的消息. */

 

延迟队列的应用场景

1、未支付订单定时取消
2、定时清理缓存对象、空闲连接等
3、下单成功后30分钟内，按不同时间间隔发送通知等（1min、3min、10min发一次）

1、设置队列的过期时间

$this->channel->queue_declare(
            $this->retry_queue(),
            false,
            true,
            false,
            false,
            false,
            new AMQPTable(
                [
                    # 不设置x-dead-letter-routing-key，使用原先的routing_key，10s过期后自动重回原先的队列里面，那x-dead-letter-exchange交换机就需绑定原先队列
                    ‘x-dead-letter-exchange‘ => $this->retry_exchange(),
                    
                    # 10s
                    ‘x-message-ttl‘ => 10000,
                ]
            )
        );

推送到该队列的所有消息（不设ttl），10s之后都会过期，根据原来的routing_key，进入到指定的exchange，进而进到指定队列。

 

2、设置消息的过期时间

$message = new AMQPMessage(
    ‘msg‘,
    array(
        # 消息持久化
        ‘delivery_mode‘ => AMQPMessage::DELIVERY_MODE_PERSITENT,
        
        # ttl过期时间
        ‘expiration‘ => 50000,
    )
);

每个消息都设置相同的过期时间，到期后消息就会失效。

3、同时设置队列、消息的过期时间

如果同时设置，则消息的过期时间会取决于较小的值，比如队列的‘x-message-ttl’设置为10s，消息的‘expiration’设置为50s，则10s之后这个消息就会失效。

 

4、后续

单单设置队列的ttl，或者单单设置相同的消息过期时间，死信队列是能正常工作的。但是设置不同的消息过期时间，就可能无法正常使用死信队列了。

队列不设ttl

$this->channel->queue_declare(
            $this->retry_queue(),
            false,
            true,
            false,
            false,
            false,
            new AMQPTable(
                [
                    # 不设置x-dead-letter-routing-key，使用原先的routing_key，10s过期后自动重回原先的队列里面，那x-dead-letter-exchange交换机就需绑定原先队列
                    ‘x-dead-letter-exchange‘ => $this->retry_exchange(),
                ]
            )
        );

第一个消息设置500s过期，先推进队列

 

 

第一个消息设置500s过期，先推进队列

$message = new AMQPMessage(
    ‘msg‘,
    array(
        # 消息持久化
        ‘delivery_mode‘ => AMQPMessage::DELIVERY_MODE_PERSITENT,
        
        # ttl过期时间
        ‘expiration‘ => 500000,
    )
);

第二个消息设置5s过期，后推进队列

$message = new AMQPMessage(
    ‘msg‘,
    array(
        # 消息持久化
        ‘delivery_mode‘ => AMQPMessage::DELIVERY_MODE_PERSITENT,
        
        # ttl过期时间
        ‘expiration‘ => 5000,
    )
);

结果发现，5s之后，队列里还存在2个消息。说明第二个消息并没有“真的过期失效”。原因是位于队列首部的消息没有过期。而rabbitmq的死信队列，是基于首部消息实现的。

5、结论

当MQ检查队列中的第一个消息时，发现其并未过期，则不会继续检查之后的消息了。即使之后的消息过期了，也会因为没在队列头部而无法流转到其他队列，这是MQ队列的特性决定的。你不能去消费队列中间的消息，队列必须先进先出。

对于设置队列TTL属性的方法，一旦消息过期，就会从队列中抹去，而设置消息头部属性，即使消息过期，也不会马上从队列中抹去，因为每条消息是否过期时在即将投递到消费者之前判定的，为什么两者得处理方法不一致？因为第一种方法里，队列中已过期的消息肯定在队列头部，RabbitMQ只要定期从队头开始扫描是否有过期消息即可，而第二种方法里，每条消息的过期时间不同，如果要删除所有过期消息，势必要扫描整个队列，所以不如等到此消息即将被消费时再判定是否过期，如果过期，再进行删除。

官方的叙述

"Only when expired messages reach the head of a queue will they actually be discarded (or dead-lettered)." 只有当过期的消息到了队列的顶端（队首），才会被真正的丢弃或者进入死信队列。