轻量级消息队列RedisQueue

消息队列（Message Queue）是分布式系统必不可少的中间件，大部分消息队列产品（如RocketMQ/RabbitMQ/Kafka等）要求团队有比较强的技术实力，不适用于中小团队，并且对.NET技术的支持力度不够。而Redis实现的轻量级消息队列很简单，仅有Redis常规操作，几乎不需要开发团队掌握额外的知识！

随着强大的.NET5发布，.NET技术栈里面怎可没有最佳的消息队列搭档？

本文从高性能Redis组件NewLife.Redis出发，借用快递业务场景，讲解.NET中如何使用Redis作为消息队列，搭建企业级分布式系统架构！

什么是消息队列

消息队列就是消息在传输过程中保存消息的容器，其核心功用是削峰和解耦！

早高峰，快递公司的货车前来各驿站卸货，多名站点工作人员使用PDA扫描到站，大量信息进入系统（1000tps），而通知快递公司的接口只有400tps的处理能力。

通过增加MQ来保存消息，让超过系统处理能力的消息滞留下来，等早高峰过后，系统即可完成处理。此为削峰！

在快递柜业务流程中，快递员投柜后需要经历扣减系统费、短信通知用户和推送通知快递公司三个业务动作。传统做法需要依次执行这些业务东西，如果其中某一步异常（例如用户手机未开机或者快递公司接口故障），将会延迟甚至中断整个投柜流程，严重影响用户体验。

如果接口层收到投柜数据后，写入消息到MQ，后续三个子系统各自消费处理，将可以完美解决该问题，并且子系统故障不影响上游系统！此为解耦！

内存消息队列

最简单的消息队列，可以由阻塞集合BlockingCollection实现

public static void Start(){    var queue = new BlockingCollection();    // 独立线程消费    var thread = new Thread(s => Consume(queue));    thread.Start();    // 发布消息    Public(queue);}private static void Public(BlockingCollection queue){    var area = new Area { Code = 110000, Name = "北京市" };    XTrace.WriteLine("Public ", area.Code, area.Name);    queue.Add(area);    Thread.Sleep(1000);    area = new Area { Code = 310000, Name = "上海市" };    XTrace.WriteLine("Public ", area.Code, area.Name);    queue.Add(area);    Thread.Sleep(1000);    area = new Area { Code = 440100, Name = "广州市" };    XTrace.WriteLine("Public ", area.Code, area.Name);    queue.Add(area);    Thread.Sleep(1000);}private static void Consume(BlockingCollection queue){    while (true)    {        var msg = queue.Take();        if (msg != null)        {            XTrace.WriteLine("Consume ", msg.Code, msg.Name);        }    }}

每秒钟生产一个消息，都被独立线程消费到。

Redis做消息队列

Redis的LIST结构，具备左进右出的功能，再使用BRPOP的阻塞弹出，即可完成一个最基本的消息队列 RedisQueue。

GetQueue取得队列后，Add方法发布消息。

TakeOne拉取消费一条消息，指定10秒阻塞，10秒内有消息立马返回，否则等到10秒超时后返回空。

public static void Start(FullRedis redis){    var topic = "EasyQueue";    // 独立线程消费    var thread = new Thread(s => Consume(redis, topic));    thread.Start();    // 发布消息    Public(redis, topic);}private static void Public(FullRedis redis, String topic){    var queue = redis.GetQueue(topic);    queue.Add(new Area { Code = 110000, Name = "北京市" });    Thread.Sleep(1000);    queue.Add(new Area { Code = 310000, Name = "上海市" });    Thread.Sleep(1000);    queue.Add(new Area { Code = 440100, Name = "广州市" });    Thread.Sleep(1000);}private static void Consume(FullRedis redis, String topic){    var queue = redis.GetQueue(topic);    while (true)    {        var msg = queue.TakeOne(10);        if (msg != null)        {            XTrace.WriteLine("Consume ", msg.Code, msg.Name);        }    }}

LPUSH 生产消息（插入列表），BRPOP 消费消息（弹出列表），因此，消息被消费后就消失了！

从日志时间可以看到，生产与消费的时间差在1~3ms之间，延迟极低！

注释消费代码后重跑，可以在Redis中看到发布的消息

需要确认的队列

如果通知快递公司的物流推送子系统处理消息时出错，消息丢失怎么办？显然不可能让上游再发一次！

这里我们需要支持消费确认的可信队列 RedisReliableQueue。消费之后，除非程序主动确认消费，否则Redis不许删除消息。

GetReliableQueue获取队列实例后，Add发布消息，TakeOneAsync异步消费一条消息，并指定10秒阻塞超时，处理完成后再通过Acknowledge确认。

public static void Start(FullRedis redis){    var topic = "AckQueue";    // 独立线程消费    var source = new CancellationTokenSource();    Task.Run(() => ConsumeAsync(redis, topic, source.Token));    // 发布消息    Public(redis, topic);    source.Cancel();}private static void Public(FullRedis redis, String topic){    var queue = redis.GetReliableQueue(topic);    queue.Add(new Area { Code = 110000, Name = "北京市" });    Thread.Sleep(1000);    queue.Add(new Area { Code = 310000, Name = "上海市" });    Thread.Sleep(1000);    queue.Add(new Area { Code = 440100, Name = "广州市" });    Thread.Sleep(1000);}private static async Task ConsumeAsync(FullRedis redis, String topic, CancellationToken token){    var queue = redis.GetReliableQueue(topic);    while (!token.IsCancellationRequested)    {        var mqMsg = await queue.TakeOneAsync(10);        if (mqMsg != null)        {            var msg = mqMsg.ToJsonEntity();            XTrace.WriteLine("Consume ", msg.Code, msg.Name);            queue.Acknowledge(mqMsg);        }    }}

LPUSH 生产消息（插入列表），BRPOPLPUSH 消费消息（弹出列表并插入另一个Ack列表），这是确保不丢消息的关键。 LREM 从Ack列表删除，用于消费完成后确认。

如果消费异常，就不会执行该确认操作，滞留在Ack列表的消息，60秒后重新回来主列表。

脑筋急转弯： 如果应用进程异常退出，未确认的消息该怎么处理？

注释消费代码后重跑，可以在Redis中看到发布的消息，跟普通队列一样，使用了LIST结构

处理“北京市”消息时，如果没有Acknowledge确认，Redis里面将会看到一个名为AckQueue:Ack:*的LIST结构，里面保存这这一条消息。所以，可信队列本质上就是在消费时，同步把消息备份到另一个LIST里面，确认操作就是从待确认LIST里面删除。

自从有了这个可信队列，基本上足够满足90%以上业务需求。

延迟队列

某一天，小马哥说，快递员投柜一定时间时候，如果用户没有来取件，那么系统需要收取超期取件费，需要一个延迟队列。

于是想到了Redis的ZSET，我们再来一个 RedisDelayQueue，Add生产消息时多了一个参数，指定若干秒后可以消费到该消息，消费用法跟可信队列一样。

public static void Start(FullRedis redis){    var topic = "DelayQueue";    // 独立线程消费    var source = new CancellationTokenSource();    Task.Run(() => ConsumeAsync(redis, topic, source.Token));    // 发布消息    Public(redis, topic);    source.Cancel();}private static void Public(FullRedis redis, String topic){    var queue = redis.GetDelayQueue(topic);    queue.Add(new Area { Code = 110000, Name = "北京市" }, 2);    Thread.Sleep(1000);    queue.Add(new Area { Code = 310000, Name = "上海市" }, 2);    Thread.Sleep(1000);    queue.Add(new Area { Code = 440100, Name = "广州市" }, 2);    Thread.Sleep(1000);}private static async Task ConsumeAsync(FullRedis redis, String topic, CancellationToken token){    var queue = redis.GetDelayQueue(topic);    while (!token.IsCancellationRequested)    {        var mqMsg = await queue.TakeOneAsync(10);        if (mqMsg != null)        {            var msg = mqMsg.ToJsonEntity();            XTrace.WriteLine("Consume ", msg.Code, msg.Name);            queue.Acknowledge(mqMsg);        }    }}

上图可以看到，每秒生产一个消息，2秒后消费到北京市，再过1秒消费到上海市（距离上海市的发布刚好2秒）。这里少了广州市，因为测试程序在生产广州市后，只等了1秒就退出。

我们从Redis中可以看到广州市这一条消息，存放在ZSET结构中。

多消费组可重复消费的队列

又一天，数据中台的小伙伴想要消费订单队列，但是不能够啊，LIST结构做的队列，每个消息只能被消费一次，如果数据中台的系统消费掉了，其它业务系统就会失去消息。

我们想到了Redis5.0开始新增的STREAM结构，再次封装RedisStream。

public static void Start(FullRedis redis){    var topic = "FullQueue";    var queue = redis.GetStream(topic);    // 独立线程消费    var source = new CancellationTokenSource();    Task.Run(() => ConsumeAsync(redis, topic, source.Token));    // 发布消息    Public(redis, topic);    //source.Cancel();}private static void Public(FullRedis redis, String topic){    var queue = redis.GetStream(topic);    queue.Add(new Area { Code = 110000, Name = "北京市" });    Thread.Sleep(1000);    queue.Add(new Area { Code = 310000, Name = "上海市" });    Thread.Sleep(1000);    queue.Add(new Area { Code = 440100, Name = "广州市" });    Thread.Sleep(1000);}private static async Task ConsumeAsync(FullRedis redis, String topic, CancellationToken token){    var queue = redis.GetStream(topic);    queue.Group = "test";    queue.GroupCreate(queue.Group);    while (!token.IsCancellationRequested)    {        try        {            var mqMsg = await queue.TakeMessageAsync(10);            if (mqMsg != null)            {                var msg = mqMsg.GetBody();                XTrace.WriteLine("Consume ", msg.Code, msg.Name);                queue.Acknowledge(mqMsg.Id);            }        }        catch (Exception ex)        {            XTrace.WriteException(ex);        }    }}

生产过程不变，消费大循环有点特别，主要是STREAM消费回来的消息，有它自己的Id，只需要对这个Id确认就可以了。

上图中，红色框是生产，紫色框是消费。

再来看看Redis中，可以看到STREAM消息还在里面。数据中台组只需要使用不同的消费组Group，即可独立消费，不用担心抢其它系统消息啦。

最佳实践

RedisQueue在中通大数据分析中，用于缓冲等待写入Oracle/MySql的数据，多线程计算后写入队列，然后由专门线程定时拉取一批（500行），执行批量Insert/Update操作。该系统队列，每天10亿条消息，Redis内存分配8G，实际使用小于100M，除非消费端故障导致产生积压。

递易智能科技全部使用可信队列 RedisReliableQueue，约200多个队列，按系统分布在各自的Redis实例，公有云2G内存主从版。积压消息小于10万时，队列专用的Redis实例内存占用小于100M，几乎不占内存空间。

公司业务每天带来100万多订单，由此衍生的消息数约1000万条，从未丢失消息！

例程代码

代码：https://github.com/NewLifeX/NewLife.Redis/tree/master/QueueDemo