报警群炸锅！秒杀开始后MQ积压1000万条消息，如何5小时内紧急恢复系统正常？

场景：

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

原因：

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

五步急救方案：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.properties
max.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：终极方案 - 并行爆破（核心操作）

操作流程：

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

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

// 批量消费者配置
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)); // 批量提交
}

救援时间线（5小时倒计时）

总结：架构师急救手册

核心原则：

1. 分而治之：拆积压 → 多Topic → 并行消费
2. 弃卒保帅：非核心消息直接丢弃
3. 空间换时间：文件转储争取处理时间
4. 超限爆破：临时超配资源应对峰值

终极思考题：

当资源受限（只能扩容3倍）时，如何组合方案实现5小时清零？
提示：降级+扩容+批量处理三合一（评论区等你方案！）