最近在研究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();
|
效果 ：
在等待了12秒后消费者等到了消息。
这样我们就实现了延迟队列的功能了。
以上就是本文的全部内容，希望对大家的学习有所帮助，也希望大家多多支持服务器之家。
原文链接：http://www.cnblogs.com/ListenCode/p/6709145.html?utm_source=tuicool&utm_medium=referral