企业官网建设流程全解析

最近又有个在创业公司带团队的朋友找我聊老问题：项目里需要一个轻量级的消息队列，又不想为了这个专门部署一套独立的中间件，他问我，直接用Redis实现消息队列到底靠不靠谱？

这个问题我前后回答过几十次，也在好几个项目里亲手落地过。先说结论：Redis实现消息队列完全可行，但可靠性和吞吐边界跟方案选择强相关。如果你要的是不丢消息、可回溯、有确认机制，直接上Stream；如果只是进程间传个数据，能容忍偶发丢失，List就够了。至于Pub/Sub，我一般不建议拿它当消息队列用，后面会说原因。

这篇文章会把Redis做消息队列的几种方式全部拆开讲，包括核心命令原理、可靠性设计、参数设置、代码示例，还有我实际踩过的坑。内容偏实战，想看原理的直接看第1章，想抄作业的从第2章开始，遇见问题可以直接跳到第4章排查。

1. 方案选型：先搞清楚Redis队列的适用边界

1.1 三种实现方式的使用场景差异

用Redis做消息队列，翻来覆去就是三种姿势：List、Pub/Sub、Stream。很多人上来就LPUSH+BRPOP一把梭，不是说不行，而是你得知道每种方式的边界在哪里。

List方案的本质是链表。生产者用LPUSH把消息塞进队列头部，消费者用BRPOP从队列尾部阻塞弹出，天然就是先进先出。为什么用BRPOP而不是RPOP？因为RPOP在没有消息时会立刻返回空，消费者只能空轮询，对Redis的压力大且实时性差。BRPOP则会阻塞住，一直等到有消息过来或者超时，这才能实现类似“订阅推送”的效果。

Pub/Sub方案走的是发布订阅模型，PUBLISH发到频道，SUBSCRIBE订阅接收。它是个广播模型：所有订阅了频道的客户端都会收到消息。问题也很致命：消息不落盘、没有确认机制、消费者离线期间的消息直接丢。真要拿这个当业务消息队列用，早晚出事。它适合做那种丢了也无所谓的实时通知，比如前端刷新、配置变更广播，但不适合做任务分发。

Stream方案是Redis 5.0引入的数据结构，可以理解为官方正式的消息队列形态。它提供了全局递增消息ID、消费者组、ACK确认机制、Pending列表，甚至还有消息转移命令。从功能完整度来说，Stream已经补上了List和Pub/Sub的所有短板。如果你的Redis版本5.0以上，新项目直接用Stream，别纠结。

1.2 什么时候选Redis，什么时候换专业MQ

很多架构评审会问：Redis能做消息队列，为什么还要引入Kafka这类专业组件？我一般会先拉一张对比表出来，然后再讲场景。

我自己的判断标准很简单：如果你只有一个业务系统内部需要解耦，比如处理完订单后发个通知、生成个缩略图、推一条短信，消息量一天也就几十万条，那完全没必要为这个上专业MQ。一个Redis实例就能扛住，运维成本几乎为零。

但如果你的系统对“消息一条都不能丢”有硬性要求，或者需要消息堆积几亿条不丢，再或者多个团队多个系统之间要做事件总线，那还是老老实实引入专业消息队列。硬让Redis去顶这种场景，最后八成是在内存和可靠性上出问题。

1.3 可靠性上限由什么决定

即便用了Redis，很多人还是会有一个错觉：Redis把数据放内存里，不就可能丢吗？也不一定。可靠性上限其实由三个因素决定。

第一个是持久化配置。Redis默认的RDB快照是全量快照，两次快照之间的数据在宕机时会丢。真要拿Redis做消息队列，建议至少开启AOF：

appendonly yes appendfsync everysec

everysec是每秒刷盘一次，丢数据的窗口控制在1秒以内。如果业务要求更高，可以用always模式，每条命令都刷盘，但性能会明显下降。做消息队列一般everysec就够了。

第二个是ACK机制。List根本没有ACK，消息一BRPOP出来就没了，消费者处理时挂了就彻底丢。Stream有ACK，可以做到“至少一次”语义。如果你连至少一次都不想要，就自己接口做幂等。

第三个是消费者侧的异常处理。很多人忽略这一点，框架把消息丢给了业务线程，业务代码抛异常没人管，消息也没回滚。我在项目里见过太多这种静默丢消息的问题。后面第4章会展开讲排查方法。

2. List实现：最经典也最容易踩坑的队列方案

2.1 核心命令与交互模型

List方案的核心命令就两对，生产消费方向可以互换：

# 生产者：从左侧写入 LPUSH queue:task "send_email_1001" # 消费者：从右侧阻塞弹出 BRPOP queue:task 0

BRPOP的第二个参数是阻塞超时秒数，0表示一直阻塞直到有消息。它返回的不是一个字符串，而是一个数组，里面有队列名和消息内容：

1) "queue:task" 2) "send_email_1001"

为什么要返回队列名？因为BRPOP支持同时监听多个队列，你可以一次传入多把钥匙：

BRPOP queue:high queue:normal queue:low 5

这样就能实现简单的优先级效果：当多个队列同时有数据时，Redis会按参数顺序优先取靠前的队列。我用这个特性搞过一个两级队列，紧急任务走queue:high，普通任务走queue:normal，消费者代码不用额外改，天然就按优先级消费了。

多个消费者并发消费时，每个消费者执行自己的BRPOP，Redis会保证一条消息只被一个消费者取走，不需要自己处理锁和竞争，这一点比很多新手想象中要简单。

2.2 消息可靠性怎么补：备份队列与重新入队

List最大的坑在于没有ACK。消息被BRPOP取走的那一刻，它在Redis里就消失了。如果消费者在执行业务逻辑时宕机，这条消息就永远找不回来。

要补救，得靠备份队列和重新入队机制。我落地过的模式是这样的：

1. 消费者BRPOP取出消息。
2. 立刻把这条消息LPUSH到另一个队列，叫queue:processing。
3. 业务逻辑处理完成之后，从queue:processing中移除这条消息。
4. 如果业务逻辑迟迟没有完成，比如超过5分钟，一个巡检线程就把queue:processing里的消息重新放回queue:task。

这里有一个细节：从queue:processing重新放回queue:task，这两步操作要保证原子性，否则可能出现重复入队或消息丢失。原生命令不是原子的，得用Lua脚本：

local msg = redis.call('RPOP', KEYS[1]) if msg then redis.call('LPUSH', KEYS[2], msg) end return msg

这段脚本的意思是：原子性地从processing队列尾部取出消息，再放回task队列。为什么用RPOP而不是LPOP？因为消息当初是从左塞进processing的，右出才是先进先出。

这个方案不完美，至少有两个代价：一是每条消息要多一次额外存储和查询，二是消费者崩溃后，消息会延迟一段时间才被重新消费，实时性受影响。但它确实能把List方案的可靠性底裤补上。我自己更推荐的做法是直接用Stream，省掉这套手工机制。

2.3 实操示例：任务分发器代码走一遍

这里给一段任务分发器的核心代码，用Java写的伪代码形式，主要讲清楚生产者和消费者的骨架。

生产者端：

public class TaskProducer { private final JedisPool jedisPool; public void produce(String queueName, String taskJson) { try (Jedis jedis = jedisPool.getResource()) { jedis.lpush(queueName, taskJson); } } }

消费者端：

public class TaskConsumer implements Runnable { private final JedisPool jedisPool; private volatile boolean running = true; @Override public void run() { while (running) { try (Jedis jedis = jedisPool.getResource()) { // 阻塞10秒，超时后继续循环，避免连接长时间挂起 List<String> result = jedis.brpop(10, "queue:task"); if (result == null) { continue; } String taskJson = result.get(1); // 先把消息放入处理中队列，作为备份 jedis.lpush("queue:processing", taskJson); try { handleTask(taskJson); // 业务处理完成后，把消息从processing队列中移除 jedis.lrem("queue:processing", 1, taskJson); } catch (Exception e) { // 记录日志，消息留在processing里等待巡检重新入队 } } catch (JedisConnectionException e) { // 连接断开要重试，不能直接退出循环 sleep(1000); } } } }

这段代码里有几个容易踩的点。

连接必须是每次从JedisPool拿新的，不能复用同一个连接做BRPOP。因为客户端长时间阻塞时，连接可能被Redis服务端断开，或者被网络环境中间设备回收。一旦断开，复用连接就会出现假死。

BRPOP超时时间不要设成0，也就是无限阻塞。虽然命令本身支持0，但在生产环境里，客户端连接会因为各种原因被服务端回收，无限阻塞的连接断了自己都不知道，表现就是消费者莫名其妙不消费了。我会把超时设为5到10秒，配合循环重试，用一个很短的时间窗换取连接的自我修复能力。

消费者宕机后，消息会滞留在queue:processing里。巡检线程可以这样写：用LRANGE扫描processing队列，对比消息入队时间，超过阈值再重新入队。这个方案虽然简陋，但我在真实环境下运行过，能撑住几万条日消息量。

2.4 性能参考与参数设置建议

List方案的性能取决于Redis单实例的极限。我自己在普通云主机上压测过，单条消息1KB左右时，LPUSH能跑到每秒几万次，BRPOP消费也能跑到类似量级。瓶颈基本不在Redis本身，而在网络带宽和客户端连接数量。

如果单次批量发送大量消息，建议用pipeline：

Pipeline pipeline = jedis.pipelined(); for (String message : messages) { pipeline.lpush("queue:task", message); } pipeline.sync();

这样能把客户端与Redis之间的网络往返次数从N次降到1次，吞吐提升非常明显。

参数设置上有几个要点：

• Redis的maxmemory-policy不要用allkeys-lru。消息队列里的key如果被LRU淘汰，消息就莫名其妙没了。如果内存不够，优先设置volatile-lru，只淘汰带过期时间的key。
• 开启了AOF之后，rewrite机制会自动运行，不用担心日志无限膨胀，但要注意磁盘IO压力。
• 如果List里积压了几十万条消息，内存占用会直线上升，该报警报警，该扩容扩容。

3. Stream实现：从基础队列迈向完整消息队列

3.1 为什么最终要引入Stream

List方案需要自己维护processing队列和巡检逻辑，本质原因是缺少原生的确认机制。这些代码我写起来总觉得是打补丁。Redis官方也意识到了这个需求，于是在5.0版本推出了Stream。

Stream的思想和Kafka那一套很像：消息按追加顺序写入，每条消息有全局唯一的ID，支持消费者组。消费者读取消息后需要显式确认，未确认的消息会一直留在Pending列表里，可以随时查询、转移。

你可以这样理解：List是个朴素链表，Stream则是一个带日志、带游标、带提交记录的追加型结构。它不是为某个特定业务设计的，而是把消息队列的通用能力都做进去了。

3.2 生产消费与消费者组的基本流程

先看生产端，向Stream追加一条消息：

XADD queue:stream * orderId 1001 userId 888

第一个参数是Stream的key，第二个参数是消息ID。这里的星号表示让Redis自动生成全局递增ID，实际生成出来的ID长这样：1700000000000-0，前半部分是毫秒时间戳，后半部分是同毫秒内的序号。这个ID非常重要，消息顺序、消费者游标都靠它。

消费端要配合消费者组使用。先创建消费者组：

XGROUP CREATE queue:stream group1 0

第三参数0表示从Stream头开始消费。如果填$表示从最新消息开始消费，相当于只消费创建组之后的新消息。

消费者组内开始消费：

XREADGROUP GROUP group1 consumer1 COUNT 10 BLOCK 5000 STREAMS queue:stream >

这里最关键的>符号表示只读取本消费者组从未投递给任何消费者的消息。如果填一个具体的消息ID，则是从指定位置读历史消息，一般用于回溯。

这个机制带来的效果：group1组内的多个消费者不会读到同一条消息，实现了组内负载均衡；而另外创建组group2去读同一个Stream，又能拿到全量消息。这就是组间广播、组内竞争的语义，和消息队列要求的模型完全一致。

3.3 ACK与Pending机制：消息不被白丢

Stream的ACK机制是它和List最大区别。消费者读取消息后，必须显式确认：

XACK queue:stream group1 1700000000000-0

不确认会怎样？这条消息会一直停留在Pending列表里，也就是PEL。用XPENDING可以查看：

XPENDING queue:stream group1

输出会显示pending消息总数、最早最晚时间、以及各消费者的pending分布。当消费者A读取了一条消息但没确认就崩溃，这条消息不会丢，它会一直挂在A消费者名下。

怎么把这个悬挂消息捞回来处理？Redis 6.2提供了一个好用的命令XAUTOCLAIM：

XAUTOCLAIM queue:stream group1 consumer2 60000 0-0

第三个参数60000表示空闲超时60秒，最后一个参数0-0表示从最早的pending消息开始扫描。这个命令会把A名下超时未确认的消息转移给consumer2，由consumer2继续处理。处理完了再XACK，消息才真正从Pending里消失。

我实际做故障恢复时就是三步走：先XPENDING看悬挂情况，再XAUTOCLAIM转移消息，最后XACK清空pending。这套流程在消息不可靠的场景下非常好用。

要理解Stream的可靠性语义，得从数据链路看：消息从生产到消费成功，要经历三个环节。第一环是写入Redis，只要XADD成功返回了ID，数据就在内存里了。第二环是Redis持久化，需要AOF配置兜底。第三环是消费者处理完成并XACK。只有第三环完成，消息才算真正处理成功。所有环节都确认过，才能叫“不丢”。

3.4 消费组扩容、负载均衡与故障恢复

Stream的消费者组在扩容上需要说点实在的。很多人以为组内消费者越多，消费速度越快。在List方案下，多消费者并发BRPOP确实能线性提升吞吐。但在Stream里，虽然XREADGROUP也能让多个消费者分组读取不同消息，但底层还是同一个委托追加结构的共享读，消费者之间没有物理分区。这意味着，如果单实例Redis已经到了CPU或网络瓶颈，单纯加消费者并不能解决性能问题。

真正要扩展Stream的处理能力，通常是对业务维度做拆分，比如按订单号尾号拆成多个Stream，每个Stream由一个独立的Redis实例承接。Redis Cluster模式下，不同key会分布到不同节点，也能起到拆分效果。

故障恢复是Stream的主场。消费者崩溃后，执行XAUTOCLAIM把超时消息转移给其他消费者，相当于自动触发retry。恢复完成后监控XPENDING归零，整个过程不需要重启服务，也不需要人工介入。

3.5 关键参数设置与常见配置

Stream的日志型结构有个问题：消息不断增加，内存迟早撑爆。所以生产环境必须设置最大长度，比如：

XADD queue:stream MAXLEN ~ 10000 * orderId 1001

这里的波浪号表示近似裁剪，Redis允许它稍微超过10000条再裁剪。之所以不精确裁剪，是因为精确裁剪需要清理更多内存块，性能花费高，而近似裁剪在大多数业务场景下足够可靠。

消费端BLOCK时间建议设置成3000到5000毫秒，不要设成0。无限阻塞的连接和List方案里BRPOP的问题一样，连接断了不好感知。超时之后循环重试，既能保持实时性，又能让客户端连接定期重建。

消息体格式我建议直接用JSON字符串，订单消息就把整个业务对象序列化成JSON存进去。Stream支持一个字段值对，但不能嵌套对象，保存二进制也不方便，所以统一用JSON字符串最方便排查，也方便跨语言消费。

4. 实战排坑：消息丢失、积压、重复消费这些坑我都踩过

4.1 消息丢失排查实录

有一次线上反馈订单处理超时，怀疑是消息丢了。我的排查顺序是这样的：先看Redis的持久化配置，再查队列里还有没有消息，最后看消费者日志。

第一步，确认AOF是否开着：

CONFIG GET appendonly CONFIG GET appendfsync

如果结果是no或everysec，就要意识到存在宕机丢数据的窗口。那次排查发现AOF配置倒是开着，问题不在持久化。

第二步，查询消息是否还存在：

XLEN queue:stream XRANGE queue:stream - +

如果消息还在，说明没有被消费，问题在消费者侧。如果消息不在了，要看是否已经被XACK。

第三步，看消费者日志，那次的问题就出在这里：消费者代码先打印了一行日志，然后执行业务，业务抛了异常，异常被某个高层框架兜住了，既没有记录错误日志也没有重新投递，消息就这么“被成功消费”了。后来我加了一条规定：凡是读取到消息但处理失败，至少要抛出去或重新入队，绝不能静默吞掉。

还有一种容易忽视的丢失：Redis设置了maxmemory-policy为allkeys-lru。内存紧张时，消息key被LRU淘汰掉，连日志都没有，最开始排查根本不知道发生了什么。后来我在生产环境统一改成volatile-lru，并给队列相关的key设置了不淘汰策略。

4.2 消息积压的快速排查与处理

消息积压是最容易发现也最容易误判的问题。订单量没涨，队列先涨了，看起来是消费者不够，其实往往是消费者进程死了。

我的排查套路：

XLEN queue:stream XPENDING queue:stream group1

XLEN看Stream总长度，XPENDING看未确认消息数。如果XLEN很高但XPENDING很低，说明消息在被正常消费，只是堆积在了业务处理线程池里。如果两个数字都在涨，说明消费者进程可能挂了或者被卡死了。

一种经常被忽略的情况是消费者消费速度慢不是因为CPU不够，而是下游依赖变慢。比如消费者在调用第三方短信接口，接口超时从500ms涨到5秒，消费自然就堵住了。这时候盲目加消费者用处不大，先把下游超时时间调短，或者加熔断。

如果确认是消费者不够，可以临时加消费者实例。但要注意前面说的Stream单实例瓶颈。真的需要大规模扩容，切分多个Stream才是正路。

4.3 重复消费与幂等设计

消息队列几乎不可能做到“精确一次”，Redis做的消息队列也不例外。重复消费的根源是“处理完业务但没确认”和“确认了但网络返回失败”，这两种情况都会触发重投。

我处理重复的第一原则：业务侧自己做幂等。

最简单的方式是给每条消息定义一个业务ID，消费前先查这个ID是否处理过。用Redis自身就能做：

SET biz:order:1001 done NX EX 86400

如果返回OK，说明这条消息第一次处理，可以继续执行后续逻辑。如果返回空，说明消息重复，直接XACK跳过。这里利用了SETNX的原子性，不会出现两个消费者同时处理同一条消息的问题。

更保险的做法是在数据库层面加唯一约束。比如订单消息处理时会写一条处理流水表，给业务ID建唯一索引，重复写入会直接报异常，不用自己做判断。

幂等这事别靠自觉，一定要在设计阶段就定好约定。以后接入的同事哪怕不知道消息队列的语义，只要遵守这个约定，也不会出大问题。

4.4 消费者异常退出的抢救流程

一次真实的事故：消费者实例因为发布变更被重启，重启后发现pending列表里有一批消息，但消费者已经换了名字，消费者A那个名字还在pending里挂着，不会自动被新实例接手。

处理步骤：

1. 先看pending情况：

XPENDING queue:stream group1

2. 找到悬挂时间已经超过阈值的那批消息，用XAUTOCLAIM派发给当前活跃消费者：

XAUTOCLAIM queue:stream group1 consumer_new 60000 0-0 COUNT 100

这条命令会转移最多100条消息给consumer_new。

3. 被转移的消息处理完之后，执行XACK确认。

事故之后我加了一个定时监控脚本：每5分钟查一次XPENDING，如果pending数量超过1000或最早消息时间超过5分钟，就触发告警。这套规则实测下来很可靠，能在消费者异常后退货前及时发现。

4.5 死信与失败重试的兜底策略

消息处理失败是常态，网络抖动、依赖服务超时、数据异常都可能让消息没法成功处理。如果失败消息直接XACK丢弃，业务就实质丢失了。如果不XACK，又会一直卡在pending里。

我常用的一种兜底方案是“错误计数 + 死信队列”：

• 消息消费失败时，不要把业务异常吞掉，记一条错误计数。
• 用一个独立的哈希记录失败次数：

HINCRBY queue:error:order:1001 count 1

• 当失败次数超过3次，把这消息转投到一个专门的死信Stream：

XADD queue:deadletter * reason "timeout" payload "{...}"

• 死信Stream里的消息，人工或者定时任务定期检查，分析失败原因后再决定是否重新推送。

在失败重试时，一定要加退避。假设消息处理异常是因为下游服务正在重启，短时间内的重试大概率还会失败。退避间隔可以先短后长：第一次重试等1秒，第二次等10秒，第三次等1分钟。没有退避的话，失败消息会像子弹一样打向下游，把它打死。

4.6 内存容量计算

消息队列占不占内存？当然占。Stream和List都基于内存，设计容量时必须算清楚。

我一般这样估算：假设一个Stream设置MAXLEN为10万条，单条消息平均1KB，那这一个Stream最多100MB。算上消息字段的额外开销、消费者组状态、pending列表等，再乘以1.5的系数，就是150MB。

如果你的系统有10个这样的队列，大约1.5GB内存。这是保守估计。如果消息体积大，比如一个消息就10KB，那同样的10万条就是1GB，要按实际场景重新算。

评估容量时还要看峰值积压量。平时一天消费10万条，占不了多少内存。但活动期间如果消费者停半小时，积压的消息可能瞬间多几十万条。内存必须按峰值积压量来预留，而不是平均量。

我自己实践下来，Redis消息队列最好部署在独立的Redis实例上，不要跟业务会话数据、缓存数据混用。这样内存抢占问题、缓存命中率问题、maxmemory淘汰互相干扰的问题，全都不会发生。

接手过几次Redis队列事故之后，我的体会是：用Redis做消息队列最怕的不是Redis本身不稳定，而是设计方案时没有把可靠性边界想清楚。List适合快速实现，Stream适合作为持久方案。消息会不会丢，从你选择方案那一刻就已经决定了，后面只是靠配置和代码去补。如果你在做技术选型，我建议直接上Stream，然后把AOF打开、消费失败不静默、Pending列表纳入监控。这套组合做下来，中小型项目的消息中间件需求完全够用，而且省掉了一个专业MQ集群的运维负担。