C#通过rabbitmq如何实现定时任务

小编给大家分享一下C#通过rabbitmq如何实现定时任务，相信大部分人都还不怎么了解，因此分享这篇文章给大家参考一下，希望大家阅读完这篇文章后大有收获，下面让我们一起去了解一下吧！

环境准备

需要在MQ中进行安装插件 地址链接插件介绍地址：https://www.rabbitmq.com/blog/2015/04/16/scheduling-messages-with-rabbitmq/

使用场景

作为一个新的预支付订单被初始化放置，如果该订单在指定时间内未进行支付，则将被认为超时订单进行关闭处理；电商系统中应用较多，用户购买商品产生订单，但未进行支付，订单产生30分钟内未支付将关闭订单（且满足该场景数量庞大），不可能采用人工干预。

代码介绍

生产者

varfactory=newConnectionFactory()
{
Uri=newUri("MQ地址")
};

usingvarconnection=factory.CreateConnection();
usingvarchannel=connection.CreateModel();


varexchangeName="delay-exchange";
varroutingkey="delay.delay";
varqueueName="delay_queueName";
//设置Exchange队列类型
varargMaps=newDictionary<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(inti=0;i<3;i++)
{
vartime=1000*5;
varmessage=$@"发送时间为{DateTime.Now:yyyy-MM-ddHH:mm:ss}延时时间为:{time}";
varbody=Encoding.UTF8.GetBytes(message);
varprops=channel.CreateBasicProperties();
//设置消息的过期时间
props.Headers=newDictionary<string,object>()
{
{"x-delay",5000}
};
channel.BasicPublish(exchange:exchangeName,
routingKey:routingkey,
basicProperties:props,
body:body);
Console.WriteLine(message);


}
Console.ReadLine();

消费者(自动绑定队列写法)

varfactory=newConnectionFactory()
{
Uri=newUri(MQ地址)
};
usingvarconnection=factory.CreateConnection();
usingvarchannel=connection.CreateModel();
varqueueName="delay_queueName";
channel.QueueDeclare(queueName,true,false,false,null);
varconsumer=newEventingBasicConsumer(channel);
consumer.Received+=(model,ea)=>
{
varbody=ea.Body;
varmessage=Encoding.UTF8.GetString(body);
varroutingKey=ea.RoutingKey;
Console.WriteLine($@"接受到消息的时间为{DateTime.Now:yyyy-MM-ddHH:mm:ss},routingKey:{routingKey}message:{message}");
};
channel.BasicConsume(queue:queueName,
autoAck:true,
consumer:consumer);
Console.ReadLine();

消费者(手动绑定队列写法)

varfactory=newConnectionFactory()
{
Uri=newUri(MQ地址)
};
usingvarconnection=factory.CreateConnection();
usingvarchannel=connection.CreateModel();
varexchangeName="delay-exchange";
varroutingkey="delay.delay";
varqueueName="delay_queueName";
varautoDelete=true;

varargMaps=newDictionary<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);
varconsumer=newEventingBasicConsumer(channel);
consumer.Received+=(model,ea)=>
{
varbody=ea.Body;
varmessage=Encoding.UTF8.GetString(body);
varroutingKey=ea.RoutingKey;
Console.WriteLine($@"接受到消息的时间为{DateTime.Now:yyyy-MM-ddHH: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)

以上是“C#通过rabbitmq如何实现定时任务”这篇文章的所有内容，感谢各位的阅读！相信大家都有了一定的了解，希望分享的内容对大家有所帮助，如果还想学习更多知识，欢迎关注恰卡编程网行业资讯频道！