简单的描述就是:P(生产者)发送消息到Q1(延时队列),Q1的消息有过期时间,比如10s,那10s后消息过期就会触发死信,从而把消息转发到Q2(死信队列)。
解决问题场景:像商城下单,未支付时取消订单场景。下单时写一条记录入Q1,延时30分钟后转到Q2,消费Q2,检查订单,支付则不做操作,没支付则取消订单,恢复库存。
生产者代码:
public static void SendMessage() { //死信交换机 string dlxexChange = "dlx.exchange"; //死信队列 string dlxQueueName = "dlx.queue"; //消息交换机 string exchange = "direct-exchange"; //消息队列 string queueName = "delay_queue"; using (var connection = RabbitMQHelper.GetConnection()) { using (var channel = connection.CreateModel()) { //创建死信交换机 channel.ExchangeDeclare(dlxexChange, type: ExchangeType.Direct, durable: true, autoDelete: false); //创建死信队列 channel.QueueDeclare(dlxQueueName, durable: true, exclusive: false, autoDelete: false); //死信队列绑定死信交换机 channel.QueueBind(dlxQueueName, dlxexChange, routingKey: dlxQueueName); // 创建消息交换机 channel.ExchangeDeclare(exchange, type: ExchangeType.Direct, durable: true, autoDelete: false); //创建消息队列,并指定死信队列,和设置这个队列的消息过期时间为10s channel.QueueDeclare(queueName, durable: true, exclusive: false, autoDelete: false, arguments: new Dictionary<string, object> { { "x-dead-letter-exchange",dlxexChange}, //设置当前队列的DLX(死信交换机) { "x-dead-letter-routing-key",dlxQueueName}, //设置DLX的路由key,DLX会根据该值去找到死信消息存放的队列 { "x-message-ttl",10000} //设置队列的消息过期时间 }); //消息队列绑定消息交换机 channel.QueueBind(queueName, exchange, routingKey: queueName); string message = "hello rabbitmq message"; var properties = channel.CreateBasicProperties(); properties.Persistent = true; //发布消息 channel.BasicPublish(exchange: exchange, routingKey: queueName, basicProperties: properties, body: Encoding.UTF8.GetBytes(message)); Console.WriteLine($"{DateTime.Now},向队列:{queueName}发送消息:{message}"); } } }
消费者代码:
public static void Consumer() { //死信交换机 string dlxexChange = "dlx.exchange"; //死信队列 string dlxQueueName = "dlx.queue"; var connection = RabbitMQHelper.GetConnection(); { //创建信道 var channel = connection.CreateModel(); { //创建死信交换机 channel.ExchangeDeclare(dlxexChange, type: ExchangeType.Direct, durable: true, autoDelete: false); //创建死信队列 channel.QueueDeclare(dlxQueueName, durable: true, exclusive: false, autoDelete: false); //死信队列绑定死信交换机 channel.QueueBind(dlxQueueName, dlxexChange, routingKey: dlxQueueName); var consumer = new EventingBasicConsumer(channel); channel.BasicQos(prefetchSize: 0, prefetchCount: 1, global: true); consumer.Received += (model, ea) => { //处理业务 var message = Encoding.UTF8.GetString(ea.Body.ToArray()); Console.WriteLine($"{DateTime.Now},队列{dlxQueueName}消费消息:{message}"); channel.BasicAck(ea.DeliveryTag, false); }; channel.BasicConsume(dlxQueueName, autoAck: false, consumer); } } }
执行代码:
向延时队列发送消息,监听死信队列,发送和收到消息时间刚好是设置的10s。
RabbitMQ管理界面:
三、延时消息设置不同过期时间上面的延时队列能解决消息过期时间都是相同的场景,能不能解决消息的过期时间是不一样的呢?
例如场景:机器人客服,为了更像人为操作,收到消息后要随机3-10秒回复客户。
1)队列不设置TTL(消息过期时间),把过期时间设置在消息上。
生产者代码:
public static void SendMessage() { //死信交换机 string dlxexChange = "dlx.exchange"; //死信队列 string dlxQueueName = "dlx.queue"; //消息交换机 string exchange = "direct-exchange"; //消息队列 string queueName = "delay_queue"; using (var connection = RabbitMQHelper.GetConnection()) { using (var channel = connection.CreateModel()) { //创建死信交换机 channel.ExchangeDeclare(dlxexChange, type: ExchangeType.Direct, durable: true, autoDelete: false); //创建死信队列 channel.QueueDeclare(dlxQueueName, durable: true, exclusive: false, autoDelete: false); //死信队列绑定死信交换机 channel.QueueBind(dlxQueueName, dlxexChange, routingKey: dlxQueueName); // 创建消息交换机 channel.ExchangeDeclare(exchange, type: ExchangeType.Direct, durable: true, autoDelete: false); //创建消息队列,并指定死信队列,和设置这个队列的消息过期时间为10s channel.QueueDeclare(queueName, durable: true, exclusive: false, autoDelete: false, arguments: new Dictionary<string, object> { { "x-dead-letter-exchange",dlxexChange}, //设置当前队列的DLX(死信交换机) { "x-dead-letter-routing-key",dlxQueueName}, //设置DLX的路由key,DLX会根据该值去找到死信消息存放的队列 //{ "x-message-ttl",10000} //设置队列的消息过期时间 }); //消息队列绑定消息交换机 channel.QueueBind(queueName, exchange, routingKey: queueName); string message = "hello rabbitmq message 10s后处理"; var properties = channel.CreateBasicProperties(); properties.Persistent = true; properties.Expiration = "10000";//消息的有效期10s //发布消息,延时10s channel.BasicPublish(exchange: exchange, routingKey: queueName, basicProperties: properties, body: Encoding.UTF8.GetBytes(message)); Console.WriteLine($"{DateTime.Now},向队列:{queueName}发送消息:{message},延时:10s"); string message2 = "hello rabbitmq message 5s后处理"; var properties2 = channel.CreateBasicProperties(); properties2.Persistent = true; properties2.Expiration = "5000";//消息有效期5s //发布消息,延时5s channel.BasicPublish(exchange: exchange, routingKey: queueName, basicProperties: properties2, body: Encoding.UTF8.GetBytes(message2)); Console.WriteLine($"{DateTime.Now},向队列:{queueName}发送消息:{message2},延时:5s"); } } }
消费者代码还是上面延时队列的不变,先试下效果。