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

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

Redis做消息队列

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

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<Area>(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<Area>(topic);while (true){var msg = queue.TakeOne(10);if (msg != null){XTrace.WriteLine("Consume {0} {1}", msg.Code, msg.Name);}}
}

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

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

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

需要确认的队列

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

这里我们需要支持消费确认的可信队列 RedisReliableQueue<T>。消费之后，除非程序主动确认消费，否则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<Area>(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<String>(topic);while (!token.IsCancellationRequested){var mqMsg = await queue.TakeOneAsync(10);if (mqMsg != null){var msg = mqMsg.ToJsonEntity<Area>();XTrace.WriteLine("Consume {0} {1}", 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<T>，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<Area>(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<String>(topic);while (!token.IsCancellationRequested){var mqMsg = await queue.TakeOneAsync(10);if (mqMsg != null){var msg = mqMsg.ToJsonEntity<Area>();XTrace.WriteLine("Consume {0} {1}", 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<String>(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<Area>(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<String>(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<Area>();XTrace.WriteLine("Consume {0} {1}", 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