【混合缓存】五种同步通道与它们的失效模式：为什么 Pub/Sub 不可靠反而是对的（五）- 惊觉

混合缓存的同步层有五个实现：Redis Pub/Sub、RocketMQ、Kafka、RabbitMQ、NATS。选型时大家比较的是「延迟、吞吐、运维成本」，但真正决定线上表现的是三种失效形态——丢、重、乱。这篇文章把五种通道拆到这三种形态上，并论证一件事：因为同步消息传的是「失效」而不是「数据」，重和乱几乎无害，只有丢要兜底。

一、痛点：三种「以为可靠」的错觉

1.1 错觉一：以为 MQ 的持久化等于「同步不会丢」

早期我把同步通道从 Redis 换成了 RocketMQ，理由是「Redis Pub/Sub 不持久化，会丢消息」。

换完之后确实没再出现「订阅者重启期间漏消息」的问题，但线上仍然有新节点读到旧值的情况。查下来是这样一条链：

节点 C 启动时，消费者还没连上 Broker；
这期间节点 A 写了一条缓存，消息被 Broker 收下；
节点 C 的消费者连上之后从最新的偏移开始消费（auto.offset.reset=latest），跳过了那条消息；
节点 C 的本地缓存里是更早的旧值，一直用到 TTL 到期。

「消息持久化了」和「这个节点会收到」是两件事。 Broker 里有消息，不代表你的消费者会读到它——取决于消费位点从哪里开始。

1.2 错觉二：以为「至少投递一次」等于「处理一次」

另一个教训来自重复投递。

某个节点处理同步消息时抛了异常，框架把它退回队列重投。重投之后又抛——因为失败的原因（某个 key 的对象反序列化失败）是稳定复现的。于是这条消息在队列里来回滚了几十次，消费线程被占满，后续所有同步消息都堵在它后面。

「至少一次」的另一面是「可能无限次」。 没有死信队列、没有重试上限的通道，一条毒消息就能把整个同步层按住。

1.3 错觉三：以为五个通道只是「换一个实现类」

我最初真的以为只要改一行 sync-type 就能换通道。实际上五个实现的差异远大于「API 不同」：

差异点	后果
广播靠什么实现	独立 group / 独立队列 / 独立 durable，每种通道做法不同
有没有 ACK	无 ACK 的通道「处理失败」和「处理成功」无法区分
消费位点从哪开始	新节点补不补历史消息，直接决定「重启后的一致性」
失败后怎么办	重投、丢弃、还是根本不重试

这些差异不会体现在接口签名上，全部藏在实现里。切换通道前不看这三件事，等于在不知道兜底逻辑的情况下换掉安全网。

1.4 我最初的错误做法

错误一：把同步当成「数据复制」来设计兜底。

一开始我担心「消息丢了一条，数据就少了一份」。其实同步消息从不携带值，只携带 cacheName + key + 操作类型；接收方收到之后只做一件事——把本地那个 key 删掉。

错误二：为「消息乱序」设计序列号。

我认真考虑过在消息里加自增序号，接收端丢弃旧序号的消息。实现到一半发现没必要：因为同步动作是「删除本地副本」而不是「写入新值」，乱序执行的结果和顺序执行完全一样——删两次和删一次等价。

错误三：认为 Redis 默认选项是「够用就行」的选择。

默认 sync-type: REDIS 看起来像「没想清楚就用最简单的」。实际上它是对的：这套系统的同步层只需要低延迟的「失效通知」，不需要可靠投递。真正必须可靠的场合，兜底不是通道，而是时间上界。

二、问题拆解：丢、重、乱，以及为什么只关心丢

2.1 同步消息到底在传什么

先看消息模型。五个字段，没有一个携带业务值：

字段	作用
`cacheName`	路由到哪个缓存实例（全限定名，含前缀）
`key`	要失效的键；`CLEAR` 时为 `null`
`operationType`	`PUT` / `EVICT` / `CLEAR`
`timestamp`	生成时间，仅用于排查
`sourceNode`	发送方节点 ID，用于回环过滤

接收端的处理逻辑（MultiLevelCache#handle）很短：

switch (message.getOperationType()) {
    case PUT -> {
        if (key != null) {
            localCache.evict(key);
        }
    }
    case EVICT -> {
        if (key != null) {
            localCache.evict(key);
        }
    }
    case CLEAR -> localCache.clear();
}

注意 PUT 分支也是 evict。别人的写入在我这里变成「删掉我的本地副本」，下次读的时候我会回远程取新的。这是整个设计的核心：同步传播的是「你不知道了」，不是「你应该知道什么」。

2.2 三种失效形态的危害不对称

形态	具体表现	对本系统的影响	需要处理吗
丢	某条消息没有被任何节点收到	本地副本陈旧直到 TTL 到期	✅ 必须兜底
重	同一条消息被处理两次以上	第二次 `evict` 一个已经不存在的 key	❌ 幂等，无害
乱	B 的消息先到、A 的后到	两次 `evict` 顺序无关	❌ 幂等，无害

这张表是本文最重要的一张。它解释了两个此前看起来奇怪的设计选择：

为什么消息里没有版本号：不需要，因为动作是幂等的删除，不是有状态的覆盖；
为什么可以接受「至少一次」：重复无害，所以不必追求「恰好一次」——而「恰好一次」在分布式消息里本来就要靠额外成本换。

于是问题简化成：五种通道各自的「丢」发生在什么时候，兜底靠什么。

2.3 丢的四种来源

来源	触发条件	影响范围
发布时无人订阅	接收方没连上 / 监听器注册失败	那一段时间内所有节点
消费位点跳过	消费者从 `latest` 开始	新启动或长期离线的节点
处理失败后丢弃	反序列化失败、重试超限	单条消息
发布方自己失败	网络、序列化异常	单条消息

这四种在五种通道上的分布完全不同，下一节逐一对。

下面这张图把「丢、重、乱」三种形态与最终结论串起来：

三、实现：五种通道，五种丢法

3.1 整体结构

看图要点：广播语义由各自的「每节点一份订阅」实现；回环过滤在消费端统一做；五种通道的分叉点在传输层，汇聚点之后的逻辑完全相同。

3.2 Redis Pub/Sub：不持久化是特性，不是缺陷

long receivers = topic.publish(messageBytes);
if (receivers == 0) {
    log.warn("发布缓存同步消息但无订阅者接收: nodeId={}, channel={}, message={}",
            nodeId, channel, message);
}

RTopic.publish 返回收到消息的订阅者数量。这个返回值在这里被用来做一件事：当它是 0 时打一条 warning。

它能发现什么：本进程的监听器没注册上、Redis 连接异常、channel 配错。
它不能发现什么：其他节点的订阅者离线。因为 receivers 只统计当前连接着的订阅者——正好是「收不到的人不会出现在这个数字里」。

Redis Pub/Sub 的失效形态很纯粹：

场景	结果
订阅者在线	立即收到，延迟最低（同机房亚毫秒级）
订阅者离线	消息直接消失，不排队、不补投
订阅者处理失败	消息已经被 Redis 扔掉，无处重试
Broker（Redis）重启	所有订阅关系重建，期间的消息全丢

这看起来是最差的可靠性，但它的代价换来了两件事：延迟最低、不需要额外的中间件（Redis 本来就在）。而丢的后果被 forceExpiryTime 兜住了——最多陈旧 N 秒。

所以「Pub/Sub 不可靠」不是缺陷，是这套设计的匹配点：同步层负责「尽快通知」，时间上界负责「一定会通知」。两者分工，不需要同步层保证可靠。

3.3 Kafka：每节点一个 group，但位点会跳过

Kafka 实现的首要问题是广播：同一个 group 里多个消费者会分区分摊，不是广播。解法是每个节点用独立的 group：

this.groupId = groupIdPrefix + "-" + nodeId;

nodeId 是 NodeIdHolder 的 ULID，进程启动时生成一次。于是每个节点是一个独立消费者组，都能收到全量消息——这就是「广播」在 Kafka 上的实现方式：N 个 group，N 份全量消费。

代价也在这里：N 个节点 = N 份存储读取，Broker 的出口带宽随节点数线性增长。

真正危险的是消费端配置：

props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "latest");
props.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, "true");
props.put(ConsumerConfig.MAX_POLL_RECORDS_CONFIG, 100);

三行配置决定了三种丢法：

配置	丢法	说明
`auto.offset.reset=latest`	跳过历史	新 group 没有已提交位点时，从头开始 = 从最新开始，之前的所有消息都不消费
`enable.auto.commit=true`	先提交后处理	处理抛异常也会在下次 poll 时提交位点，那条消息永久丢失
`max.poll.records=100`	批量失败	一批里一条失败，其余照常处理，失败那条不重试

latest 与「每节点独立 group」叠加，产生了一个反直觉的后果：节点重启后 nodeId 变了（ULID 每次进程启动重新生成），group 名随之变化，Broker 把它当成一个全新的消费者组——而新组的位点策略是 latest。

所以 Kafka 通道在节点重启时的行为是：离线期间的所有同步消息都不会被补投。持久化没帮上忙，因为消费位点是从「现在」开始的。

这个组合不是 bug——它和 Redis 的定位其实一样（只保证在线期间的投递），但它披着「消息持久化」的外衣，容易让人误以为更可靠。

把「每节点独立 group」与「latest 位点」叠加后的时序画出来，就能看清为什么持久化没帮上忙：

3.4 RocketMQ：会重投，但重投不等于重新处理成功

RocketMQ 用同样的思路做广播：每节点独立 consumerGroup。消费端的语义比 Kafka 明确：

return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
// 或
return ConsumeConcurrentlyStatus.RECONSUME_LATER;

RECONSUME_LATER 让 Broker 稍后重投。这带来两个后果：

后果	说明
重投可能重复	处理成功但返回失败，或 ACK 丢失，都会导致重复消费
重投没有上限	稳定复现的毒消息会反复重投，直到达到 Broker 的最大重试次数

ConsumeConcurrently 这个模式名也说明了并发消费、不保证顺序。

但回到 2.2 那张表：重复和乱序对「删除本地副本」这个动作无害。真正需要担心的是重投期间的延迟——一条消息要等重试间隔过去才会再次尝试，这期间该节点的本地副本是陈旧的。

3.5 RabbitMQ：Fanout 广播，失败处理要自己定

RabbitMQ 是唯一一个「原生广播」的通道：Fanout Exchange 把消息复制到所有绑定队列，每个节点声明自己的队列：

channel.exchangeDeclare(exchange, BuiltinExchangeType.FANOUT, durable, false, null);
queueName = queuePrefix + "-" + sanitizeQueueName(nodeId);
channel.queueDeclare(queueName, durable, false, false, null);
channel.queueBind(queueName, exchange, "");
channel.basicQos(1);

basicQos(1) 表示一次只预取一条——对同步消息这种小体量、低延迟的场景合理（处理完再取下一条，避免一个节点堆积后批量处理）。

手动 ACK，失败的处理分两种：

if (msg == null) {
    log.warn("反序列化缓存同步消息为空");
    channel.basicNack(envelope.getDeliveryTag(), false, false);   // requeue=false → 丢弃
    return;
}
// ...
} catch (Exception e) {
    channel.basicNack(envelope.getDeliveryTag(), false, true);    // requeue=true → 重投
}

这两行是整套实现里最有判断力的地方：

失败类型	requeue	理由
反序列化失败	`false`（丢弃）	字节流已经坏了，重投一万次也解不出来
处理异常	`true`（重投）	可能是瞬时故障（如本地缓存被关闭），值得重试

「不可重试的错误不重投」这条判断，比任何 QoS 配置都重要。把毒消息留在队列里，代价是整个队列的吞吐。

不过 requeue=true 在 RabbitMQ 上有个已知行为：消息会被立即重新入队，如果没有延迟队列配套，稳定复现的异常会导致热循环重投——消息几乎零间隔地反复回到消费者手上，CPU 和日志都会被刷爆。

3.6 NATS JetStream：durable 名字里藏着一个坑

NATS 的实现看起来最「正确」：JetStream 持久化、每条消息 ACK/NAK、每节点独立 durable consumer。

String durableName = consumerPrefix + "-" + sanitizeConsumerName(nodeId);
PushSubscribeOptions pso = PushSubscribeOptions.builder()
        .stream(streamName)
        .durable(durableName)
        .configuration(ConsumerConfiguration.builder()
                .deliverPolicy(DeliverPolicy.New)
                .build())
        .build();

durable 的语义是「这个消费者的位点由服务端记住，断开重连后继续」。看起来能解决 Kafka 的「重启就跳过」问题。

但 durable 名字里带了 nodeId，而 nodeId 是进程启动时生成的 ULID。所以：

进程重启 → nodeId 变化 → durable 名字变化；
服务端把它当成一个全新的 durable，没有已提交位点；
DeliverPolicy.New 生效 → 从「现在」开始，之前的消息一笔勾销。

结果和 Kafka 完全一样：离线期间的消息不补投。 durable 的断点续传能力，被一个每次都会变的消费者名字抵消掉了。

这个坑特别值得记：「我用了 durable」和「我得到了续传」之间，隔着消费者标识的稳定性。任何把「实例 ID」放进消费者名的设计，都自动放弃了续传。

3.7 回环过滤：所有通道共用一段

五个实现里都有一段相同的判断：

if (NodeIdHolder.isCurrentNode(message.getSourceNode())) {
    return;   // 自节点回环，正常路径：直接吞掉不处理
}

sourceNode 在发布端写入，消费端与本节点 ID 比对。这段逻辑不依赖通道，所以它被放在消费回调的最前面。

Redis 实现里还有一段值得看的注释——它纠正了一个容易犯的错：

// Redisson 侧信息（注意：RTopic.MessageListener 收到的是 CharSequence 频道 + payload，
// Redisson 4.x 的 Message<byte[]> 没有 getRedissonClientId() 之类的方法——
// 这里 channelFromRedis 是发布端/订阅端配置的 channel 名，跟 sourceNode 不是一回事。
// 想区分"是不是本节点发出去的"仍要靠反序列化后的 message.sourceNode 与 NodeIdHolder 比对。）

为什么「频道名」不能用来判断回环：频道名是配置项，同一个应用的所有节点用的是同一个频道。它标识的是「哪个缓存主题」，不是「谁发的」。

另一个细节：addListener 返回的 listenerId 小于 0 时打 error 日志。这是一个「注册失败」的显式信号——如果监听器没注册上，本节点从此收不到任何同步消息，而发布端看不出任何异常（别人的 receivers 里根本没有你）。

3.8 热切换：为什么旧发布器要延迟销毁

同步配置本身支持热更新（reloadSyncPublisherIfChanged），流程是「先建新 → 校验 → 原子切换 → 延迟销毁旧」：

// Step 1：创建并初始化新发布器（失败则不切换）
newPublisher = syncPublisherFactory.create(newSyncConfig);
// Step 2：原子替换（旧发布器不再接收新的 publish 调用）
CacheSyncPublisher oldPublisher = this.syncPublisher;
this.syncPublisher = newPublisher;
// Step 5：旧发布器延迟销毁（消费端继续接收窗口期未完成的消息）
Duration delay = newSyncConfig.getOldPublisherDestroyDelay();

为什么旧的要留一段时间：旧发布器的消费端可能还有 in-flight 消息没处理完（RabbitMQ 的手动 ACK、NATS 的 ack 都可能在路上）。立即销毁等于主动丢掉这批消息。

这里有一处注释与实现不一致值得记下：KafkaCacheSync 的类注释写「热切换时旧实例保留 2 小时」，MultiLevelCacheManager 的方法注释也写「旧发布器 2 小时后完全销毁」，但配置项 oldPublisherDestroyDelay 的默认值是 60 秒。

以配置为准：默认窗口是 60 秒。注释里的「2 小时」大概是早期版本的设定，改配置时没同步改注释。这类偏差在排查「为什么切换后还有旧消息进来」时会造成误判。

热切换的完整流程如下，注意「延迟销毁」这一步是刻意保留的窗口：

四、权衡、边界与坑

4.1 通道对照表（本文核心产出）

通道	广播实现	持久化	消费位点	丢的时机	兜底手段
Redis Pub/Sub	同频道所有订阅者	❌ 无	无位点概念	订阅者离线 / 处理失败	`forceExpiryTime`
RocketMQ	每节点独立 group	✅	Broker 管理	重试超限 / 位点过期	重试 + `forceExpiryTime`
Kafka	每节点独立 group	✅	`latest` + 自动提交	新 group 跳过历史 / 自动提交先于处理	`forceExpiryTime`
RabbitMQ	Fanout + 每节点队列	✅（`durable` 可配）	手动 ACK + QoS 1	反序列化失败丢弃 / 队列未持久化	ACK + `forceExpiryTime`
NATS JetStream	每节点 durable	✅（File 存储）	durable + `DeliverPolicy.New`	节点重启后 durable 名变化 → 跳过历史	`forceExpiryTime`

最后一行（兜底手段）是关键：五种通道的兜底全部包含 forceExpiryTime。这不是巧合——它说明同步通道在这套设计里从来不是一致性保证的来源，只是一条「尽快」的路径。

4.2 选型判据

你的约束	选择	理由
已有 Redis，节点数 < 20，能接受 N 秒陈旧	`REDIS`	零额外组件，延迟最低
已有 Kafka，节点数多，不想再引入 MQ	`KAFKA`	高吞吐，N 份消费的成本可接受
需要可靠投递 + 原生广播	`RABBITMQ`	Fanout 语义最直接，手动 ACK 可控
已用 RocketMQ，需要事务消息	`ROCKETMQ`	生态一致
云原生、节点数 100+	`NATS`	轻量，JetStream 可靠广播

注意这张表里没有任何一行是因为「一致性」而选的。选通道看的是运维成本和延迟量级，一致性由时间上界统一兜底——这是本文想说清的分工。

4.3 踩过的四个坑

坑 1：以为换了 MQ 就不会丢消息。

现象：从 Redis 换到 Kafka 后，仍有节点读到旧值。
定位：节点重启后 nodeId 变化 → group 名变化 → 新 group 从 latest 开始，离线期间的消息全部跳过。
修法：要么接受（靠 TTL 兜底），要么把消费者标识做稳定（不能用每次启动都变的 ULID），并把位点策略改成 earliest。

坑 2：RabbitMQ 上一条毒消息把队列打满。

现象：某个节点日志刷屏，同步消息处理延迟飙升。
定位：basicNack(requeue=true) 是即时重投，稳定复现的异常形成热循环。
修法：区分「可重试」与「不可重试」，反序列化失败必须 requeue=false；再配合重试上限或死信队列。

坑 3：以为 durable consumer 就能续传。

现象：NATS 节点重启后，离线期间的变更没同步到。
定位：durableName = consumerPrefix + "-" + nodeId，nodeId 每次启动重新生成；加上 DeliverPolicy.New，等于每次重启都是新消费者。
修法：用稳定的消费者名（如 hostname + 固定序号，或由配置显式指定），并评估是否要改成 DeliverPolicy.All。

坑 4：以为热切换后旧通道立即失效。

现象：切换通道后，日志里还能看到旧通道的消息被处理。
定位：旧发布器延迟销毁（默认 60 秒），这期间消费端仍在工作。这是设计如此——为了不丢 in-flight 消息。
修法：不用改，但要理解这段窗口内两条通道并存。切换后的短暂重复消费是无害的（幂等删除）。

4.4 同步层防不住什么

风险	能否防住	说明
在线节点收不到失效通知	✅ 基本能	通道正常时延迟很低
离线节点错过变更	❌	五种通道都会跳过离线期间的消息
毒消息阻塞整个通道	⚠️ 部分	取决于通道的失败策略与重试上限
跨应用误消费	⚠️	靠 `cacheName`（含前缀）过滤，配错前缀会串
一致性窗口	❌	同步只让窗口更小，上界永远由 TTL 决定
Redis 整体不可用时的同步	❌	`REDIS` 通道跟随 Redis 一起失效

第一行和第五行要分开看：同步能把「通常情况下的不一致窗口」压到毫秒级，但它不改变「最坏情况」——最坏情况永远由 forceExpiryTime 决定。这就是为什么专题里 S-1 先讲时间上界、S-4 才讲通道：通道是优化项，上界是保底项。

五、可以直接抄走的清单

先明确同步消息传的是「失效」还是「数据」。传失效时重复与乱序无害，可以放弃「恰好一次」。
在消息里带 sourceNode，消费端第一件事就是过滤回环，不要用频道名判断。
RTopic.publish 的返回值是当前在线订阅者数，为 0 时打 warning——它能发现注册失败，但发现不了别人离线。
用 MQ 做广播时，每个节点必须有独立的 group / 队列 / durable，否则会变成集群内分摊消费。
消费者名字里不要放每次启动都变的 ID，否则 durable、位点续传全部失效。
Kafka 的 auto.offset.reset 默认 latest，意味着新节点不补历史消息；要补就改 earliest，但要评估重放量。
enable.auto.commit=true 是「先提交后处理」，处理失败也会前进——需要严格语义时改手动提交。
手动 ACK 的通道必须区分可重试与不可重试：反序列化失败直接丢弃，不要 requeue=true。
别用 requeue=true 做重试，除非有延迟队列或重试上限，否则会热循环。
热切换同步通道时，旧发布器要延迟销毁，给 in-flight 消息留处理窗口；默认 60 秒，别看注释里的「2 小时」。
所有通道的兜底都是同一个 forceExpiryTime。不要指望换通道能把「最坏陈旧时长」变小。
LOCAL 模式不发布同步消息；REMOTE 模式收到消息直接返回——这两条是省开销的，不要「统一处理」掉。

附：一条消息的一生

阶段	可能出的问题	后果	兜底
序列化	Kryo 未注册类	发布端抛异常	显式注册 `CacheSyncMessage`
发布	无订阅者 / 网络失败	这条变更没传出去	TTL
传输	通道不持久化	消息消失	TTL
消费入口	反序列化失败	丢弃或重投	丢弃（`requeue=false`）
回环过滤	`sourceNode` 缺失	自节点重复处理	无害（幂等）
执行	本地缓存已关闭	抛异常 → 重投	重投 + TTL

最后一行值得单独说：消费阶段抛异常时，最坏结果不是「数据不一致」，而是「消息堆积」。因为真正的一致性上界由 TTL 保证，堆积只影响延迟。这也是为什么可以接受「不来也行」的通道设计——系统的正确性不依赖消息一定到达。

把「一条消息的一生」按阶段串成流程图，可以一眼看出每个环节的兜底落在哪里：

转载自 CSDN-专业IT技术社区

原文链接：https://blog.csdn.net/jkhyf/article/details/170378726

【混合缓存】五种同步通道与它们的失效模式：为什么 Pub/Sub 不可靠反而是对的（五）