新代数控系统密码

新代数控系统密码通过上面两点,正常队列的消息始终不会被消费,而是等待消息TTL到期,进入死信队列,让死信消费者进行消费,从而达到延迟队列的效果。

上面看上去似乎没什么问题,实测一下就会发现消息不会“如期死亡”。

当先生产一个TTL为60s的消息,再生产一个TTL为5s的消息,第二个消息并不会再5s后过期进入死信队列,而是需要等到*个消息TTL到期后,与*个消息一同进入死信队列。这是因为RabbitMQ 只会判断队列中的*个消息是否过期。

通过插件实现延迟队列

架构

对于上文的问题,自然有解决方法,那就是通过 RabbitMQ 的 rabbitmq_delayed_message_exchange 插件来解决。本文不赘述 RabbitMQ和插件的安装,你可以参考此文安装或使用Docker来安装。

此插件的原理是将消息在交换机处暂存储在mnesia(一个分布式数据系统)表中,延迟投递到队列中,等到消息到期再投递到队列当中。

简单了解了插件的原理,我们便可以如此设计延迟队列。

实现

生产者实现的关键点:

1.在声明交换机时不在是direct类型,而是x-delayed-message类型,这是由插件提供的类型;

2.交换机要增加"x-delayed-type": "direct"参数设置;

3.发布消息时,要在 Headers 中设置x-delay参数,来控制消息从交换机过期时间;

err = mqCh.Publish(constant.Exchange1, constant.RoutingKey1, false, false, amqp.Publishing{
   ContentType: "text/plain",
   Body:        []byte(message),    //Expiration: "10000", // 消息过期时间(消息级别),毫秒
   Headers: map[string]interface{}{        "x-delay": "5000", // 消息从交换机过期时间,毫秒(x-dead-message插件提供)
   },
})