消息顺序性保证与乱序治理

保证：同一分区（Partition）内，消息按 offset 递增顺序写入与读取
不保证：不同分区之间的任何顺序关系
不保证：生产端「发送顺序」== 分区内「写入顺序」（取决于重试与并发）

需要顺序：状态机流转、事件溯源、账户记账、CDC 变更回放
不需要顺序：日志采集、指标上报、独立事件通知、可交换的更新

// 有 key：按 key 的 hash 取模分区 → 同 key 必同分区
producer.send(new ProducerRecord<>("orders", order.getUserId(), payload));

// 无 key：粘性分区（Sticky Partitioner）→ 同批粘在一个分区，批间可能换
producer.send(new ProducerRecord<>("orders", payload));

原则：选「业务上必须有序」的那个维度作为 key
订单状态流   → key = orderId（同一订单有序）
用户行为流   → key = userId（同一用户有序）
设备上报流   → key = deviceId（同一设备有序）
账户流水     → key = accountId（同一账户有序）

分区 0: 省A 60% 流量  ← 热点，Lag 飙升
分区 1~11: 其余 40% 分摊

// 同 key 打散为 N 个子流：orderId + ":" + (seq % 4)
// 代价：同 orderId 的顺序需在下游按 orderId 重排
String key = orderId + ":" + (hash(orderId) % 4);

分区数增加 → key 的 hash % 分区数 结果改变 → 同 key 可能迁到新分区
后果：变更前后同 key 的消息可能落在不同分区 → 顺序断裂

发送顺序：m1, m2（同一分区）
m1 失败重试，m2 成功 → 分区内实际顺序：m2, m1  ← 乱序！

props.put("enable.idempotence", "true");              // 开启幂等
props.put("max.in.flight.requests.per.connection", "5"); // 幂等下可安全开到 5
props.put("acks", "all");                              // 幂等要求 acks=all
props.put("retries", Integer.MAX_VALUE);               // 幂等下重试不再乱序

保证：单生产者会话（Producer Session）内，单分区的有序 + 去重
不保证：生产者重启后（新 PID）的跨会话顺序
不保证：跨分区顺序

// 危险：线程 A/B 竞争同一 key
executor.submit(() -> producer.send(new ProducerRecord<>("t", "k1", "create")));
executor.submit(() -> producer.send(new ProducerRecord<>("t", "k1", "pay")));
// 谁先 send 谁先进分区，顺序不确定

一个分区在同一时刻只被同组内的一个消费者消费
→ 单线程顺序 poll、顺序 process = 顺序处理

// 危险：多线程并发处理同一分区的消息
records.parallelStream().forEach(this::process);   // 顺序全乱

// 按 key 哈希到 N 个工作线程，保证同 key 串行
int worker = Math.floorMod(record.key().hashCode(), N);
workers[worker].submit(() -> process(record));
// 注意：提交 offset 需等所有 worker 完成「按分区最小未完成 offset」

消费者 A 处理到 offset=100，尚未提交 → 分区被收走
消费者 B 接管，从 offset=90 开始 → 重读 90~100 的消息

分区数 = 1 → 全 topic 严格有序
代价：无并行消费，吞吐被单分区上限锁死（通常几 MB/s）

需求：同一订单的所有事件有序
方案：key = orderId → 该订单的事件落同一分区 → 天然有序
不同订单之间无序 → 业务上通常可接受

orders:   创建订单 → 支付中
payments: 支付成功
// 消费者可能先看到 payments 的支付成功，再看到 orders 的支付中

生产者：每条消息带全局单调序号 seq（如 Snowflake ID）
消费者：维护滑动窗口，按 seq 排序后再处理
        窗口内缺失的 seq 等待 N 毫秒超时后跳过

① 业务序号断层：消息带 seq，消费者检测 seq 是否连续
② 事件时间倒挂：event_time 小于已处理的最大 event_time
③ 状态机非法流转：收到「已支付」却未收到「已创建」
④ 端到端断言：对账时比对源库与目标库的最终状态

// 简化版重排缓冲区
class ReorderBuffer {
    long nextExpected = 0;
    TreeMap<Long, Msg> pending = new TreeMap<>();

    void onMessage(Msg m) {
        if (m.seq == nextExpected) {
            deliver(m);
            nextExpected++;
            // 连续投递缓冲区中已就绪的后续消息
            while (pending.containsKey(nextExpected)) {
                deliver(pending.remove(nextExpected));
                nextExpected++;
            }
        } else if (m.seq > nextExpected) {
            pending.put(m.seq, m);   // 未来消息，缓存
        } // m.seq < nextExpected：重复，丢弃
    }
}

max.in.flight=1：吞吐 ≈ 基线 40%，顺序严格
幂等生产者：吞吐 ≈ 基线 95%，单分区严格有序
多线程消费：吞吐 ≈ 基线 × 线程数，需按 key 路由保序
单分区全局有序：吞吐 ≈ 单分区上限，通常不可接受