Go RabbitMQ 死信消息队列

Posted fengchuiyizh

tags:

篇首语:本文由小常识网(cha138.com)小编为大家整理,主要介绍了Go RabbitMQ 死信消息队列相关的知识,希望对你有一定的参考价值。

实现原理:

   /**

 (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)." 只有当过期的消息到了队列的顶端(队首),才会被真正的丢弃或者进入死信队列。

 

 

 

以上是关于Go RabbitMQ 死信消息队列的主要内容,如果未能解决你的问题,请参考以下文章

RabbitMQ实战-死信队列

rabbitmq死信队列及延迟队列

rabbitMQ 死信 延时队列 [go版本]

RabbitMQ 死信队列

使用RabbitMQ处理死信队列

利用rabbitMq的死信队列实现延时消息