要做一个“不考虑分区、Group 等”的 MQ 核心接收转发，本质是实现一条最小可用的消息通道：生产者把消息写进 Broker，Broker 可靠落盘（可选），消费者从 Broker 拉取或被推送，Broker 负责**路由、投递、重试、确认**。设计上把复杂度压在 Broker 端，协议与数据结构保持最小闭环。 ## 核心边界与最小闭环最小闭环包含四个动作： - **Produce**：Producer 发送消息到指定 Topic。 - **Fetch**：Consumer 拉取消息（pull 更易控，先做 pull）。 - **Ack**：Consumer 确认已处理，Broker 将其从 inflight/ready 中移除并推进位点。 - **Retry/Redeliver**（可选但强烈建议）：超时未 Ack 重新投递，或进入死信（DLQ）队列。不做的范围（显式砍掉复杂度）： - 无分区（Partition），每个 Topic 单一队列。 - 无消费者组（Group），一个订阅者即一个消费位点；或更简单：一个 Topic 仅允许一个 Consumer（极简模型）。 - 不做跨 Topic 事务、不做幂等生产、不做多副本复制。 ## 关键数据模型 ### 消息与元数据 ```java class Message { String topic; long id; // 单调递增或雪花 byte[] payload; long bornTimeMs; int deliveryCount; // 投递次数 } ``` 需要两个关键标识： - **id**：消息唯一标识，用于 Ack 与去重（消费端可选）。 - **deliveryCount**：超限进入 DLQ。 ### Broker 内部队列：ready + inflight Broker 端要区分“可投递”与“已投递未确认”： - **readyQueue**：待投递消息队列（FIFO）。 - **inflightMap**：正在飞行的消息（已下发，等待 Ack），记录 deadline（超时重投）。 ```java class Inflight { Message msg; long deadlineMs; // now + ackTimeout } ``` 这套结构对应最小语义：**At-least-once（至少一次）**。Exactly-once 不做。 ## 模块拆分：Broker 的三层结构 ### 1）网络层：协议与连接为了实现快，建议用 Netty（或 JDK NIO）做长连接，协议保持简单，可用 JSON 或自定义二进制帧。最小命令集： - `PRODUCE {topic, payload}` - `FETCH {topic, maxBytes|maxMessages}` - `ACK {topic, messageId}` - `PING`（可选心跳）响应： - `OK {messageId}` / `ERROR {code, msg}` - `MESSAGES [{id, payload, bornTimeMs, deliveryCount}]` 二进制协议更像 MQ，但 JSON 易调试；做核心功能时 JSON 足够。 ### 2）路由层：Topic -> Queue 无分区场景下，路由就是一个 Map： ```java ConcurrentHashMap topics; ``` 每个 `TopicQueue` 管 ready/inflight/持久化句柄。 ### 3）存储层：内存队列 + 可选 WAL 若不落盘，Broker 宕机消息丢失；最小可用仍建议加一个顺序追加日志： - **WAL（Write-Ahead Log）**：先追加到文件，再入 readyQueue。 - Ack 后可异步做清理（简化：不做物理删除，仅维护消费位点/已确认表）。极简 WAL 结构（每条记录）： - `len|topicLen|topic|msgId|bornTime|payload|crc` 只要保证“追加写 + fsync 策略”即可。为了压复杂度，可用： - 每 N 条 fsync 一次，或每 X ms 批量 fsync。 ## TopicQueue 的并发与调度 TopicQueue 需要三个循环： 1. 生产入队（append + enqueue） 2. 消费出队（ready -> inflight） 3. 超时扫描（inflight -> ready）设计要点： - readyQueue 可用 `LinkedBlockingDeque`。 - inflightMap 用 `ConcurrentHashMap`。 - 超时扫描用 `DelayQueue` 更省 CPU：把每个 inflight 用 deadline 包装成 Delayed 元素，到期自动出队触发重投。示意： ```java class TopicQueue { final BlockingDeque ready = new LinkedBlockingDeque<>(); final ConcurrentHashMap inflight = new ConcurrentHashMap<>(); final DelayQueue redeliverQ = new DelayQueue<>(); final long ackTimeoutMs = 10_000; final int maxDeliveries = 16; } ``` 投递逻辑（pull）： - Consumer 发 FETCH - Broker 从 ready poll N 条 - 放入 inflight，并投递 deadline 任务 - 返回消息列表 Ack 逻辑： - 收到 ACK(id) - 从 inflight 移除 - 记录已确认（内存 set 或持久化位点）超时重投： - redeliverQ 到期任务触发 - 若 inflight 仍存在且未 Ack： - deliveryCount++ - 超限 -> DLQ（另一个 TopicQueue 或文件） - 未超限 -> 移回 ready（通常放队首以加快重试，或队尾以避免饿死） ## Broker 端核心流程（伪代码） ### Produce ```java long onProduce(String topic, byte[] payload) { TopicQueue q = topics.computeIfAbsent(topic, TopicQueue::new); Message m = new Message(topic, nextId(), payload, now(), 0); wal.append(m); // 可选 q.ready.addLast(m); return m.id; } ``` ### Fetch（pull） ```java List onFetch(String topic, int maxMessages) { TopicQueue q = topics.get(topic); if (q == null) return List.of(); List out = new ArrayList<>(); for (int i = 0; i < maxMessages; i++) { Message m = q.ready.pollFirst(); if (m == null) break; long deadline = now() + q.ackTimeoutMs; q.inflight.put(m.id, new Inflight(m, deadline)); q.redeliverQ.put(new RedeliverTask(topic, m.id, deadline)); out.add(m); } return out; } ``` ### Ack ```java void onAck(String topic, long messageId) { TopicQueue q = topics.get(topic); if (q == null) return; Inflight inf = q.inflight.remove(messageId); if (inf != null) { wal.markAcked(topic, messageId); // 可选：追加 ACK 记录或更新位点 } } ``` ### Redeliver（延迟队列消费者线程） ```java void redeliverLoop(TopicQueue q) { while (running) { RedeliverTask task = q.redeliverQ.take(); // 到期阻塞返回 Inflight inf = q.inflight.get(task.messageId()); if (inf == null) continue; // 已 ACK if (now() < inf.deadlineMs) continue; // 处理抖动 // 仍未 ACK：重投 Message m = inf.msg; if (q.inflight.remove(m.id, inf)) { m.deliveryCount++; if (m.deliveryCount > q.maxDeliveries) { dlq(topic(m), m); } else { q.ready.addFirst(m); long deadline = now() + q.ackTimeoutMs; q.inflight.put(m.id, new Inflight(m, deadline)); q.redeliverQ.put(new RedeliverTask(m.topic, m.id, deadline)); } } } } ``` 注：上面为了保持“重投后仍进入 inflight 并继续超时监控”，把 ready->inflight 合并在重投中；也可以先放 ready，等下次 FETCH 再进 inflight。后一种更干净，但重投的时效取决于消费者拉取频率。 ## 消费语义与幂等位置在此模型下，Broker 只能保证至少一次；重复投递不可避免。幂等点通常放在 Consumer 侧： - 以 `messageId` 做去重（本地 LRU/持久化 KV）。 - 或业务侧天然幂等（订单状态机、唯一约束等）。 Broker 侧的 Ack 必须是显式的；自动 Ack 会把失败处理压扁成消息丢失。 ## 持久化的最小落地方式（WAL + 恢复）如果加 WAL，恢复流程要能把 ready/inflight 重建出来。因为不做 Group，最简单是： - WAL 只存 PRODUCE（消息体）与 ACK（messageId） - 启动时顺序扫 WAL： - 读到 PRODUCE：先放入临时 Map（id -> message） - 读到 ACK：从 Map 移除 - 扫完后 Map 剩余消息按 id 顺序放入 readyQueue 这样不需要随机写、不需要索引；缺点是 WAL 变大，需要滚动文件与截断（可以做 segment + checkpoint，留到后续）。 ## 工程化检查点（避免写成“能跑但不稳”的玩具） - **背压**：readyQueue 上限，满了 produce 返回错误或阻塞；否则内存打爆。 - **最大单条消息大小**：协议层先挡。 - **连接状态**：消费者断线不影响 ready；inflight 的消息靠超时回收。 - **线程模型**：每个 TopicQueue 一个重投线程可能太多，可用共享线程池 + DelayQueue（按 Topic 维度拆分或全局一套）。 - **指标**：ready 深度、inflight 数、投递/ACK 延迟、重试次数、DLQ 数。 - **顺序性**：ready FIFO；但重试插队会破坏严格顺序，若要严格顺序，重试只能回队首并阻塞后续消息，吞吐会下降。 ## 一个可实现的最小类图（落到 Java） - `BrokerServer`：Netty/NIO 入口、命令分发 - `TopicManager`：topic 映射与创建 - `TopicQueue`：ready/inflight/redeliver - `Wal`：appendProduce / appendAck / recover - `IdGenerator`：AtomicLong 或雪花 - `RedeliveryService`：消费 DelayQueue 在这个骨架上，MQ 的“核心接收转发”已经闭环：生产进来、消费拿走、显式确认、超时重投、可选落盘恢复。复杂特性（分区、Group、复制、一致性、事务）都可以后置叠加，不会推翻主干结构。

1 个回答