★ FEATURED ARTICLE 最近有位做电商系统的朋友来找我说他们的订单状态同步服务一到活动促销就出问题Redis里的消息越堆越多消费者进程看着像卡死了一样重启没用换了个更大的机器也撑不住。我问他用的什么队列方案他说Redis Pub/Sub。电话这边我基本就猜到问题出在哪了——不是Redis不行是压根没分清Redis、RabbitMQ、Kafka这三类消息中间件各自管的是哪块地。这篇文章我想把这件事讲透。我们在Python后端架构里聊“消息队列”的时候通常默认它们是同一个东西但Redis的Stream、RabbitMQ的AMQP协议、Kafka的日志模型从底层设计上就是三条不同的路。你选哪个不取决于哪个“更火”而取决于你的业务到底需要解耦、削峰、异步还是需要可回放的事件流。这篇文章适合正在做技术选型的后端开发也适合已经把某个中间件跑起来了、但遇到奇奇怪怪生产问题不知道怎么排查的人。1. 事故复盘Redis 里的消息越堆越多问题不在 Redis1.1 现场A同学的订单状态同步服务是怎么崩的A同学的项目并不复杂用户下单后订单服务把“订单已创建”这个事件发布出来下游有好几个服务要消费这个事件比如库存扣减、积分发放、通知推送。他一开始图省事直接用Redis的Pub/Sub通道订单服务负责publish其他服务各自subscribe。平时流量低没毛病但一到活动促销消费者处理速度跟不上发布速度他就开始收到告警消息堆积、延迟陡增、下游数据对不上。最诡异的是他看Redis内存和CPU都很健康没有OOM没有慢查询发布端也确认每次都执行成功了。但消费者就是消息时有时无重启之后有些消息彻底消失了。他以为是自己代码写错了反复看了几十遍发布代码也没找到问题。1.2 拆解Pub/Sub 模型为什么会“看起来像队列又不像队列”这个问题拆开其实很清晰。Redis Pub/Sub本质上是一个即时广播通道不是队列。发布者把消息发给频道之后Redis不会把消息存到任何地方消费者如果当时不在线或者网络抖动导致没收到这条消息就直接丢了没有任何重试、补偿、回溯机制。也就是说Pub/Sub既不做持久化也不提供确认机制它迎合的是“实时通知”场景而不是“可靠消息传递”场景。A同学看到的现象是某个消费者进程在活动开始后因为处理超时被监控系统kill掉了重启的间隙里发布端发过来的那批订单事件全部蒸发。他以为消息“堆在Redis里”其实Redis里根本啥也没有。他看到的“堆积”是另一个更隐蔽的问题——消费者虽然活着但它的CPU被某些重业务逻辑占满处理不过来而Pub/Sub又没有堆积缓冲区消息只能被网络层丢掉。这一来一回丢消息就成了必然。这个案例给了我一个很重要的判断标准选消息中间件之前先想清楚你允不允许丢消息。如果一条消息丢了会导致对账不平、用户权益损失那Redis Pub/Sub这种“即发即忘”的模型从一开始就不该出现在选项里。Redis里真正能承担可靠队列任务的是Stream这个后面会细说。2. 三个底层模型的本质区别队列、路由器、日志2.1 Redis从 List 到 Pub/Sub 再到 Stream它到底算什么Redis在消息领域其实有三副面孔。最早大家用List结构做队列LPUSH往左推BRPOP从右边阻塞弹出数据是存在Redis里的理论上不怕消费者离线。但List天然没有ACK机制消费者把消息pop走了如果处理过程中崩了这条消息就再也找不回来。后来出了Pub/Sub又把持久化这个优点给丢了。直到Redis 5.0引入了Stream才算真正补齐了消息领域的核心能力。Stream的数据结构可以理解成一条只能追加的日志每条消息都有一个自动递增的ID消费者组Consumer Group机制让多个消费者可以分工读取读取之后必须显式ACK没ACK的消息会留在Pending列表里供崩溃恢复后重新处理。它还支持从任意历史位置读取这个能力非常像Kafka的offset机制。所以你在2025年的今天如果还要在Redis上做消息队列我的建议是直接用Stream别再惦记Pub/Sub和List那套老玩法了。但Redis Stream也有它的天花板它毕竟是内存型数据库的附属品消息堆积量受内存上限约束持久化靠RDB/AOF极端场景下的数据可靠性不如专为消息设计的系统。它更适合消息量可控、不想额外引入中间件、希望一鱼两吃既当缓存又当队列的场景。2.2 RabbitMQ一个不折不扣的路由分发网络RabbitMQ的底层模型跟Redis完全不同。它不叫队列而叫Exchange Binding的路由体系。生产者不直接往队列里扔消息而是把消息发给ExchangeExchange再根据Routing Key和Binding规则把消息复制到匹配的一个或多个Queue里。消费者订阅的是QueueQueue有独立的持久化、ACK、死信、TTL策略。这个设计的好处是路由能力极强。你可以用Direct Exchange做点对点精确分发用Topic Exchange做按通配符匹配的“按业务类型订阅”用Fanout Exchange做广播。一个系统里可以同时存在多个Queue接收同一类消息但每个Queue的消费者互不干扰各自独立ACK。生产上RabbitMQ更被人看重的是它的确认机制做得非常细致生产端有Publisher Confirm消费者有Manual ACK消息没被确认就不会从Queue中删除消费者断开连接后消息会被重新投递给其他消费者。这种“宁可重复不可丢失”的模型特别适合订单系统、支付回调这类不允许丢消息的业务。RabbitMQ的吞吐量虽然不如Kafka但单机几万条每秒对于绝大多数内部业务完全够用。2.3 Kafka只追加的日志天然为回放和吞吐而生Kafka跟前面两个最大的区别在于它从设计之初就不是“队列”而是一个分布式的提交日志。消息按Topic划分每个Topic又拆成多个Partition消息永远只追加到Partition尾部消费者自己维护读取位置offset。所以Kafka里消息不会因为被消费了就消失它会在磁盘上保留你配置的时长或大小过期后才被清理。这意味着任何时刻你想让一个新消费者从三天前开始重新读一遍整个事件流都是完全可行的。正是这种“日志”特性让Kafka成了数据管道事实上的标准。日志采集、用户行为追踪、数据库变更捕获CDC、多个异构系统之间的数据同步这些场景的核心诉求不是“发条消息让下游干个活”而是“把发生过的事实完整保存下来供多个系统各取所需地消费和回放”。Kafka的单分区内保证有序全局吞吐可以靠增加分区数水平扩展单集群一天处理几百亿条消息是很常见的数字。2.4 比方说冰箱、驿站和仓库账本为了记住三者差异我常用的类比是这样的。Redis Stream像是一个带保存功能的普通队列东西放进去能存但你得自己看好容量崩溃了可能丢一部分。RabbitMQ像一个智能快递驿站它能按贴的标签把每件物品分到正确的货架上签收确认做得很好但驿站每天能处理的快件总数有上限。Kafka则像一个仓库的流水账本每一笔进出都按顺序记下来账本不会因为东西出库就撕掉所以什么时候想对账都能从头翻。这套类比基本对应了它们的真实定位Redis重“轻巧和灵活”RabbitMQ重“分发和可靠”Kafka重“吞吐和留存”。3. 硬指标与软肋对照延迟、吞吐、可靠性和有序性如何取舍3.1 一张表看清三种消息中间件的边界很多人在选型时会问“谁快谁慢”但比单一指标没意义。我干脆把常见维度整理成一张表方便对照着看。维度Redis StreamRabbitMQKafka消息模型追加日志 消费者组队列 路由规则分区日志 消费者组底层存储内存 / RDB / AOF磁盘可持久化磁盘多副本典型吞吐量万级/秒单实例上限明显万级到十万级/秒百万级/秒可水平扩展消息确认ACK但整体机制较轻量生产端Confirm 消费端Manual ACK消费者位移提交At-least-once消息重放支持从历史ID读取不支持按时间重放重入队靠死信/重新发布支持任意offset和时间回溯有序性单Stream内有序单队列单消费者有序单Partition内有序路由能力弱主要按Stream分类强Exchange可按Routing Key灵活绑定弱靠Topic和Partition Key延迟毫秒级内存型毫秒级到几十毫秒十毫秒到数百毫秒轮询拉取运维复杂度低复用Redis中需要管理RabbitMQ集群高需要管理ZooKeeper或KRaft、分区、副本Python生态redis-pypika / aio-pikaconfluent-kafka / kafka-python这张表不是我拍脑袋做的是根据我在不同项目里实际压测和线上观察总结出来的量级经验具体数值会受硬件、网络、消息大小、分区数影响但相对大小关系是稳定的。3.2 可靠性阶梯从最多丢一点到基本不丢消息丢失这件事在这三者身上有不同的含义。Redis Pub/Sub是“根本不保证不丢”Redis Stream在默认配置下如果发生故障切换AOF落盘延迟范围内的消息可能丢你必须配合WAIT和合理的持久化配置来降低风险。RabbitMQ在持久化队列加持久化消息、生产端confirm全开的情况下能实现很高的可靠性但也不是零丢失磁盘故障或者集群脑裂时依然存在窗口。Kafka依靠ISR机制和副本同步配合acksall配置在Broker正常工作的前提下能做到不丢已提交消息。我见过很多团队在选型时把“可靠性”当成一个二值变量要么觉得丢一点无所谓要么觉得必须绝对不丢。实际业务里更常见的划分是这条消息丢了损失是几块钱还是几百万。普通通知推送用Redis Stream足够订单支付回调必须RabbitMQ或Kafka级别的确认机制审计日志则只有Kafka那种留存模型能让追责变得可行。3.3 消费模型推、抢占式拉取、还是按分区协作三个系统的消费模型差异也直接影响你的代码组织方式。RabbitMQ是推模型Broker主动把消息推给消费者消费者处理完ACK多个消费者共享一个队列时消息按Round-Robin分发谁先处理完谁接下一单天然负载均衡。Redis Stream的消费者组也类似但更偏向“抢占式拉取”消费者主动从Stream里读读完ACK谁抢到归谁。Kafka则是拉模型加分区协作消费组里的每个消费者分到固定Partition同一Partition的消息只会发给同一个消费者不存在谁抢到归谁的说法。这个差异带来一个很实际的选型判断如果你的下游处理能力波动很大希望快的消费者多干点活RabbitMQ的竞争消费模式更舒服如果你的下游需要按某个业务键保证顺序比如同一个订单的所有事件必须被同一个处理节点按序处理Kafka的Partition机制天然满足Redis Stream只要保证消息ID单调递增也能在这个Stream内有序但多了消费者组设计成本。3.4 有序性全局有序本来就是伪需求居多先说结论大部分业务要的有序性其实是“同一个业务实体的事件按顺序处理”比如同一订单的创建、支付、发货、完成必须是这个顺序。这属于Partition内有序或单队列有序三者都能通过合理设计满足。Redis Stream用同一Stream并发的消费者组处理或者用Kafka把同一个订单号作为Partition Key路由到同一Partition。RabbitMQ则要保持单队列单消费者或按业务路由分散到不同队列再单消费者处理。“全局有序”几乎是个伪需求因为一旦涉及多消费者并行处理全局有序和水平扩展就是矛盾的。如果真有强制全局有序的场景比如某种只能单线程处理的批任务那选择面会非常窄大概率只能接受Redis Stream单消费者或者Kafka单分区、单消费者这种低并发方案。我在项目里遇到不合理的“必须全局有序”需求时通常的做法是先追问业务细节大多最后都能拆成局部维度。4. Python 三个可运行方案从 Hello World 到生产可用4.1 Redis Stream轻量任务队列的标准接入姿势在PyPI里装redis-py之后Stream的核心API并不复杂。我下面给的是一个生产者加一个消费者的最小闭环复制到本地改一下Redis连接就能跑通。import json import redis r redis.Redis(hostlocalhost, port6379, decode_responsesTrue) STREAM task:queue GROUP task-worker # 创建消费组。如果已存在会报 BUSYGROUP这里做幂等处理。 try: r.xgroup_create(STREAM, GROUP, id0, mkstreamTrue) except redis.exceptions.ResponseError as exc: if BUSYGROUP not in str(exc): raise def publish(payload: dict): r.xadd(STREAM, {data: json.dumps(payload)}) def consume(): while True: # 从消费组读取新消息 表示只读没投递过的消息 entries r.xreadgroup(GROUP, worker-1, {STREAM: }, count10, block5000) if not entries: continue for _stream, msgs in entries: for msg_id, fields in msgs: try: handle_task(json.loads(fields[data])) except Exception: # 处理失败的消息不要 ACK让它留在 Pending 里 continue # 处理成功才 ACK否则重启后会重新投递 r.xack(STREAM, GROUP, msg_id)消费逻辑里最关键的细节是这条处理成功之后再xack。如果先xack再处理业务业务崩了消息就没了如果不ack消息会一直在Pending列表里同一消费者恢复后还能从Pending里捞回。很多团队第一次用Redis Stream踩的坑就是漏了xack导致消费组位移一直不走消息被反复投递跟死循环一样。另外注意xreadgroup的block参数设置成5000毫秒而不是0是为了让循环每隔几秒能退出来做点别的事比如上报心跳、刷新配置写业务代码时灵活很多。4.2 RabbitMQ pika手动确认和预取限制是必须做的功课RabbitMQ在Python里最常用的库就是pika。最简单能跑的代码长这样import pika params pika.ConnectionParameters( hostlocalhost, heartbeat600, ) connection pika.BlockingConnection(params) channel connection.channel() channel.queue_declare(queuetask.queue, durableTrue) channel.basic_qos(prefetch_count1) def callback(ch, method, properties, body): try: handle_task(body) ch.basic_ack(delivery_tagmethod.delivery_tag) except Exception as exc: # 可以根据异常类型决定是否重新入队、投递到死信队列 ch.basic_nack(delivery_tagmethod.delivery_tag, requeueFalse) print(f处理失败: {exc}) channel.basic_consume(queuetask.queue, on_message_callbackcallback) channel.start_consuming()我这里特意加了两行很多教程不会提的配置durableTrue和basic_qos(prefetch_count1)。durable保证队列和消息在Broker重启后不消失prefetch_count1则告诉Broker每次只给这个消费者发一条未确认消息等它ACK之后再发下一条。没有这个限制RabbitMQ会把队列里所有消息一口气推给消费者堆在本地内存里消费者一旦崩溃这批消息全部重新入队下游会出现大面积重复消费。pika的BlockingConnection在单线程程序里很好用但如果你打算在Web框架的多线程worker里共享连接强烈建议改成线程内独立连接或者直接用aio-pika写异步版本。这不是性能问题而是pika的BlockingConnection本身不是线程安全的多个线程同时调用容易触发不可预期的状态错乱。4.3 Kafka confluent-kafka消费者组参数是第一个分水岭Python连Kafka我一般优先用confluent-kafka它是librdkafka的绑定性能比纯Python实现的kafka-python好一个量级接口也更贴近生产环境。import json from confluent_kafka import Producer, Consumer, KafkaError, KafkaException producer Producer({bootstrap.servers: localhost:9092}) def delivery_report(err, msg): if err is not None: print(f消息发送失败: {err}) else: print(f已写入 {msg.topic()} 分区 {msg.partition()} 偏移量 {msg.offset()}) def send(topic: str, key: str, value: dict): producer.produce( topic, keykey.encode(), valuejson.dumps(value).encode(), callbackdelivery_report, ) producer.poll(0) consumer_conf { bootstrap.servers: localhost:9092, group.id: task-consumer-group, auto.offset.reset: earliest, enable.auto.commit: False, max.poll.interval.ms: 300000, } consumer Consumer(consumer_conf) consumer.subscribe([task.topic]) while True: msg consumer.poll(1.0) if msg is None: continue if msg.error(): if msg.error().code() KafkaError._PARTITION_EOF: continue raise KafkaException(msg.error()) try: handle_task(json.loads(msg.value().decode())) consumer.commit() except Exception as exc: print(f处理失败: {exc})这段代码里的enable.auto.commitFalse是很多人容易忽略的重点。默认情况下Kafka消费者会每5秒自动提交一次消费位移如果你业务处理刚好在自动提交之后崩了这条消息虽然没处理成功但位移已经提交重启后就直接跳过了。关掉自动提交在handle_task成功之后再手动commit()才能保证“处理成功才算消费成功”也才能让Kafka做到常见文档里说的at-least-once语义。同理max.poll.interval.ms如果太小消费者在处理一批消息时花费时间超过这个阈值就会被判定为“失联”踢出消费者组触发一次rebalance其他消费者就会接手它的分区而这个消费者手里的消息可能还没处理完重复消费就来了。生产中这个值我通常会结合单条消息处理时长和单次拉取条数一起算给到5分钟以上是比较稳妥的起点。4.4 关于消息规范和序列化设计的补充三个系统的代码跑通之后还有一件跟具体中间件无关、但比选型本身更重要的事消息体长什么样。我强烈建议所有消息在进入队列/Stream/Topic之前都规范成类似下面这种结构{ message_id: uuid-xxx, event_type: order.created, version: 1, timestamp: 1736000000, payload: {} }message_id给幂等消费用event_type方便网关做分类version是为了将来字段演进时兼容老消费者。不要图省事直接把业务表模型序列化丢进去也不要直接pickle一个Python对象。一旦你哪天想把消费端换成另一个语言写的服务pickle就直接卡死升级路径了。5. 生产环境排错实录三类工具各自的翻车点5.1 RedisMetirc一查发现消费者在无限重启有一次帮另一个团队排查现象是Redis Stream里消息堆积持续走高消费者日志每隔几分钟就打一行异常然后进程退出被守护进程拉起。翻代码后发现他们把xreadgroup的循环写在了队列消费函数里这个函数又被Web框架的业务线程调用结果每个请求都创建一个消费者抢到一条消息处理后线程结束连接关闭没ACK的消息又回到Pending。一个消费者组里有几十个临时消费者在打架。这种问题不看监控指标很难发现因为单看Redis内存是正常的只有Stream的pending和lag指标能暴露真相。我后来养成了一个习惯接入Redis Stream的第一天就把xlen、pending、lag这三个指标接进监控大盘。xlen是总消息数pending是会重复投递的未确认消息lag是消费落后程度。三个指标放到一起消费者是死是活、消息是否重复处理、消费速度是否跟得上一眼就能看清。5.2 RabbitMQunacked堆积与prefetch的关系RabbitMQ的队列里有个状态叫unacked表示已经被推给消费者但还没收到ACK的消息。默认情况下RabbitMQ在消费者数量不足时会不停向消费者推送消息没有任何窗口限制。如果消费者处理每条消息需要几百毫秒而队列里瞬间涌入几千条消息这几千条消息会全部堆积在消费者的客户端缓冲区里表现为RabbitMQ管理界面里unacked数值飙升。很多人遇到这个第一反应是加消费者机器但其实问题根子是无界分发。你加再多的消费者只要prefetch不设限每台机器都会先被灌满一批消息再慢慢处理负载均衡完全失效。正确做法是先设置一个合理的prefetch值我一般从1开始调加并发再用多消费者或 RabbitMQ 的高性能客户端比如多channel。调到多少算合理一个经验公式是prefetch 单消费者期望的并发处理量如果你想让消费者同时处理10条消息就设为10同时做好内存允许等待消息数量的评估。5.3 Kafka一次max.poll.interval.ms的连环效应Kafka消费者翻车最经典的场景是业务方在回调里调了一个外部接口这个接口偶尔会阻塞几十秒结果单次消息处理时间超过了max.poll.interval.ms默认的5分钟消费者被判定死亡触发rebalance。rebalance之后这个分区被分配给组内另一个消费者那个消费者从头重新消费又把同样卡住的外部接口调了一遍再次超时再次rebalance。整个消费者组陷入无限rebalance的循环吞吐量掉到几乎为零而监控里看到的只有一条“消费者组正在rebalance”的告警。这种问题排查的时候去看看消费者的poll间隔和离开组日志就能定案。修复方式通常有两个方向一是调大max.poll.interval.ms二是把耗时的外部调用挪到独立线程或异步队列里让主线程持续poll保持心跳。我本人更推荐第二种因为把业务处理和消息拉取解耦能避免很多隐藏的连锁问题。5.4 通用排查方法论从指标倒推代码而不是读代码猜指标有没有一套通用的排错思路我在三个系统上踩完坑之后总结下来其实是同一套打法先看消费位点相关的指标再结合日志定位消费者状态最后才看业务代码。Redis看pending和lagRabbitMQ看ready和unackedKafka看consumer lag和rebalance次数。指标能告诉你是“消息没到消费者手里”还是“到了但处理失败”这两种情况的排查路径完全不同。不要一上来就翻代码也不要一上来就怪中间件。我见过太多团队把消息中间件的参数和配置调来调去结果最后发现是业务代码里有条慢SQL让消费者线程卡住了。中间件是背锅侠但它也只负责把消息可靠送到不负责你的处理器是否高效。6. 选型决策把“哪个火”换成“哪个合适”6.1 一张决策表直接照着选综合前面的底层模型、运行指标和实战坑点我最后给一套可以直接套用的选型建议。你的核心诉求推荐方案一句话理由异步任务、简单解耦、消息量级小、不想新增部署组件Redis Stream最轻量复用已有Redis够用且省心订单/支付等核心业务可靠性和路由能力要求高RabbitMQACK机制完善Exchange模型灵活适合内部系统日志采集、行为事件、CDC、多系统数据同步Kafka高吞吐加日志留存天然支持重放和多订阅者已有Redis临时需要发布订阅通知且允许少量丢失Redis Pub/Sub仅限实时通知不能当可靠队列用团队运维能力有限但又要高可靠消息RabbitMQ比Kafka好运维得多生态文档丰富消息量会随业务爆发式增长需要水平扩展Kafka分区机制就是为大规模并行设计的未来需要按时间回溯消费历史数据Kafka消息保留期内可任意回放Redis和RabbitMQ都较难这张表解决的是“主选型”。很多团队最后会发现系统里同时存在两种中间件是常态比如用Redis Stream处理轻量异步任务用Kafka处理核心事件流。中间件不是单选题混用也没问题只要维护边界清晰。6.2 延迟队列、事务消息、顺序消息怎么落地进阶需求里最常见的三个是延迟队列、事务消息和顺序消息。延迟队列在RabbitMQ里可以通过DLX加TTL实现消息先发到一个带TTL的队列过期后自动转入真正的业务队列。Redis可以用一个ZSet模拟延迟队列score存执行时间戳定时任务取到期的消息。Kafka没有内建的延迟消息通常做法是拆成多个延迟分级Topic比如delay-5m、delay-30m由定时任务扫描转发。事务消息这里要提醒一下消息队列的事务和数据库事务不是一个东西。RabbitMQ的Publisher Confirm配合生产者本地事务表可以做到“业务操作和发消息在同一事务控制下”Kafka有事务性API可以实现对多个分区的原子写入Redis Stream则没有类似事务语义如果你需要事务型消息还是不要选择Redis做核心通道。顺序消息前面提过Kafka靠Partition Key解决RabbitMQ靠单队列单消费者解决Redis Stream靠单个Stream内顺序消费解决。没有银弹需求里必须想清楚顺序维度是什么然后把这个维度变成中间件的路由键。6.3 架构师的真正功课幂等、监控和演进空间选型只是起点。无论你最后选了哪个消息队列都必须把下面三件事一起做了不然迟早会在生产环境交学费。第一件是消费幂等。几乎所有消息中间件在异常场景下都只会给你at-least-once语义重复消费不是偶然而是常态。常用做法是在消费端拿message_id去查重比如Redis的SETNX或者数据库唯一索引。幂等设计一定要在消息规范阶段就定下来不要在出问题后再补。第二件是监控度量。消息积压指标要在上线第一天就接入监控告警而不是等业务报警了再去找。消费lag超过一定阈值就该告警这个阈值可以根据业务容忍度来定比如订单事件延迟超过10分钟就要介入。第三件是演进空间。消息体带版本号消费端预留字段扩展消息路由不要写死。这样将来哪怕要从Redis Stream迁移到RabbitMQ或Kafka也只需要在网关层做一次转发业务代码改动最小。我见过太多系统因为消息体没有版本概念升级时所有消费者被迫同时发版零发布窗口的痛谁排过谁知道。最后再分享一点经验做了这么多年后端我的体会是消息队列选型最大的敌人不是性能不够而是“拿复杂工具解决简单问题”和“拿简单工具解决复杂问题”这两件事反着来。Redis Stream在大多数中小项目里完全够用RabbitMQ能扛住绝大多数业务的可靠性要求Kafka也只有在数据量或重放需求真正出现时才值得引入那套更重的运维体系。如果你还在纠结不妨先做一次最小验证把同一条业务链路分别用Redis Stream和RabbitMQ各实现一遍跑一天看看吞吐、堆积和排查问题时的心智负担差异有多大。多数情况下你心里自然就有答案了。
阅读完成 · 觉得有帮助？
