你项目中用消息队列，消费者和消费者之间是怎么联系的？

**消费者之间基本不直接联系**。它们通过“消息模型 + 外部存储/协调组件”间接协作：**同一 Topic、分区与位点管理、消费组语义、重平衡、共享状态/锁、以及必要时的事件-回执 Topic**。下面把可落地的做法按场景列清： # 1）基础：靠“消费组”而不是点对点通讯 * **同组内水平扩展**：同一 `groupId` 下多实例**共享分区**，一条消息只会被**组内一个实例**处理。联系=**分区所有权**与\*\*位点(offset)\*\*协调（由 Kafka/RocketMQ 做）。 * **不同组彼此独立**：A 组失败不影响 B 组（各自重试/位点独立）。 > 结论：消费者之间“不聊微信”，**通过队列本身的分区/位点机制**完成“谁来处理”的协作。 # 2）顺序与幂等：用“键控路由+幂等表” * **顺序**：对同一业务键（如 `orderId`）设为**分区键**，确保同键消息进同一分区→同一实例顺序处理。 * Kafka 例：`producer.send(new ProducerRecord<>(topic, orderId, payload))` * **幂等**：每实例都按相同规则做**去重**（DB 唯一键/Redis setnx/幂等表），无需相互通知。 * **长事务**：复杂逻辑先本地算好，再一次 CAS/事务写入，避免跨实例沟通。 # 3）“需要配合时”怎么办？ 用**外部可见状态**，而不是直连： * **共享状态/进度**：落库或 Redis（如 `task_status:{id}`），任何实例都能读到，避免点对点问询。 * **分布式锁/选主**：少数“全局任务”（补偿、对账）用 Redis `SET NX EX` 或 ZK/etcd 临时节点**选主**；其余实例被动等待。 * **反压与限速**：共享桶（Redis 令牌桶）或各自本地限流；真实背压由队列“堆积”天然体现，无需互相打招呼。 * **缓存一致**：用**广播 Topic**做“缓存失效通知”，所有实例订阅并清本地缓存。 # 4）失败转交与重试 * **本地重试 → 延迟/死信队列**：失败 N 次投到 `retry`/`DLQ`，由“专用消费者”捞回再处理；不找其他实例“求助”。 * **可见性超时（SQS/RMQ 类）**：超时未 ack 自动回到队列，被**别的实例**捞起，天然交接。 # 5）请求-响应（需要“回话”时） 仍不直连，用**回执 Topic**： * 生产者在消息头写 `correlationId`、`replyTo`。 * 消费者处理后发到 `replyTo`，带回同一个 `correlationId`。 * 这样形成**异步 RPC**，但依旧解耦、可扩展。 # 6）重平衡与分区迁移 * **协调器负责**把分区在实例间**转移**（实例上下线、扩缩容）。 * 实例需实现：`onPartitionsRevoked/Assigned`（Kafka）或对应回调，**提交位点、暂停处理、幂等再入场**。 * 这就是“消费者之间换班”的正规途径。 # 7）监控与可观测 * **心跳、Lag、处理时延**全部打到**统一监控**（Prometheus/ELK/控制台），而不是互相通知。 * 告警触发扩容/限流策略。 # 8）最常用的三种“联系模式”概括 1. **共享消费组 + 键控分区**：谁处理由协调器决定（默认）。 2. **共享外部状态/锁**：当有“唯一执行者/全局任务”要求时。 3. **事件-回执 Topic**：确需点对点答复时的异步 RPC。 ## 代码/配置片段（Kafka 例） **生产路由保证同键顺序** ```java producer.send(new ProducerRecord<>(topic, orderId, payload)); // orderId 作为 key ``` **消费者回调处理重平衡** ```java consumer.subscribe(List.of(topic), new ConsumerRebalanceListener() { public void onPartitionsRevoked(Collection<TopicPartition> ps) { commitOffsets(); // 提交位点，安全交接 } public void onPartitionsAssigned(Collection<TopicPartition> ps) { seekFromStoredOffsets(ps); // 从存储位点继续 } }); ``` **幂等去重（示意）** ```java if (dedupRepo.tryMarkProcessed(correlationId)) { handle(msg); } ``` ### 一句话总结 在分布式系统里，**消费者与消费者不直接联系**；**联系=通过消息系统的“分区、位点、重平衡”+ 外部存储/锁/回执 Topic 间接完成协作**。这样可扩、可观测、可恢复，也不会把拓扑搞成“网状地狱”。