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() { {"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 () { {"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() { {"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如何实现定时任务”这篇文章的所有内容,感谢各位的阅读!相信大家都有了一定的了解,希望分享的内容对大家有所帮助,如果还想学习更多知识,欢迎关注恰卡编程网行业资讯频道!
目录