C#实现rabbitmq 延迟队列功能实例代码
作者:ThisIsTest 发布时间:2023-03-18 02:59:34
最近在研究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秒后消费者等到了消息。
这样我们就实现了延迟队列的功能了。
来源:http://www.cnblogs.com/ListenCode/p/6709145.html?utm_source=tuicool&utm_medium=referral
猜你喜欢
- monaco editor创建//创建和设置值if (!this.monacoEditor) { this.monacoEdit
- 一. break1. 作用break关键字可以用于for、while、do-while及switch语句中,用来跳出整个语句块,结束当前循环
- 在以往的 Tomcat 项目中,一直习惯用 Ant 打包,使用 build.xml 配置,通过 ant -buildfile 的方式在机器上
- 本文为大家分享了java门禁系统面向对象程序设计,供大家参考,具体内容如下下面是对门禁系统的介绍:需求1、某公司安装了电子门,要设计门禁系统
- 简介本文用示例介绍java的Duration的用法。Duration和Period说明Duration类通过秒和纳秒相结合来描述一个时间量,
- jcasbin简介:jcasbin 是一个用 Java 语言打造的轻量级开源访问控制框架https://github.com/casbin/
- URL(Uniform Resource Locator)是统一资源 * ,它是对可以从互联网上得到的资源的位置和访问方法的一种简洁的表示,
- 一.优先队列的应用优先队列在程序开发中屡见不鲜,比如操作系统在进行进程调度时一种可行的算法是使用优先队列,当一个新的进程被fork()出来后
- 见过一句夸张的话,叫做“没有阅读过jdk源码的人不算学过java”。从今天起开始精读源码。而适合精读的源码无非就是java.io,.util
- 本文实例为大家分享了SSM实现学生管理系统的具体代码,供大家参考,具体内容如下概述基于Spring + Spring MVC 的学生管理系统
- 本文通过解决老王经常搞错借书人的问题,来引出行为型模式中的命令模式。为了在案例之上理解的更加透彻,我们需要了解命令模式在源码中的应用。最后指
- 1. 什么是AOPAOP (Aspect Oriented Programming)意为:面向切面编程,通过预编译方式和运行期 * 实现在
- 今天有同事用swagger2开发时,有一方法返回Map<String,List<Object>>出现无法解析错误。P
- 参考dubbo和shenyu网关实现自定义的SPISPI标注注解标注提供SPI能力接口的注解@Documented@Retention(Re
- 最近在搭建springmvc的框架,遇到的这样的问题:在地址栏访问登陆界面访问不了,http://localhost/XXXX/WEB-IN
- 前两天发现 idea 终于更新了2020.1版本,新增了好多的特性,这里不介绍,主要写一下中文插件的安装首先下载新版 安装包 https:/
- Lua是目前国内使用最多的热更语言,基于Lua的热更框架也非常多,最近学习了一下ToLua的热更框架,主要使用的问题在于C#和Lua之间的互
- 1. 概述在 Spring Security 5.2 中增强了 DSL 的功能:允许使用 Lambda 表达式来配置 HTTP securi
- 以前也用过爬虫,比如使用nutch爬取指定种子,基于爬到的数据做搜索,还大致看过一些源码。当然,nutch对于爬虫考虑的是十分全面和细致的。
- 如何创建可以存放各种类型的数组?根据JavaSE的语法知识储备,如果现在让你们创建如标题一样的数组,你会怎么创建呢?答案是:使用 Objec