DotNet Core&RabbitMQ死信队列

过期时间

RabbitMQ可以为消息和队列设置过期时间Time To Live(TTL)。其目的即过期。

223944337_53ad3c1c-7a7f-49cb-b5cd-8e316adb6180

消息过期时间

消息存储在队列中时，如果想为其设置一个有限的生命周期，而不是一直存储着，可以为其设置过期时间。比如，一条消息，我想要三分钟内有效，三分钟后再接收到该消息就算过时了，如果在队列中存储已经超过三分钟，消费者再去接收就是过时了，那便没有意义了。

223947216_f51a3105-002c-4438-b2e0-dc0bbbac07cc 为消息设置过期时间可以从两方面着手，一是为消息本身设置过期时间，二是为消息的承载体队列设置过期时间。两者同时设置情况下取最短生命周期。

为消息设置

在BasicPublish方法中，可以设置BasicProperties中的Expiration来设置过期时间(单位为毫秒)。

var connFactory = new ConnectionFactory
{
    HostName = "xxx.xxx.xxx.xxx",
    Port = 5672,
    UserName = "rabbitmqdemo",
    Password = "rabbitmqdemo@test",
    VirtualHost = "rabbitmqdemo"
};
using (var conn = connFactory.CreateConnection())
{
    using (var channel = conn.CreateModel())
    {
        var queueName = "messagettl_queue";
        channel.QueueDeclare(queue: queueName, durable: false, exclusive: false, autoDelete: false, arguments: null);


        while (true)
        {
            Console.WriteLine("消息内容(exit退出):");
            var message = Console.ReadLine();
            if (message.Trim().ToLower() == "exit")
            {
                break;
            }


            var body = Encoding.UTF8.GetBytes(message);
            var basicProperties = channel.CreateBasicProperties();
            basicProperties.Expiration = "10000";//10秒
            channel.BasicPublish(exchange: "", routingKey: queueName, basicProperties: basicProperties, body: body);
            Console.WriteLine("消息内容发送完毕:" + message);
        }
    }
}

如此一来，设置消息生命周期为10秒，当超过该时间后，再由消费者去获取该消息，则获取不到，可以直接从Web界面看到，经过10秒后，消息过期(未启动消费者)。 223948447_90191544-7335-4591-9138-6471332dbe0d

为队列设置

为每个消息设置过期时间可能不符合一些特定的场景，当需要设定特定队列中的消息都是指定的过期时间时，可以为队列中的消息统一设置过期时间。

223949658_dc10deeb-9e7a-4335-8a18-e320eabf09af 队列声明时可以指定参数，其中设置x-message-ttl参数。

var arguments = new Dictionary<string, object>
{
    { "x-message-ttl", 10000 }
};
channel.QueueDeclare(queue: queueName, durable: false, exclusive: false, autoDelete: false, arguments: arguments);

如此一来，该队列发送消息时，如消息本身没有设置过期时间，则使用队列的过期时间。生产者发送的7条消息过期时间都为10s，一段时间后，消息全部过期。

223951059_669b5411-13ca-42fe-892c-57a3dc21d34a 当ttl设置为0时，仅当消息能够立即被消费，否则消息立马过期，Web面板中只能见到消息发送，队列中没有消息，都被立马过期了。

223952505_b1f54223-fded-4ea3-a04b-b4e1e6c6c31e

过期策略

为队列中的消息统一设置过期时间时，当超出了过期时间，消息立马过期挪出队列。先发送的消息在队列头部，先过期的也在队列头部，因此可从头部扫描清除过期消息。

223954070_e8107d84-e7f2-4981-a253-69182290c57f 而为直接为消息设置过期时间时，各个消息的过期时间不尽相同，扫描时得队列全局扫描才能识别哪些消息是过期的。

223955627_7ce7fe11-e352-4771-bb1a-e21dd9193656 因此，设定在投递给消费者前判断是否过期，超出过期时间消息仍在队列中。

队列过期时间

此处的队列过期时间与消息中的为队列设置过期时间不同，此处是为队列本身考虑，队列自身没有消费者超过一段时间内且没有重新生命该队列，则无需考虑存在。

队列声明时指定参数，其中设置x-expires参数，需大于0，单位毫秒。

var arguments = new Dictionary<string, object>
{
    { "x-expires", 20000 }
};
channel.QueueDeclare(queue: queueName, durable: false, exclusive: false, autoDelete: false, arguments: arguments);

生产者启动应用创建队列，发送消息。 223956874_e149ff3b-9c15-4837-97ad-182043e4e87b 空闲20s不发送消息后队列被删除，再次发送消息匹配不到队列，消息被回退。

223958314_d7d0eff8-d917-4bbc-9838-7a1484eb005a

死信队列

死信交换机即Dead-Letter-Exchange(DLX)，和备份交换机一样，没有什么特殊，只是属性上标记了下，其绑定的队列称之为死信队列。当消息在一个队列中变成死信之后，被重发到死信交换机，存储到死信队列中。

223959818_b2034ebf-e106-4242-9541-d1b9feb589a3

消息变为死信情况

消息被拒绝且不重入队列
消息超出过期时间
队列达到最大长度

设置死信队列

声明队列时可以给定参数，其中设置x-dead-letter-exchange来指明该队列对应的死信交换机。

//死信交换机和死信队列
var dlxExchangeName = "dlx_exchange";
channel.ExchangeDeclare(exchange: dlxExchangeName, type: "fanout", durable: false, autoDelete: false, arguments: null);
var dlxQueueName = "dlx_queue";
channel.QueueDeclare(queue: dlxQueueName, durable: false, exclusive: false, autoDelete: false, arguments: null);
channel.QueueBind(queue: dlxQueueName, exchange: dlxExchangeName, routingKey: "");


//常规队列
var queueName = "normalmessage_queue";
var arguments = new Dictionary<string, object>
{
    { "x-message-ttl", 10000},
    { "x-dead-letter-exchange", dlxExchangeName }
};
channel.QueueDeclare(queue: queueName, durable: false, exclusive: false, autoDelete: false, arguments: arguments);

如下可见到，生产者初期发送的消息过期后经死信交换机路由进入到死信队列中，后期发送的暂未过期的消息仍在原队列中。当有该部分过期消息的需要时，消费者可以监听死信队列获取消息。 224001008_144e5d37-46e0-4b09-bbd6-dc4ddf5a8d19 当死信交换机的类型为direct时，可以指定RoutingKey(不指定默认使用原队列RoutingKey)。

224002646_faf6c5d3-ffc1-47b7-b94d-a3703ef52e00 可在声明常规队列的属性中设置x-dead-letter-routing-key，以能够匹配上死信交换机与死信队列绑定时的routingkey。

//死信交换机和死信队列
var dlxExchangeName = "dlxroutingkey_exchange";
channel.ExchangeDeclare(exchange: dlxExchangeName, type: "direct", durable: false, autoDelete: false, arguments: null);
var dlxQueueName1 = "dlx_queue1";
channel.QueueDeclare(queue: dlxQueueName1, durable: false, exclusive: false, autoDelete: false, arguments: null);
channel.QueueBind(queue: dlxQueueName1, exchange: dlxExchangeName, routingKey: "waring");
var dlxQueueName2 = "dlx_queue2";
channel.QueueDeclare(queue: dlxQueueName2, durable: false, exclusive: false, autoDelete: false, arguments: null);
channel.QueueBind(queue: dlxQueueName2, exchange: dlxExchangeName, routingKey: "info");
var dlxQueueName3 = "dlx_queue1";
channel.QueueDeclare(queue: dlxQueueName3, durable: false, exclusive: false, autoDelete: false, arguments: null);
channel.QueueBind(queue: dlxQueueName3, exchange: dlxExchangeName, routingKey: "error");


//常规队列
var queueName = "normalmessage_queue";
var arguments = new Dictionary<string, object>
{
    { "x-message-ttl", 10000},
    { "x-dead-letter-exchange", dlxExchangeName },
    { "x-dead-letter-routing-key", "info" }
};
channel.QueueDeclare(queue: queueName, durable: false, exclusive: false, autoDelete: false, arguments: arguments);

当消息过期时，死信交换机根据常规队列绑定的routingkey，匹配到相应的死信队列存储。先发送的消息过期后存入到对应队列中，后续暂未过期消息仍然保持在原队列。 224003955_683c0cc6-cc21-481a-bc45-517e628fef39

延迟队列

延迟队列算是死信队列的一种应用场景，其本身并不在RabbitMQ或是AMQP协议中有所体现。当不想消费者立马获取到消息，而是等待一段时间才让消费者消费时，比如订单限时支付，超出时间未支付则取消订单，那么可以使用到延迟队列来处理这一场景。

224005487_ebfe539b-2a90-4117-bd2f-975d1f4d40ad

消息过期时间

发送消息时设置过期时间场景下，消息可能并不会在过期后立马从队列中删除，而是要等到消费时候才会判断该消息是否过时。当出现以下场景时则会有点问题。

第一个消息的过期时间很长，而后续的消息的过期时间很短，后续的消息过期后不会立马删除，而是要等到第一个消息过期删除后才会被删除，那么对应延迟队列来说会有点问题，时间超出了设定的延迟时间。

224007064_afa79b4b-4708-4ebf-8afc-e6071d95a821 注：对队列设定消息过期时间不存在该问题。

解决方案

RabbitMQ提供了延迟队列的插件，提供延迟队列类型交换机，其不会根据第一个消息是否过期来判断，解决了如上提到的第一个没有过期，后续消息过期的场景，不会受消息先后顺序的影响，而是关注过期时间，先过期的先发送。

224008420_5b8efb38-6b1d-4010-a80c-a8a35e84e32c

生产者代码

声明延迟交换机时类型使用x-delayed-message，需要注意声明交换机类型时需要给定参数x-delayed-type，至于值是哪种类型可依据匹配队列的需要选择。延迟交换机和延迟队列都需要持久化。消息发送时需要使用消息头并设置x-delay来设置延迟时间。

var connFactory = new ConnectionFactory
{
    HostName = "xxx.xxx.xxx.xxx",
    Port = 5672,
    UserName = "rabbitmqdemo",
    Password = "rabbitmqdemo@test",
    VirtualHost = "rabbitmqdemo"
};
using (var conn = connFactory.CreateConnection())
{
    using (var channel = conn.CreateModel())
    {
        //延迟交换机
        var delayExchangeName = "delay_exchange";
        var delayArguments = new Dictionary<string, object>
        {
            { "x-delayed-type", "direct" } //x-delayed-type必须，否则启动报错
        };
        channel.ExchangeDeclare(exchange: delayExchangeName, type: "x-delayed-message", durable: true, autoDelete: false, arguments: delayArguments); //持久化必须，否则启动报错


        //延迟队列
        var delayQueueName = "delay_queue";
        channel.QueueDeclare(delayQueueName, durable: true, exclusive: false, autoDelete: false);//持久化必须，否则启动报错
        channel.QueueBind(queue: delayQueueName, exchange: delayExchangeName, routingKey: delayQueueName);


        channel.BasicReturn += new EventHandler<RabbitMQ.Client.Events.BasicReturnEventArgs>((sender, e) =>
        {
            var message = Encoding.UTF8.GetString(e.Body.ToArray());
            Console.WriteLine($"收到回退消息：{message}");
        });


        while (true)
        {
            Console.WriteLine("消息内容(exit退出):");
            var message = Console.ReadLine();
            if (message.Trim().ToLower() == "exit")
            {
                break;
            }


            var body = Encoding.UTF8.GetBytes(message);
            var basicProperties = channel.CreateBasicProperties();
            basicProperties.Headers = new Dictionary<string, object>
            {
                { "x-delay", message == "aaa" ? 30000 : 10000 }//延时时间从header赋值
            };
            channel.BasicPublish(exchange: delayExchangeName, routingKey: delayQueueName, mandatory: true, basicProperties: basicProperties, body: body);
            Console.WriteLine("消息内容发送完毕:" + message + $" {DateTime.Now}");
        }
    }
}

生产者发送消息，到延迟交换机，消息将在设定的延迟时间后路由到相应的延迟队列。 224009800_f51cd113-2faf-4ee7-9a6b-bed7333eeaff 此处设置了mandatory为true，消息最终到了延迟队列，但也存在回退到了生产者，实际上对于该延迟交换机插件并不支持mandatory，官方不建议使用该参数。

2022-08-29,望技术有成后能回来看见自己的脚步