实际业务里我见过太多接口里同步处理耗时任务的案例请求进来业务逻辑直接去发邮件、生成报表、调用第三方接口用户在前端一直转圈等结果。流量小的时候没感觉等并发稍微上来接口就开始成片超时。后来我在项目里逐步引入生产者消费者模式发现它解决的核心问题就一句话——让任务的“生产”和“消费”解耦中间放一个队列做缓冲。生产者负责把任务推到队列里消费者按自己的节奏从队列里取出来处理两边不直接关联也不互相等待。这篇文章就聊两个落地方向一是C#里用内存队列实现这个模式二是用Redis队列实现跨进程、跨机器的版本。适合正在做异步化、削峰填谷、任务调度或者想把高吞吐写入和慢消费处理拆开的开发者参考。1. 先理清生产者和消费者到底解决了什么问题1.1 模式的核心思想是什么把生产者消费者模式拆开看只有三个角色生产者、队列、消费者。生产者的职责是产生数据消费者的职责是消费数据队列夹在中间做存储。这里的关键在于生产者和消费者不是一一对应、也不是同步调用的关系。生产者往队列里塞数据不需要关心消费者此刻是否空闲消费者从队列里取数据也不需要关心数据是哪一刻被生产的。这是解耦的第一个层面——时间上的解耦。第二个层面是速率上的解耦。生产速度可能忽高忽低消费速度相对稳定队列天然充当缓冲池把两边的速率差吸收掉。这个思想在生活里很常见。比如餐厅里后厨相当于生产者服务员相当于消费者出餐口的保温台就是队列。顾客点单高峰期后厨拼命出菜服务员不可能同时把所有菜都端走保温台就起到了缓冲作用。如果没有这个缓冲后厨就必须和服务员一对一配合效率只会更低。1.2 哪些场景真正需要它先说结论不是所有业务都适合引入这个模式引入之前要判断是否具备以下典型特征。第一种是任务处理耗时明显高于任务产生耗时的场景。比如下单业务接收请求本身很快但后续要发通知、更新搜索索引、同步外部系统这些操作耗时远高于请求写入本身。如果全部放在请求线程里同步执行用户会一直等。把这些任务丢进队列让消费者异步处理是最常规的解法。第二种是存在突发流量、需要削峰的场景。比如数据采集服务平时每分钟几千条大促或活动日可能瞬间涌进来几万条。消费者的处理能力是固定的那就让数据在队列里排队空闲时慢慢消费避免把下游系统打垮。队列的容量本身就是天然的流量缓冲。第三种是需要保证任务不丢失、或者需要重试机制的场景。内存队列适合进程内部的任务分发Redis队列因为数据持久化在外部存储中即使消费者进程崩溃任务还在Redis里重启后可以继续消费。这正好对应到本文的两个实现方向。用不上的场景也有实时性要求极高且任务本身很轻的调用或者要求同步拿到处理结果的事务性操作强行加队列反而拖延链路还会引入额外复杂度。判断标准很简单任务能不能容忍延迟消费者处理失败后能不能接受重试。2. 内存队列实现从BlockingCollection到Channel2.1 BlockingCollection最经典的阻塞队列在C#里最早、使用最广的内存队列是BlockingCollection。它本质上是生产者消费者模式的标准封装生产者调用 Add 往里放数据消费者调用 Take 往外取数据。队列为空时Take 会阻塞当前线程直到新数据进来达到容量上限时Add 会阻塞生产者防止无限制堆积。一个最小实现长这样using System.Collections.Concurrent; var queue new BlockingCollectionstring(boundedCapacity: 1000); // 生产者 Task.Run(() { for (int i 0; i 100; i) { queue.Add($任务-{i}); } queue.CompleteAdding(); // 标记不再添加 }); // 消费者 Task.Run(() { foreach (var item in queue.GetConsumingEnumerable()) { Console.WriteLine($消费: {item}); } });几个关键点值得单独说。首先是 boundedCapacity 有界容量如果不设置默认是无界队列。生产速度一直大于消费速度时内存会被任务对象塞满最终 OOM这是新手最容易踩的坑。建议任何生产环境都设置上限。其次GetConsumingEnumerable 会随着队列变空自动阻塞并且在调用 CompleteAdding 之后会安全地遍历完剩余元素再退出循环避免消费者线程永远挂起。BlockingCollection 底层支持多种集合实现默认是 ConcurrentQueue也就是先进先出。也可以换成 ConcurrentStack 做后进先出或者 ConcurrentBag 做无序处理只需要在构造函数里传入不同实现。实际业务里九成以上都用先进先出默认就好。2.2 Channel现代异步首选相比 BlockingCollectionChannel 是更现代的方案也是我在新项目里的首选。它的设计目标是高性能异步生产者消费者生产者和消费者都可以是异步方法核心用法是在内存里维护队列通过 WaitToReadAsync / ReadAsync 异步等待数据。using System.Threading.Channels; var channel Channel.CreateBoundedstring(new BoundedChannelOptions(1000) { FullMode BoundedChannelFullMode.Wait, SingleReader false, SingleWriter false }); // 生产者 async Task ProduceAsync(ChannelWriterstring writer) { for (int i 0; i 100; i) { await writer.WriteAsync($任务-{i}); } writer.Complete(); } // 消费者 async Task ConsumeAsync(ChannelReaderstring reader) { await foreach (var item in reader.ReadAllAsync()) { Console.WriteLine($消费: {item}); } }Channel 比 BlockingCollection 强在哪第一它原生支持 async/await 的完整流程而 BlockingCollection 的 Take 是同步阻塞的在 ASP.NET Core 这种线程池线程有限的环境里长期阻塞会加剧线程饥饿。第二Channel 内部针对单写单读场景做了大量优化如果业务上只有一个生产者、一个消费者性能会非常可观。第三它在背压处理上提供了更细致的策略。这里必须解释一下 FullMode。当生产速度远大于消费速度时有界 Channel 会触发背压策略BoundedChannelFullMode.Wait生产者异步等待直到队列有空位。这是最常用的模式适合严格控制内存占用、又不允许丢弃任务的场景。DropWrite新写入的数据被丢弃适合允许丢弃非关键数据的场景。DropNewest丢弃队列中最新的数据保留旧数据。DropOldest丢弃最旧的数据保留新数据适合保证数据新鲜度的场景。选择时核心判断依据只有一个任务能不能丢。能丢就按数据优先级选丢弃策略不能丢就选 Wait。2.3 两种方案的对比与取舍直接给结论如果代码整体是同步的BlockingCollection 简洁直观尤其适合传统多线程场景如果项目已经是异步风格或者跑在 ASP.NET Core 里无脑选 Channel几乎没有理由回头。对比维度上BlockingCollection 胜在简单、底层集合可替换、阻塞语义直白适合快速实现Channel 胜在性能、异步友好、背压策略丰富适合长期维护。两者在功能上都覆盖了“有界阻塞队列”这个核心模型但 Channel 的设计更贴近现代 .NET 的异步生态。还有一点经常被人忽略Channel 的 SingleReader / SingleWriter 参数。如果确认只有一个消费者把 SingleReader 设为 trueChannel 内部会走一条无锁的快速路径性能还能再提一截。但一旦实际上有多个消费者读同一个 Channel这个参数设错会导致行为异常。所以保守起见不确定时保持 false。3. Redis队列实现什么时候必须上Redis以及怎么上3.1 进程内队列的天然边界内存队列虽然快但生命周期和进程一致。进程一重启队列里的任务全部丢失。多个应用实例之间内存队列互相不可见。比如你在两台机器上部署同一个服务A机器的队列里有100个任务B机器的消费者完全感知不到。这时候就需要把队列放到外部存储里。Redis 作为高性能键值存储提供了 List、Streams 等数据结构天然适合做队列。而且 Redis 支持多客户端并发读写多个消费者可以同时消费同一个队列这正是分布式任务分发的基础。所以结论很简单单体应用、单进程、任务允许丢失内存队列就够了多实例部署、跨进程、任务需要保留优先考虑 Redis 队列。3.2 基于Redis List做队列LPUSH BRPOP 的组合拳Redis List 做队列最经典的组合是生产者用 LPUSH 把任务推到列表头部消费者用 BRPOP 从列表尾部阻塞式弹出。LPUSH queue:task task-1 BRPOP queue:task 0BRPOP 的第二个参数是超时时间0 表示永久阻塞。它跟 RPOP 的区别在于列表为空时 BRPOP 会一直等待而不是立刻返回 null。这个特性在消费者端非常关键消费者线程不用轮询Redis 会在新数据到达时把结果返回给客户端。C# 端用 StackExchange.Redis 实现大概是这样using StackExchange.Redis; var redis ConnectionMultiplexer.Connect(localhost:6379); var db redis.GetDatabase(); // 生产者 await db.ListLeftPushAsync(queue:task, task-1); // 消费者 while (true) { var item await db.ListRightPopAsync(queue:task, TimeSpan.FromSeconds(30)); if (!item.IsNull) { // 处理任务 Console.WriteLine(item); } }使用中有个问题必须先想清楚BRPOP 弹出任务的同时任务已经从 Redis 里消失了。如果消费者在处理过程中崩溃这个任务就永久丢失。这是 List 做队列最脆弱的一环解决办法下面单独说。3.3 可靠消费与消息确认BRPOPLPUSH 方案解决任务丢失的标准做法是 BRPOPLPUSH消费者从主队列尾部弹出任务的同时把任务副本推入一个备份队列。处理完成后再从备份队列里删除这条任务。如果消费者中途崩溃任务还留在备份队列中其他消费者可以捞回来重新处理。StackExchange.Redis 对应命令可以这样封装// 从主队列尾部弹出并备份到备份队列 var item await db.ListRightPopLeftPushAsync(queue:task, queue:task:pending, TimeSpan.FromSeconds(30)); // 处理完成确认 await db.ListRemoveAsync(queue:task:pending, item);这个方案的可靠性还体现在细节上如果消费者在处理任务后没有及时从备份队列移除会产生重复任务。所以下游处理逻辑必须设计成幂等。比如任务里带着唯一业务ID处理前先查询本地缓存或数据库判断是否已经处理过。Redis List 做队列本身不提供精确的一次消费语义我们只能尽量做到“至少一次”幂等兜底才是关键。3.4 Redis Streams比 List 更完善的队列方案Redis 5.0 引入的 Streams 是官方推荐的队列数据结构它解决了 List 做队列时的好几个痛点。最关键的改进有支持消费者组。一组消费者共同消费一个 Stream每条消息只被组内一个消费者消费天然实现负载均衡。支持消息确认。消费者读取消息后需要显式 XACK未确认的消息会留在 Pending Entries List 里可以随时查看和处理。消息持久化在 Redis 中重启不丢失。支持 Pending 超时重试XAUTOCLAIM 可以把超时未确认的消息重新分配给其他消费者。一个最基本的 Stream 消费流程XADD queue:stream * task 1 XGROUP CREATE queue:stream worker 0 XREADGROUP GROUP worker consumer1 COUNT 10 BLOCK 5000 STREAMS queue:stream C# 里对应的操作// 添加消息 await db.StreamAddAsync(queue:stream, new NameValueEntry[] { new(task, 1) }); // 创建消费者组 await db.StreamCreateConsumerGroupAsync(queue:stream, worker, 0); // 读取消息 var messages await db.StreamReadGroupAsync(queue:stream, worker, consumer1, count: 10); // 确认消息 foreach (var message in messages) { // 处理业务 await db.StreamAcknowledgeAsync(queue:stream, worker, message.Id); }Streams 确实更完善但也更复杂。我的建议是如果只是简单任务队列、要求快速落地用 List BRPOPLPUSH 就够了如果任务还需要多消费者组、失败重试、消息确认、消费进度管理这些正经消息队列特性直接上 Streams别折腾。注意 XGROUP CREATE 在流不存在时会报错通常需要先判断流是否存在或者在创建时使用 MKSTREAM 参数让 Redis 自动创建空流。4. 实操对比手写一个统一的队列封装4.1 设计一个简单但可扩展的接口既然内存队列和 Redis 队列都做同一件事那从业务层复用角度我建议先抽象一个接口上层业务只面向接口编程底层实现随时切换。public interface IMessageQueue { Task PushAsync(string message, CancellationToken ct default); Taskstring? PopAsync(CancellationToken ct default); Task AckAsync(string message, CancellationToken ct default); }这个接口有几个取舍点。PushAsync 负责推送PopAsync 负责取出内存实现返回 null 表示暂时没有消息AckAsync 负责确认内存队列里确认是空操作Redis Streams 里是 XACKList 方案里是移除备份。这种设计让上层代码完全感知不到底层是内存还是 Redis后续从单机迁移到多实例也不需要改业务代码。4.2 内存队列封装基于Channel的实现基于 Channel 的实现非常短核心就是把 Channel 的读写包装成接口定义的方法。public class ChannelMessageQueue : IMessageQueue { private readonly Channelstring _channel Channel.CreateBoundedstring( new BoundedChannelOptions(1000) { FullMode BoundedChannelFullMode.Wait }); public Task PushAsync(string message, CancellationToken ct default) _channel.Writer.WriteAsync(message, ct).AsTask(); public async Taskstring? PopAsync(CancellationToken ct default) { var result await _channel.Reader.WaitToReadAsync(ct); if (!result) return null; return await _channel.Reader.ReadAsync(ct); } public Task AckAsync(string message, CancellationToken ct default) Task.CompletedTask; }这里需要注意一个细节WaitToReadAsync 返回 false 表示 Channel 已经关闭且数据读尽这时候应该让消费者退出。如果只是简单的单消费者循环也可以直接用 ReadAllAsync 遍历。实现里的 PopAsync 每次只取一条是为了配合多个消费者并行消费的模型每个消费者任务循环里不断调用 PopAsync。4.3 Redis队列封装基于Streams的实现对应的 Redis 封装会多一些代码因为要处理消费者组和确认逻辑。public class RedisStreamMessageQueue : IMessageQueue { private readonly IConnectionMultiplexer _redis; private readonly string _stream; private readonly string _group; private readonly string _consumer; public RedisStreamMessageQueue(IConnectionMultiplexer redis, string stream, string group, string consumer) { _redis redis; _stream stream; _group group; _consumer consumer; } public async Task PushAsync(string message, CancellationToken ct default) { var db _redis.GetDatabase(); await db.StreamAddAsync(_stream, new NameValueEntry[] { new(data, message) }); } public async Taskstring? PopAsync(CancellationToken ct default) { var db _redis.GetDatabase(); var result await db.StreamReadGroupAsync(_stream, _group, _consumer, count: 1, block: TimeSpan.FromSeconds(2)); if (result.Length 0) return null; var entry result[0]; return entry.Values.FirstOrDefault(v v.Name data).Value.ToString() ?? ; } public async Task AckAsync(string message, CancellationToken ct default) { // 消息确认需要把原消息ID关联回来这里示例里先省略 } }这里必须坦白一个设计缺陷AckAsync 只靠 message 字符串没法定位到 Redis Stream 里的具体 EntryId示例里没法直接确认。实际项目中我更建议接口里的返回值不要只返回 string而是定义一个包含 message 和 messageId 的包装对象不然确认环节根本没法做。这是一个很容易在设计阶段埋下的坑。更好的做法是定义这样一个类型public record QueueMessage(string Id, string Body); public interface IMessageQueue { Task PushAsync(string message, CancellationToken ct default); TaskQueueMessage? PopAsync(CancellationToken ct default); Task AckAsync(QueueMessage message, CancellationToken ct default); }这样内存实现里 Id 可以是空字符串或自增编号Redis Streams 实现里 Id 就是 Stream 的 entryId确认时才不会抓瞎。4.4 性能实测与参数调优我用一个模拟场景分别测过两种方案生产者每秒写入 2000 条消息消费者处理每条消息耗时 2 毫秒模拟业务逻辑。内存 Channel 方案的吞吐量非常可观单消费者轻松每秒处理 10000 条以上整个链路延迟基本在亚毫秒级别。瓶颈几乎都在消费者业务本身而不在队列。调 Channel 容量时我的经验值是容量放到预期峰值消息量的 1 到 2 倍即可太小会导致生产者频繁等待太大则浪费内存。Redis Streams 方案吞吐量明显受网络和 Redis 实例性能影响。本地回环地址下单消费者每秒大约 5000 条左右如果 Redis 在远程机器这个数字会缩水到 1000 到 2000 条。所以 Redis 队列适合任务量没有高到每秒几万级但对可靠性和跨实例有要求的场景。几个实测调优参数Redis 连接池连接数是关键StackExchange.Redis 的默认配置基本合理但高吞吐场景建议打开 ForceMultiplexing 或者调整 AbortOnConnectFail避免连接成为瓶颈。消费端的批量读取比逐条读取明显更快。一次 StreamReadGroup 拿几十条批量处理再批量 ACK比一次一条快不少。消费者的循环里建议使用阻塞读取避免 Redis 空转查询。比如 BLOCK 时间设置 2 到 5 秒没有消息时让客户端挂在 Redis 端等待而不是频繁发命令。5. 常见问题与排查技巧实录5.1 内存队列的典型坑第一个坑是任务积压导致内存暴涨。很多新手为了省事使用无界 BlockingCollection 或 Channel 的 Unbounded生产速度一上来内存直线增长。排查方法很直接看进程内存曲线如果持续上涨且 GC 无法回收多半就是队列里积压的任务对象。解决方式是改成有界队列并启用背压。宁可让生产者慢一点也不能让内存爆掉。第二个坑是消费者线程泄漏。如果用 BlockingCollection 时调用了 GetConsumingEnumerable但忘记在生产完成后调用 CompleteAdding消费者线程就会永远阻塞等待。这在线程池场景里会慢慢吃光线程表现为应用越来越卡、吞吐下降但 CPU 不高。排查时抓线程栈通常会看到大量线程停在 Take 或 MoveNext 的位置全都在等同一个队列。这是很典型的定位思路。第三个坑是 Channel 的 SingleReader 参数误用。前面说过这个参数是性能优化开关但它不是“断言”没有运行时校验。设置成 true 之后如果有多个消费者并发读Channel 不会报错但内部行为可能错乱比如数据混乱或丢失。这个现象比较隐蔽容易定位很久才发现是参数问题。所以多消费者场景一定要显式设置为 false。5.2 Redis队列的典型坑Redis List 方案最常见的坑就是消费端崩溃导致消息丢失。前面说过用 BRPOPLPUSH 做备份队列能显著降低丢失概率。但还有个相关细节备份队列里的堆积任务没人处理怎么办建议单独做一个巡检任务每隔一段时间扫描备份队列长度超过阈值就告警避免任务在备份队列里躺太久无人问津。第二个坑是消费组重复消费。由于 XREADGROUP 读取后如果不及时 XACK消息一直处于未确认状态再次读取时或者 XAUTOCLAIM 触发后会重新分配给消费者。这本身是特性不是 Bug。真正的坑在于业务处理逻辑没有做幂等导致重复消费引发数据重复。解决思路只有一条下游处理尽量幂等带上订单号、业务ID重复消息直接忽略。第三个坑是 Redis 本身宕机。即使 Streams 比 List 可靠Redis 主节点挂掉且未完成主从切换时队列依然会暂时不可用。如果业务对可用性要求很高需要额外考虑 Redis 高可用方案比如哨兵或集群。同时消费端的循环要有完善的异常重试逻辑不能一拿到连接异常就把整个消费者停了。StackExchange.Redis 里连接恢复后会自动重连但消费者的循环要注意捕获 ConnectionMultiplexer 抛出的异常并继续运行。5.3 生产环境的避坑清单我整理一份自己在项目中沉淀的检查清单照着核对基本能避开大多数问题内存队列一律使用有界容量根据峰值消息量设置 1 到 2 倍缓冲。消费者处理逻辑必须捕获异常不能让单条消息异常导致消费线程死掉。处理失败的消息要进入重试或死信通道。使用 Redis 队列时优先 Streams 而不是 List除非团队对 List 方案已经非常熟。任何队列消费端都要做幂等设计QueueMessage 里应该有一个全局唯一的业务ID。生产前确认 Redis 连接配置的 AbortOnConnectFail 行为避免服务启动时短暂网络问题导致连接失败引发连锁故障。监控指标提前埋好队列长度、消费延迟、堆积任务数、消费失败数。没有监控的队列等于盲跑。Streams 可以用 XLEN 查看长度用 XPENDING 查看未确认消息数这些都要接到告警里。最后分享一个我自己的使用习惯。在单体应用内部处理批量数据、事件分发这种场景我会优先用 Channel 内存队列简单、快、没有外部依赖。一旦系统需要横向扩展、或者任务需要具备可恢复性就切换到 Redis Streams。两种技术并不冲突它们解决的是不同阶段的问题。踩过几次坑之后我才意识到选型前先想清楚任务能不能丢、是否需要跨实例消费、延迟容忍度是多少这比纠结某个 API 的用法重要得多。队列本身只是工具真正决定系统稳定性的是生产者消费者的节奏配合和失败兜底设计。