全网整合营销服务商

电脑端+手机端+微信端=数据同步管理

免费咨询热线:400-708-3566

C#实现rabbitmq 延迟队列功能实例代码

最近在研究rabbitmq,项目中有这样一个场景:在用户要支付订单的时候,如果超过30分钟未支付,会把订单关掉。当然我们可以做一个定时任务,每个一段时间来扫描未支付的订单,如果该订单超过支付时间就关闭,但是在数据量小的时候并没有什么大的问题,但是数据量一大轮训数据库的方式就会变得特别耗资源。当面对千万级、上亿级数据量时,本身写入的IO就比较高,导致长时间查询或者根本就查不出来,更别说分库分表以后了。除此之外,还有优先级队列,基于优先级队列的JDK延迟队列,时间轮等方式。但如果系统的架构中本身就有RabbitMQ的话,那么选择RabbitMQ来实现类似的功能也是一种选择。 我们项目中用到了rabbitmq,可以做一个延迟队列完美的解决这个问题。

rabbitmq本身不具有延时消息队列的功能,但是可以通过TTL(Time To Live)、DLX(Dead Letter Exchanges)特性实现。其原理给消息设置过期时间,在消息队列上为过期消息指定转发器,这样消息过期后会转发到与指定转发器匹配的队列上,变向实现延时队列。利用rabbitmq的这种特性,应该有了一个大概的思路。、

网上搜了一下  rabbitmq-delayed-message-exchange 这个插件也可以实现延迟队列的功能。今天介绍的是如何用C#来实现。

首先了解一下TTL和DLX

消息的TTL(Time To Live)

消息的TTL就是消息的存活时间。RabbitMQ可以对队列和消息分别设置TTL。对队列设置就是队列没有消费者连着的保留时间,也可以对每一个单独的消息做单独的设置。超过了这个时间,我们认为这个消息就死了,称之为死信。如果队列设置了,消息也设置了,那么会取小的。所以一个消息如果被路由到不同的队列中,这个消息死亡的时间有可能不一样(不同的队列设置)。这里单讲单个消息的TTL,因为它才是实现延迟任务的关键。

Dead Letter Exchanges

Exchage的概念在这里就不在赘述。一个消息在满足如下条件下,会进死信路由,记住这里是路由而不是队列,一个路由可以对应很多队列。

1. 一个消息被Consumer拒收了,并且reject方法的参数里requeue是false。也就是说不会被再次放在队列里,被其他消费者使用。

2. 上面的消息的TTL到了,消息过期了。

3. 队列的长度限制满了。排在前面的消息会被丢弃或者扔到死信路由上。

Dead Letter Exchange其实就是一种普通的exchange,和创建其他exchange没有两样。只是在某一个设置Dead Letter Exchange的队列中有消息过期了,会自动触发消息的转发,发送到Dead Letter Exchange中去。

 首先我建了两个控制台项目一个是生产者,一个是消费者。

生产者代码如下

      var factory = new ConnectionFactory() { HostName = "127.0.0.1", UserName = "test", Password = "test" };
      using (var connection = factory.CreateConnection())
      {
        while (Console.ReadLine() != null)
        {
          using (var channel = connection.CreateModel())
          {

            Dictionary<string, object> dic = new Dictionary<string, object>();
            dic.Add("x-expires", 30000);
            dic.Add("x-message-ttl", 12000);//队列上消息过期时间,应小于队列过期时间 
            dic.Add("x-dead-letter-exchange", "exchange-direct");//过期消息转向路由 
            dic.Add("x-dead-letter-routing-key", "routing-delay");//过期消息转向路由相匹配routingkey 
            //创建一个名叫"zzhello"的消息队列
            channel.QueueDeclare(queue: "zzhello",
              durable: true,
              exclusive: false,
              autoDelete: false,
              arguments: dic);

            var message = "Hello World!";
            var body = Encoding.UTF8.GetBytes(message);

            //向该消息队列发送消息message
            channel.BasicPublish(exchange: "",
              routingKey: "zzhello",
              basicProperties: null,
              body: body);
            Console.WriteLine(" [x] Sent {0}", message);
          }
        }
      }

      Console.ReadKey();

消费者代码如下:

 var factory = new ConnectionFactory() { HostName = "127.0.01", UserName = "test", Password = "test" };

      using (var connection = factory.CreateConnection())
      {
        using (var channel = connection.CreateModel())
        {
          channel.ExchangeDeclare(exchange: "exchange-direct", type: "direct");
          string name = channel.QueueDeclare().QueueName;
          channel.QueueBind(queue: name, exchange: "exchange-direct", routingKey: "routing-delay");

          //回调,当consumer收到消息后会执行该函数
          var consumer = new EventingBasicConsumer(channel);
          consumer.Received += (model, ea) =>
          {
            var body = ea.Body;
            var message = Encoding.UTF8.GetString(body);
            Console.WriteLine(ea.RoutingKey);
            Console.WriteLine(" [x] Received {0}", message);
          };

          //Console.WriteLine("name:" + name);
          //消费队列"hello"中的消息
          channel.BasicConsume(queue: name,
                     autoAck: true,
                     consumer: consumer);

          Console.WriteLine(" Press [enter] to exit.");
          Console.ReadLine();
        }
      }

      Console.ReadKey();

效果 :

在等待了12秒后消费者等到了消息。

 这样我们就实现了延迟队列的功能了。

以上就是本文的全部内容,希望对大家的学习有所帮助,也希望大家多多支持。


# rabbitmq  # 延迟队列  # 队列  # c#  # RabbitMQ 实现延迟队列的两种方式详解  # RabbitMQ死信机制实现延迟队列的实战  # 手把手带你掌握SpringBoot RabbitMQ延迟队列  # 如何通过Python实现RabbitMQ延迟队列  # RabbitMQ延迟队列及消息延迟推送实现详解  # Rabbitmq延迟队列实现定时任务的方法  # Spring Boot与RabbitMQ结合实现延迟队列的示例  # rabbitmq延迟队列的使用方式  # 中有  # 做一个  # 后会  # 来实现  # 以对  # 的是  # 有什么  # 就会  # 在这里  # 放在  # 才是  # 有可能  # 就有  # 死了  # 长时间  # 较高  # 也可  # 一大  # 我们可以  # 可以通过 


相关文章: 已有域名能否直接搭建网站?  制作表格网站有哪些,线上表格怎么弄?  网站制作网站,深圳做网站哪家比较好?  建站为何优先选择香港服务器?  如何确保FTP站点访问权限与数据传输安全?  网站专业制作公司,网站编辑是做什么的?好做吗?工作前景如何?  宝盒自助建站智能生成技巧:SEO优化与关键词设置指南  高端网站建设与定制开发一站式解决方案 中企动力  模具网站制作流程,如何找模具客户?  ,网站推广常用方法?  枣阳网站制作,阳新火车站打的到仙岛湖多少钱?  建站之星3.0如何解决常见操作问题?  太平洋网站制作公司,网络用语太平洋是什么意思?  建站之星IIS配置教程:代码生成技巧与站点搭建指南  建站之星如何配置系统实现高效建站?  南阳网站制作公司推荐,小学电子版试卷去哪里找资源好?  三星网站视频制作教程下载,三星w23网页如何全屏?  合肥制作网站的公司有哪些,合肥聚美网络科技有限公司介绍?  详解免费开源的.NET多类型文件解压缩组件SharpZipLib(.NET组件介绍之七)  如何通过西部数码建站助手快速创建专业网站?  子杰智能建站系统|零代码开发与AI生成SEO优化指南  c++ stringstream用法详解_c++字符串与数字转换利器  如何零成本快速生成个人自助网站?  详解一款开源免费的.NET文档操作组件DocX(.NET组件介绍之一)  东莞专业制作网站的公司,东莞大学生网的网址是什么?  如何通过网站建站时间优化SEO与用户体验?  电商网站制作价格怎么算,网上拍卖流程以及规则?  已有域名和空间如何搭建网站?  c# Task.Yield 的作用是什么 它和Task.Delay(1)有区别吗  如何高效利用200m空间完成建站?  淘宝制作网站有哪些,淘宝网官网主页?  建站主机与服务器功能差异如何区分?  建站主机数据库如何配置才能提升网站性能?  北京网站制作网页,网站升级改版需要多久?  详解免费开源的DotNet二维码操作组件ThoughtWorks.QRCode(.NET组件介绍之四)  如何在Ubuntu系统下快速搭建WordPress个人网站?  如何用wdcp快速搭建高效网站?  如何通过万网虚拟主机快速搭建网站?  ui设计制作网站有哪些,手机UI设计网址吗?  用v-html解决Vue.js渲染中html标签不被解析的问题  中山网站制作网页,中山新生登记系统登记流程?  表情包在线制作网站免费,表情包怎么弄?  如何在西部数码注册域名并快速搭建网站?  文字头像制作网站推荐软件,醒图能自动配文字吗?  网站制作培训多少钱一个月,网站优化seo培训课程有哪些?  制作销售网站教学视频,销售网站有哪些?  长沙做网站要多少钱,长沙国安网络怎么样?  西安大型网站制作公司,西安招聘网站最好的是哪个?  北京专业网站制作设计师招聘,北京白云观官方网站?  如何在自有机房高效搭建专业网站? 

您的项目需求

*请认真填写需求信息,我们会在24小时内与您取得联系。