C#通过rabbitmq实现定时任务(延时队列)
作者:初夏的阳光丶 时间:2021-11-02 23:11:17
本文主要讲解如何通过RabbitMQ实现定时任务(延时队列)
环境准备
需要在MQ中进行安装插件 地址链接
插件介绍地址:https://www.rabbitmq.com/blog/2015/04/16/scheduling-messages-with-rabbitmq/
使用场景
作为一个新的预支付订单被初始化放置,如果该订单在指定时间内未进行支付,则将被认为超时订单进行关闭处理;电商系统中应用较多,用户购买商品产生订单,但未进行支付,订单产生30分钟内未支付将关闭订单(且满足该场景数量庞大),不可能采用人工干预。
代码介绍
生产者
var factory = new ConnectionFactory()
{
Uri = new Uri("MQ地址")
};
using var connection = factory.CreateConnection();
using var channel = connection.CreateModel();
var exchangeName = "delay-exchange";
var routingkey = "delay.delay";
var queueName = "delay_queueName";
//设置Exchange队列类型
var argMaps = new Dictionary<string, object>()
{
{"x-delayed-type", "topic"}
};
//设置当前消息为延时队列
channel.ExchangeDeclare(exchange: exchangeName, type: "x-delayed-message", true, false, argMaps);
channel.QueueDeclare(queueName, true, false, false, argMaps);
channel.QueueBind(queueName, exchangeName, routingkey);
for (int i = 0; i < 3; i++)
{
var time = 1000 * 5;
var message = $@"发送时间为 {DateTime.Now:yyyy-MM-dd HH:mm:ss} 延时时间为:{time}";
var body = Encoding.UTF8.GetBytes(message);
var props = channel.CreateBasicProperties();
//设置消息的过期时间
props.Headers = new Dictionary<string, object>()
{
{ "x-delay", 5000 }
};
channel.BasicPublish(exchange: exchangeName,
routingKey: routingkey,
basicProperties: props,
body: body);
Console.WriteLine(message);
}
Console.ReadLine();
消费者(自动绑定队列写法)
var factory = new ConnectionFactory()
{
Uri = new Uri(MQ地址)
};
using var connection = factory.CreateConnection();
using var channel = connection.CreateModel();
var queueName = "delay_queueName";
channel.QueueDeclare(queueName, true, false, false, null);
var consumer = new EventingBasicConsumer(channel);
consumer.Received += (model, ea) =>
{
var body = ea.Body;
var message = Encoding.UTF8.GetString(body);
var routingKey = ea.RoutingKey;
Console.WriteLine($@"接受到消息的时间为 {DateTime.Now:yyyy-MM-dd HH:mm:ss},routingKey:{routingKey} message:{message} ");
};
channel.BasicConsume(queue: queueName,
autoAck: true,
consumer: consumer);
Console.ReadLine();
消费者(手动绑定队列写法)
var factory = new ConnectionFactory()
{
Uri = new Uri(MQ地址)
};
using var connection = factory.CreateConnection();
using var channel = connection.CreateModel();
var exchangeName = "delay-exchange";
var routingkey = "delay.delay";
var queueName = "delay_queueName";
var autoDelete = true;
var argMaps = new Dictionary<string, object>()
{
{"x-delayed-type", "topic"}
};
channel.ExchangeDeclare(exchange: exchangeName, type: "x-delayed-message", true, false, argMaps);
channel.QueueDeclare(queueName, true, false, false, argMaps);
channel.QueueBind(queue: queueName, exchange: exchangeName, routingKey: routingkey);
//channel.QueueDeclare(queueName, true, false, false, null);
var consumer = new EventingBasicConsumer(channel);
consumer.Received += (model, ea) =>
{
var body = ea.Body;
var message = Encoding.UTF8.GetString(body);
var routingKey = ea.RoutingKey;
Console.WriteLine($@"接受到消息的时间为 {DateTime.Now:yyyy-MM-dd HH:mm:ss},routingKey:{routingKey} message:{message} ");
};
channel.BasicConsume(queue: queueName,
autoAck: true,
consumer: consumer);
Console.ReadLine();
最终实现效果(两个消费者)
在上述实现中,其实主要靠以下参数来帮我们实现当前功能
声明Exchange中的 type: "x-delayed-message" 这个表明当前队列为延时消息队列
声明Exchange中arguments中的 {"x-delayed-type", "topic"} 当前表明当前队列为Topic模式
最后 我们在CreateBasicProperties的Header中设置 { "x-delay", 5000 }来达到消息延时的功能(单位为ms)
建议
如果使用当前模式来做定时任务,在要求消息不丢失的前提下,需要运维同学提供稳定的MQ环境
来源:https://www.cnblogs.com/ancold/p/14705601.html
标签:C#,rabbitmq,定时任务
0
投稿
猜你喜欢
解决Maven中关于依赖导入不进的问题
2023-09-05 23:13:08
C#生成带注释的dll并引用实现
2023-08-31 16:01:09
利用Java计算某个日期是星期几
2023-11-17 05:49:42
java中使用数组进行模拟加密的方法
2023-11-18 15:37:39
java创建多级目录文件的实例讲解
2023-11-27 09:29:02
深入分析java并发编程中volatile的实现原理
2023-11-23 05:02:25
聊聊Controller中RequestMapping的作用
2021-12-08 20:48:45
Spring Data Jpa 复杂查询方式总结(多表关联及自定义分页)
2023-11-21 10:01:27
java关于字符串的常用API
2022-09-13 22:19:03
Java jpa外连接查询join案例详解
2022-12-17 18:31:15
Java8如何使用Lambda表达式简化代码详解
2023-02-20 04:44:58
Android中实现根据资源名获取资源ID
2023-06-20 04:18:30
Java实现简单的飞机大战游戏(控制主飞机篇)
2023-11-14 13:52:56
JAVA操作MongoDB数据库实例教程
2023-11-18 13:22:27
maven无法依赖spring-cloud-stater-zipkin的解决方案
2023-09-12 01:54:51
Java实现克隆的三种方式实例总结
2021-11-21 15:26:14
spring 注解如何开启声明式事务
2023-04-09 23:07:42
Java基于虹软实现人脸识别、人脸比对、活性检测等
2023-02-18 15:29:09
java实现订餐系统
2023-08-12 04:39:06
浅谈关于Mybatis的mapper-locations配置问题
2023-09-24 06:06:16