.NET中使用RabbitMQ延时队列和死信队列

金骏五金

RabbitMQ延时队列和死信队列

延时队列和死信队列

延时队列是RabbitMQ中的一种特殊队列，它可以在消息到达队列后延迟一段时间再被消费。
延时队列的实现原理是通过使用消息的过期时间和死信队列来实现。当消息被发送到延时队列时，可以为消息设置一个过期时间，这个过期时间决定了消息在延时队列中等待的时间。如果消息在过期时间内没有被消费者消费，则会被自动发送到一个预先指定的死信队列中。

在RabbitMQ中，延时队列的实现可以通过以下步骤来完成：

• 创建一个普通的队列作为延时队列，设置x-message-ttl参数为消息的过期时间。
  
• 创建一个死信队列，用于接收延时队列中过期的消息。
  
• 将延时队列设置为普通队列的死信交换机，并指定死信路由键。
  
• 将消费者绑定到死信队列，以消费延时队列中过期的消息。
  

使用场景

• 订单在十分钟内未支付则自动取消。
  
• 新创建的店铺，如果十天内都没有上传过商品，则自动发送消息提醒。
  
• 账单在一周内未支付，则自动结算。
  
• 用户发起退款，如果三天内没有得到处理则通知相关运营人员。
  
• 还有很多场景就不一一例举了。
  

TTL设置

方式一：

创建队列时设置x-message-ttl的属性，所有被投递到该队列的消息最多都不会超过60s

1. var args = new Dictionary<string,object>();
2. args.Add("x-message-ttl",60000); //单位为毫秒
3. model.QueueDeclare("myqueue",false,false,false,args);

复制代码

方式二：

为每条消息设置TTL，为每条消息设置过期时间。

1. IBasicProperties props = model.CreateBasicProperties();
2. props.ContentType = "text/plain";
3. props.DeliveryMode = 2;
4. props.Expiration = "60000"
5. model.BasicPublic(exchangeName,routingKey,props,messageBodyBytes);

复制代码

代码实践

模拟支付业务
整个项目由三部分组成

• Web API项目：用于发送订单请求，生产者。
  
• 控制台项目一：用于处理订单支付，延时队列。
  
• 控制台项目二：用于处理超时未支付的订单，死信队列。
  

Web API项目

订单类，就简单的写一个用于演示，真实业务肯定不是这样~

1. public class OrderDto
2. {
3.     /// <summary>
4.     /// 订单名称
5.     /// </summary>
6.     public string Name { get; set; }
7.     /// <summary>
8.     /// 订单状态
9.     /// 0 未支付
10.     /// 1 已支付
11.     /// 2 超时
12.     /// </summary>
13.     public int Status { get; set; }
14. }

复制代码

控制器

1. [ApiController]
2. [Route("api/orders")]
3. public class OrdersController : ControllerBase
4. {
5.     private readonly IOrderService _orderService;
7.     public OrdersController(IOrderService orderService)
8.     {
9.         _orderService = orderService;
10.     }
12.     [HttpPost]
13.     public IActionResult CreateOrder([FromBody] OrderDto orderDto)
14.     {
15.         // 处理订单逻辑
16.         _orderService.ProcessOrder(orderDto);
17.         return Ok();
18.     }
19. }

复制代码

订单服务

1. public interface IOrderService
2. {
3.     void ProcessOrder(OrderDto orderDto);
4. }
6. public class OrderService : IOrderService
7. {
8.     private readonly RabbitMQConnectionFactory _connectionFactory;
9.    
10.     public OrderService(RabbitMQConnectionFactory connectionFactory)
11.     {
12.         _connectionFactory = connectionFactory;
13.     }
15.     public void ProcessOrder(OrderDto orderDto)
16.     {
17.         using (var channel = _connectionFactory.CreateChannel())
18.         {
19.             var properties = channel.CreateBasicProperties();
20.             properties.Headers = new Dictionary<string, object>
21.             {
22.                 { "x-delay", 1000 * 20  } // 设置20秒延时
23.             };
25.             var message = JsonConvert.SerializeObject(orderDto);
26.             var body = Encoding.UTF8.GetBytes(message);
28.             channel.BasicPublish("delayed_exchange", "routing_key", properties, body);
29.         }
30.     }
31. }

复制代码

支付处理项目

ProcessPay类，用于接收订单消息

1. public class ProcessPay : IHostedService
2. {
3.     private readonly ConnectionFactory _factory;
4.     private IConnection _connection;
5.     private IModel _channel;
7.     public ProcessPay()
8.     {
9.         _factory = new ConnectionFactory()
10.         {
11.             HostName = "ip",
12.             Port = 5672,
13.             UserName = "用户名",
14.             Password = "密码"
15.         };
16.     }
18.     public Task StartAsync(CancellationToken cancellationToken)
19.     {
20.         Console.WriteLine(" Press [enter] to exit.");
21.         _connection = _factory.CreateConnection();
22.         _channel = _connection.CreateModel();
24.         _channel.ExchangeDeclare("delayed_exchange", ExchangeType.Direct, true, false, null);
25.         //关键代码，绑定死信交换机
26.         var arguments = new Dictionary<string, object>
27.         {
28.             { "x-dead-letter-exchange", "dead_letter_exchange" },
29.             { "x-dead-letter-routing-key", "dead_letter_routing_key" }
30.         };
31.         _channel.QueueDeclare("delayed_queue", true, false, false, arguments);
32.         _channel.QueueBind("delayed_queue", "delayed_exchange", "routing_key");
34.         var consumer = new EventingBasicConsumer(_channel);
35.         consumer.Received += (model, ea) =>
36.         {
37.             var body = ea.Body.ToArray();
38.             var message = Encoding.UTF8.GetString(body);
40.             // 处理支付逻辑
41.             var orderDto = JsonConvert.DeserializeObject<OrderDto>(message);
42.             Console.WriteLine($"订单信息:{orderDto.Name}");
43.             Console.WriteLine("请输入价格（模拟支付）：");
45.             // 超时未支付
46.             string? many = "";
47.             // 支付处理
48.             Console.WriteLine("请输入:");
49.             // 超时未支付进行处理
50.             Task.Factory.StartNew(() =>
51.             {
52.                 many = Console.ReadLine();
53.                 Console.WriteLine($"many:{many}");
54.             }).Wait(20 * 1000);
55.             if (string.Equals(many, "100"))
56.             {
57.                 orderDto.Status = 1;
58.                 Console.WriteLine("支付完成");
59.                 _channel.BasicAck(ea.DeliveryTag, true);
60.             }
61.             else
62.             {
63.                 //重试几次依然失败
64.                 Console.WriteLine("等待一定时间内失效超时未支付的订单");
65.                 _channel.BasicNack(ea.DeliveryTag, false, false);
66.             }
67.         };
69.         _channel.BasicConsume("delayed_queue", false, consumer);
71.         return Task.CompletedTask;
72.     }
74.     public Task StopAsync(CancellationToken cancellationToken)
75.     {
76.         _channel?.Close();
77.         _connection?.Close();
78.         _channel?.Dispose();
79.         _connection?.Dispose();
81.         return Task.CompletedTask;
82.     }
83. }

复制代码

在Main方法中使用单例模式注册该服务，当然直接将代码写在Main方法也是没有问题的，只不过这种方式方便管理。

1. static void Main(string[] args)
2. {
3.     var host = new HostBuilder()
4.         .ConfigureServices((hostContext, services) =>
5.                            {
6.                                services.AddSingleton<IHostedService,ProcessPay>();
7.                            })
8.         .Build();
10.     host.Run();
11. }

复制代码

支付超时项目

创建一个死信队列服务，用于订阅死信队列中的订单消息，这里我就直接把代码写在Main方法中了

1. using (var connection = factory.CreateConnection())
2. {
3.     using (var channel = connection.CreateModel())
4.     {
5.         channel.ExchangeDeclare("dead_letter_exchange", ExchangeType.Direct, true, false, null);
7.         channel.QueueDeclare("dead_letter_queue", true, false, false, null);
9.         channel.QueueBind("dead_letter_queue", "dead_letter_exchange", "dead_letter_routing_key");
10.         
11.         var consumer = new EventingBasicConsumer(channel);
12.         consumer.Received += (model, ea) =>
13.         {
14.             var body = ea.Body.ToArray();
15.             var message = Encoding.UTF8.GetString(body);
17.             // 处理超时未支付的订单逻辑
18.             var orderDto = JsonConvert.DeserializeObject<OrderDto>(message);
19.             orderDto.Status = 2;
20.             Console.WriteLine($"订单信息:{orderDto.Name},{orderDto.Status}");
21.             Console.WriteLine("超时未支付");
23.             channel.BasicAck(ea.DeliveryTag, true);
24.         };
26.         channel.BasicConsume("dead_letter_queue", false, consumer);
27.         
28.         Console.WriteLine(" Press [enter] to exit.");
29.         Console.ReadLine();
30.     }
31. }

复制代码

效果展示

代码看不出效果，直接上图。
首先是3个项目各自运行效果图

然后演示正常消费效果

接下来是超时未支付效果

结尾

这就是一个简单的延时队列和死信队列的代码，模拟了支付超时的场景，这里的数据都写死了的，真实运用的时候肯定是中数据库中获取，修改数据库实体的值。然后死信队列是用于处理在一定时间内未被处理的消息，死信交换机也只是一个普通的交换机，只不过他是用于处理超时的消息的交换机。
对于RabbitMQ的文章基本就结束了，可能还会有一篇RabbitMQ集群搭建的文章，但不是很想去写，最近太懒了~
有问题欢迎指出，活到老学到老~
RabbitMQ系列文章

• ZY知识库 · ZY - 在.NET Core中使用RabbitMQ (pljzy.top)
  

参考资料

• 【【2021最新.NET/C#】RabbitMQ从零到高可用集群】
  

来源:https://www.cnblogs.com/ZYPLJ/archive/2023/07/30/17591838.html
免责声明：由于采集信息均来自互联网，如果侵犯了您的权益，请联系我们【E-Mail:cb@itdo.tech】 我们会及时删除侵权内容，谢谢合作！