美团三面：秒杀开始后MQ积压1000万条消息，给出5小时内恢复系统正常的应急方案

安宇雨 - 随手采集
2025-08-04 16:57:12
随手采集
0000-未整理-等待研究

面试题

场景：

秒杀活动开始后，订单MQ积压1000万条，消费者处理速度200条/秒。

原因：

消费能力不足：单消费者200条/秒 × 50实例 = 1万条/秒
消息类型混杂：核心订单消息被日志消息阻塞
拉取策略缺陷：单次拉取量过大导致线程阻塞

问题：

给出5小时内恢复系统正常的应急方案。

五步急救方案：5小时清零积压

步骤1：水平扩容（30分钟）

操作命令

# Kafka分区扩容kafka-topics.sh --alter --topic orders \--partitions 200 --bootstrap-server kafka1:9092# 消费者实例扩容 (K8s)kubectl scale deployment order-consumer --replicas=100

步骤2：消息降级（紧急止血）

降级策略：

// 消费者过滤非核心消息public void handleMessage(Message msg) {if (msg.getPriority() < Message.PRIORITY_HIGH) {    // 非核心消息直接ACK不处理    msg.acknowledge();     return;}processOrder(msg);}

步骤3：流量控制（防雪崩）

RocketMQ限流配置：

// 设置单队列拉取阈值consumer.setPullThresholdForQueue(100); // 单队列积压>100则暂停拉取// 调整单次拉取量consumer.setPullBatchSize(20); // 每次拉20条

Kafka优化配置：

# consumer.propertiesmax.poll.records=50          # 单次拉取50条fetch.max.bytes=1048576       # 1MB/次max.partition.fetch.bytes=1048576

步骤4：临时转储（异步恢复）

文件转储代码：

def dump_messages():    with open('/data/backup/msgs.bin', 'wb') as f:        while True:            msgs = consumer.poll(1000)  # 拉取1000条            if not msgs: break            # 二进制批量写入            f.write(msgpack.packb(msgs))             # 跳过非核心消息            filtered = [m for m in msgs if m.priority > 1]            process_messages(filtered)

步骤5：终极方案 - 并行爆破（核心操作）

操作流程：

1.创建新Topic

kafka-topics --create --topic orders_emerg \--partitions 300 --replication-factor 3

2.启动消息转发

// 积压消息转发服务while (true) {    Message msg = originalConsumer.receive();    emergencyProducer.send(msg); // 转发到新Topic}

3.启动批量消费者

// 批量消费者配置props.put("max.poll.records", 1000); // 单次拉1000条props.put("fetch.max.bytes", 5242880); // 5MB/次// 并行处理线程池ExecutorService pool = Executors.newFixedThreadPool(32);consumer.subscribe("orders_emerg");while (running) {    ConsumerRecords records = consumer.poll(1000);    pool.submit(() -> batchProcess(records)); // 批量提交}