4. 基于消息队列的恢复方案：Kafka/RocketMQ 的消息回溯与重放

订单簿数据丢了，怎么办？

别慌。只要你的数据进了消息队列，我们就有办法把它捞回来。

我个人习惯把消息队列比作「黑匣子」。飞机上的黑匣子能记录飞行数据，消息队列能记录每一笔订单的变更。说白了，Kafka 和 RocketMQ 天生就支持消息回溯，这是它们作为分布式日志系统的基本能力。

4.1 为什么消息队列能扛住订单簿恢复？

订单簿的数据流，本质上是一个追加写的过程。每一笔订单的创建、成交、撤单，都是一个个不可变的事件。消息队列把这些事件按顺序存下来，你想回放多少条，就回放多少条。

我在项目中遇到过这样的情况：某次数据库主库挂了，从库同步延迟了十几秒。等我们切回从库时，发现订单簿的状态已经乱掉了。怎么办？

嗯，我们直接让订单簿服务从 Kafka 里重新消费最近 30 秒的消息。30 秒的数据量，对于高频交易系统来说可能也就几万条。重放一遍，订单簿状态就恢复如初了。

核心思想：消息队列保存的是事件流，不是状态快照。状态可以丢，事件不能丢。只要事件还在，状态就能重建。

4.2 Kafka 的消息回溯机制

Kafka 的消息回溯，说白了就是控制消费者从哪里开始读。

Kafka 的消费者通过 offset 来定位消息位置。默认情况下，消费者从最新的 offset 开始消费。但我们可以手动指定 offset，回到任意时间点。

4.2.1 基于时间戳的回溯

这是最常用的方式。你告诉 Kafka：「我要从 10 分钟前的消息开始消费。」

// 获取指定时间戳对应的 offset
Map<TopicPartition, Long> timestampsToSearch = new HashMap<>();
timestampsToSearch.put(partition, System.currentTimeMillis() - 10 * 60 * 1000);

Map<TopicPartition, OffsetAndTimestamp> offsets = consumer.offsetsForTimes(timestampsToSearch);

// 跳转到该 offset
for (Map.Entry<TopicPartition, OffsetAndTimestamp> entry : offsets.entrySet()) {
    consumer.seek(entry.getKey(), entry.getValue().offset());
}

// 开始消费
while (true) {
    ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(100));
    for (ConsumerRecord<String, String> record : records) {
        // 重建订单簿
        rebuildOrderBook(record.value());
    }
}

这段代码我用了很多次。注意一点：offsetsForTimes 返回的是大于等于指定时间戳的第一个 offset。所以如果你指定了 10 分钟前，实际拿到的可能是 9 分 59 秒的数据，这完全够用。

4.2.2 基于 offset 的精确回溯

如果你知道数据丢失的具体位置，可以直接跳到指定 offset。

consumer.seek(partition, 123456L);

这种方式更精确，但前提是你得知道 offset 值。我在做容灾演练时，会把关键 checkpoint 的 offset 记录下来，方便快速恢复。

我的习惯：每处理 1000 条消息，就把当前 offset 写入一个本地文件。这样即使服务挂了，重启后也能从最近的 checkpoint 开始恢复，不用从头重放。

4.3 RocketMQ 的消息回溯机制

RocketMQ 的回溯方式跟 Kafka 类似，但 API 更简洁一些。

4.3.1 按时间回溯

// 创建消费者
DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("orderbook_group");
consumer.setNamesrvAddr("localhost:9876");
consumer.subscribe("orderbook_topic", "*");

// 设置从 10 分钟前开始消费
consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_TIMESTAMP);
consumer.setConsumeTimestamp(UtilAll.timeMillisToHumanString3(
    System.currentTimeMillis() - 10 * 60 * 1000));

consumer.registerMessageListener((MessageListenerConcurrently) (msgs, context) -> {
    for (MessageExt msg : msgs) {
        rebuildOrderBook(new String(msg.getBody()));
    }
    return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
});

consumer.start();

RocketMQ 的 CONSUME_FROM_TIMESTAMP 模式，会自动找到该时间点之后的第一条消息。我个人觉得 RocketMQ 的 API 设计更人性化，少写了不少代码。

4.3.2 按 offset 回溯

RocketMQ 也支持按 offset 回溯，但需要手动操作。

// 获取指定队列的 min 和 max offset
long minOffset = consumer.minOffset(MessageQueue);
long maxOffset = consumer.maxOffset(MessageQueue);

// 从指定 offset 开始拉取
PullResult pullResult = consumer.pull(MessageQueue, "*", targetOffset, 100);

嗯，这里要注意：RocketMQ 的 offset 是队列级别的，不是 topic 级别的。如果你有多个队列，需要分别处理。

4.4 消息回溯的实战策略

光知道 API 怎么用还不够。在实际的订单簿恢复中，我们需要一套完整的策略。

场景 回溯方式 恢复时间 数据完整性
服务重启（秒级） 从上次 checkpoint 开始 < 1 秒 高
服务重启（分钟级） 按时间回溯 5 分钟 1-5 秒 高
数据库故障 按时间回溯 30 分钟 5-30 秒 中
全量重建 从最早 offset 开始 分钟级 完全

你想想看，不同的场景对恢复时间的要求不一样。如果是服务重启，我们当然希望越快越好，所以从 checkpoint 开始最合适。但如果是数据库故障，可能丢失了部分数据，那就需要回溯更长的时间来补全。

我曾经踩过的坑：有一次我设置了从 30 分钟前回溯，结果那 30 分钟里产生了 200 万条消息。订单簿服务直接 OOM 了。后来我加了一个限流机制：重放时每秒最多处理 5 万条消息，给 GC 留点喘息空间。

4.5 消息重放的幂等性设计

消息重放最怕什么？

怕重复处理。

比如一条「撤单」消息被重放了两次，订单簿里同一个订单被撤了两次，状态就乱了。

解决方案很简单：幂等性。

我常用的做法是给每条消息加一个全局唯一的 ID，然后在订单簿里记录已经处理过的 ID。

public class OrderBookService {
    private Set<String> processedIds = new ConcurrentHashSet<>();
    
    public void processMessage(OrderEvent event) {
        // 检查是否已经处理过
        if (processedIds.contains(event.getId())) {
            return; // 跳过重复消息
        }
        
        // 处理消息
        applyToOrderBook(event);
        
        // 记录已处理
        processedIds.add(event.getId());
    }
}

当然，processedIds 不能无限增长。我一般会定期清理，只保留最近 10 分钟的 ID。因为消息重放通常只发生在短时间内，超过 10 分钟的消息不太可能被重复处理。

4.6 消息队列的容灾架构

单机版的 Kafka 或 RocketMQ 肯定不行。订单簿这种核心数据，必须上集群。

我画了一张图，展示消息队列在订单簿容灾中的位置：

订单簿消息队列容灾架构 交易引擎 产生订单事件 消息队列集群 Kafka / RocketMQ 3副本 + ISR 订单簿消费者组 多实例 + 负载均衡 订单簿内存状态 Redis / 本地内存 消息回溯路径 正常流程：交易引擎 → 消息队列 → 消费者 → 订单簿 恢复流程：订单簿 → 回溯消息队列 → 重新消费

这张图里，消息队列集群是核心。我建议至少部署 3 个副本，配合 ISR（In-Sync Replicas）机制，确保即使一台机器挂了，数据也不会丢。

消费者组里的多个实例，可以并行消费不同的分区。这样回溯时也能利用多核能力，加快恢复速度。

4.7 避坑指南

最后，分享几个我踩过的坑：

  • 消息积压问题：回溯时如果消息量太大，会导致消费延迟。我建议先暂停正常消费，专心做回溯。等回溯完了再恢复。
  • 顺序问题：同一个订单的事件必须发到同一个分区，否则回溯时顺序会乱。Kafka 里用订单 ID 做 key 就能保证。
  • 回溯时间精度：Kafka 的时间戳回溯精度是毫秒级，但如果你用的是 RocketMQ 的老版本，精度可能只有秒级。注意版本兼容性。
  • 不要依赖消息队列做永久存储：消息队列的存储是有期限的。Kafka 默认保留 7 天，RocketMQ 默认 72 小时。超过这个时间，消息会被删除。所以长期归档还是得靠数据库。

嗯，消息队列的回溯能力，说白了就是给了你一次「后悔药」的机会。但药不能乱吃，得按剂量来。我建议每个季度做一次容灾演练，模拟消息队列故障，验证回溯方案是否真的能跑通。

我的建议：把回溯方案写成自动化脚本，一键触发。别等到真出事了再手忙脚乱地敲命令。自动化脚本里要包含：停止消费、设置回溯 offset、启动消费、验证数据一致性。每一步都要有日志输出。

消息队列的回溯与重放，是订单簿数据恢复的最后一道防线。用好了，它能帮你从绝大多数灾难中恢复过来。

公众号:蓝海数据掘金营，微信deep3321

← 上一章 📖 返回目录 下一章 →