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

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

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

本文例程代码需要新建控制台项目后，从nuget引用 NewLife.Redis 方能使用。

实践场景：

中通快递，RedisQueue用于大数据计算，统计数据凑批上传降低数据库落盘压力，消息量每日10亿。

递易智能，RedisQueue用于分布式系统解耦，消息Topic共300多个，涉及应用系统50多个，消息量每日1000万。

什么是消息队列

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

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

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

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

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

内存消息队列

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

public static void Start()
{
    var queue = new BlockingCollection<Area>();

    // 独立线程消费
    var thread = new Thread(s => Consume(queue));
    thread.Start();

    // 发布消息
    Publish(queue);
}

private static void Publish(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);
        }
    }
}

每秒钟生产一个消息（1号线程），都被独立线程（9好线程）消费到。

Redis做消息队列

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

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

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

注意：作为消费端实例化的Redis，需要指定大于阻塞时间的超时时间，如 new FullRedis { Timeout = 11_000 } 。

消费端实际上是大循环消费，建议每个Topic开8个线程或Task执行大循环消费，（8线程可根据业务需要调整为其它数字，如1/2/5/16），每个线程GetQueue后得到自己独占的队列实例，所有实例共同消费处理该Topic下的消息。那么消息Topic的总消费者数量就是部署实例数乘以线程数8，并且可以按需增减，没有分区倾斜问题，非常灵活。

public static void Start(FullRedis redis)
{
    var topic = "EasyQueue";

    // 独立线程消费
    var thread = new Thread(s => Consume(redis, topic));
    thread.Start();

    // 发布消息。实际上生产和消费位于不同服务器
    Publish(redis, topic);
}

private static void Publish(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 生产消息（1号线程插入列表），BRPOP 消费消息（11号线程弹出列表），因此，消息被消费后就消失了！

注：NewLife.Redis 消息队列默认对消息进行Json序列化，消费时以String消费得到消息，再反序列化为消息对象。

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

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

注意：RedisQueue的缺点是消息被消费后即从Redis中删除，如果消费端处理消息失败，将有丢失消息的风险！重要业务请使用需要确认的RedisReliableQueue队列。

需要确认的队列

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

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

RedisReliableQueue采用Redis的LIST结构，LPUSH发布消息，再使用BRPOPLPUSH的阻塞弹出，同时备份消息到挂起列表。消费成功后确认时，再从挂起列表中删除。如果消费处理失败，消息将滞留在挂起列表，一定时间后自动转移回去主队列，重新分配消费。BRPOPLPUSH确保每个消息都被消费，且仅消费一次。

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

注意：作为消费端实例化的Redis，需要指定大于阻塞时间的超时时间，如 new FullRedis { Timeout = 11_000 } 。

消费端实际上是大循环消费，建议每个Topic开8个线程或Task执行大循环消费，（8线程可根据业务需要调整为其它数字，如1/2/5/16），每个线程GetReliableQueue后得到自己独占的队列实例，所有实例共同消费处理该Topic下的消息。那么消息Topic的总消费者数量就是部署实例数乘以线程数8，并且可以按需增减，没有分区倾斜问题，非常灵活。

public static void Start(FullRedis redis)
{
    var topic = "AckQueue";

    // 独立线程消费
    var source = new CancellationTokenSource();
    Task.Run(() => ConsumeAsync(redis, topic, source.Token));

    // 发布消息。实际上生产和消费位于不同服务器
    Publish(redis, topic);

    source.Cancel();
}

private static void Publish(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 生产消息（1号线程插入列表），BRPOPLPUSH 消费消息（6号5号线程弹出列表并插入另一个Ack挂起列表），这是确保不丢消息的关键。 LREM 从Ack挂起列表删除，用于消费完成后确认。

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

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

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

处理“北京市”消息时，如果没有Acknowledge确认，Redis里面将会看到一个名为AckQueue:Ack:*的LIST结构，里面保存这这一条消息。

所以，可信队列本质上就是在消费时，同步把消息备份到另一个LIST里面，确认操作就是从待确认LIST里面删除。

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

延迟队列

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

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

消费端实际上是大循环消费，建议每个Topic开8个线程或Task执行大循环消费，（8线程可根据业务需要调整为其它数字，如1/2/5/16），每个线程GetDelayQueue后得到自己独占的队列实例，所有实例共同消费处理该Topic下的消息。那么消息Topic的总消费者数量就是部署实例数乘以线程数8，并且可以按需增减，没有分区倾斜问题，非常灵活。

public static void Start(FullRedis redis)
{
    var topic = "DelayQueue";

    // 独立线程消费
    var source = new CancellationTokenSource();
    Task.Run(() => ConsumeAsync(redis, topic, source.Token));

    // 发布消息。实际上生产和消费位于不同服务器
    Publish(redis, topic);

    source.Cancel();
}

private static void Publish(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结构中。

多消费组RedisStream

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

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

注意：作为消费端实例化的Redis，需要指定大于阻塞时间的超时时间，如 new FullRedis { Timeout = 11_000 } 。

消费端实际上是大循环消费，建议每个Topic开8个线程或Task执行大循环消费，（8线程可根据业务需要调整为其它数字，如1/2/5/16），每个线程GetStream后得到自己独占的队列实例，所有实例共同消费处理该Topic下的消息。那么消息Topic的总消费者数量就是部署实例数乘以线程数8，并且可以按需增减，没有分区倾斜问题，非常灵活。

public static void Start(FullRedis redis)
{
    var topic = "FullQueue";

    // 两个消费组各自独立消费
    var source = new CancellationTokenSource();
    {
        var queue = redis.GetStream<Area>(topic);
        queue.Group = "Group1";

        _ = queue.ConsumeAsync(OnConsume, source.Token);
    }
    {
        var queue = redis.GetStream<Area>(topic);
        queue.Group = "Group2";

        _ = queue.ConsumeAsync(OnConsume2, source.Token);
    }

    // 发布消息。实际上生产和消费位于不同服务器
    Publish(redis, topic);

    Thread.Sleep(1000);
    source.Cancel();
}

private static void Publish(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 void OnConsume(Area area)
{
    XTrace.WriteLine("Group1.Consume {0} {1}", area.Code, area.Name);
}

private static Task OnConsume2(Area area, Message message, CancellationToken token)
{
    XTrace.WriteLine("Group2.Consume {0} {1} Id={2}", area.Code, area.Name, message.Id);

    return Task.CompletedTask;
}

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