服务器之家

服务器之家 > 正文

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

时间:2021-12-31 14:38     来源/作者:ThisIsTest

最近在研究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中去。

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

生产者代码如下

?
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
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();

消费者代码如下:

?
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
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();

效果 :

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

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

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

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

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

原文链接:http://www.cnblogs.com/ListenCode/p/6709145.html?utm_source=tuicool&utm_medium=referral

标签:

相关文章

热门资讯

yue是什么意思 网络流行语yue了是什么梗
yue是什么意思 网络流行语yue了是什么梗 2020-10-11
背刺什么意思 网络词语背刺是什么梗
背刺什么意思 网络词语背刺是什么梗 2020-05-22
2020微信伤感网名听哭了 让对方看到心疼的伤感网名大全
2020微信伤感网名听哭了 让对方看到心疼的伤感网名大全 2019-12-26
蜘蛛侠3英雄无归3正片免费播放 蜘蛛侠3在线观看免费高清完整
蜘蛛侠3英雄无归3正片免费播放 蜘蛛侠3在线观看免费高清完整 2021-08-24
2021年耽改剧名单 2021要播出的59部耽改剧列表
2021年耽改剧名单 2021要播出的59部耽改剧列表 2021-03-05
返回顶部