/**
 * RocketMQ 刷盘配置
 */
// broker.conf
// flushDiskType=SYNC_FLUSH  — 同步刷盘
// flushDiskType=ASYNC_FLUSH — 异步刷盘（默认）

// RocketMQ 异步刷盘源码核心逻辑（简化版）
public class AsyncFlushService extends FlushCommitLogService {
    // 异步刷盘线程，默认每 500ms 执行一次批量刷盘
    private static final long FLUSH_INTERVAL = 500; // ms

    @Override
    public void run() {
        while (!isStopped()) {
            try {
                // 等待刷盘条件触发
                waitForRunning(FLUSH_INTERVAL);
                // 执行批量刷盘
                doFlush();
            } catch (Exception e) {
                log.error("异步刷盘异常", e);
            }
        }
    }

    private void doFlush() {
        // 获取当前写入位置
        CommitLog.CommitLogData data = getCurrentData();
        if (data != null) {
            // 将 PageCache 中的数据强制刷入磁盘
            data.getFileChannel().force(false); // false = 不更新元数据
        }
    }
}

Q: 异步刷盘会不会丢数据？
A: 在 Broker 宕机或断电时，PageCache 中未刷盘的数据会丢失。
   丢失的数据量取决于刷盘间隔（默认 500ms），约几百毫秒的数据。

Q: 如何避免？
A: 1) 使用同步刷盘（牺牲性能）
   2) 配备 UPS 电源 + 电池保护写缓存
   3) 开启主从同步，Master 挂掉后 Slave 接管

/**
 * Kafka 零拷贝 — NIO 实现
 */
// Kafka 底层使用 Java NIO FileChannel.transferTo()
// 该方法映射到操作系统 sendfile 系统调用

public class ZeroCopyExample {

    /**
     * 传统 IO（4 次拷贝）
     */
    public void traditionalIO(File file, Socket socket) throws IOException {
        byte[] buffer = new byte[4096];
        try (FileInputStream fis = new FileInputStream(file);
             OutputStream out = socket.getOutputStream()) {

            int read;
            while ((read = fis.read(buffer)) != -1) {
                out.write(buffer, 0, read);
            }
            out.flush();
        }
        // 性能瓶颈：4 次拷贝、2 次上下文切换
    }

    /**
     * 零拷贝（2 次拷贝）
     */
    public void zeroCopy(File file, Socket socket) throws IOException {
        try (FileChannel fileChannel = new FileInputStream(file).getChannel();
             SocketChannel socketChannel = (SocketChannel) socket.getChannel()) {

            long position = 0;
            long count = fileChannel.size();

            // transferTo() → sendfile 系统调用
            // 数据直接从 PageCache → 网卡
            while (count > 0) {
                long transferred = fileChannel.transferTo(position, count, socketChannel);
                position += transferred;
                count -= transferred;
            }
        }
        // 性能优势：2 次拷贝、0 次用户态参与
    }
}

/**
 * 性能测试对比
 */
public class ZeroCopyBenchmark {
    public static void main(String[] args) throws IOException {
        File file = new File("/tmp/test.data");
        // 创建 100MB 测试文件
        try (RandomAccessFile raf = new RandomAccessFile(file, "rw")) {
            raf.setLength(100 * 1024 * 1024);
        }

        ServerSocket server = new ServerSocket(9999);
        Socket client = new Socket("localhost", 9999);
        Socket serverConn = server.accept();

        long start = System.currentTimeMillis();
        // 传统 IO：约 300ms 传输 100MB
        new ZeroCopyExample().traditionalIO(file, serverConn);
        System.out.println("传统 IO: " + (System.currentTimeMillis() - start) + "ms");

        start = System.currentTimeMillis();
        // 零拷贝：约 80ms 传输 100MB（快 3-4 倍）
        new ZeroCopyExample().zeroCopy(file, serverConn);
        System.out.println("零拷贝: " + (System.currentTimeMillis() - start) + "ms");

        client.close();
        serverConn.close();
        server.close();
    }
}

Q: Kafka 为什么这么快？
A: 三个核心原因：
   1) 顺序写入磁盘（顺序 IO 速度 ≈ 内存随机读）
   2) 零拷贝（sendfile → 2 次拷贝替代 4 次）
   3) 批量压缩 + 批量发送（减少网络开销）

Q: 零拷贝的实现原理？
A: 通过操作系统的 sendfile() 系统调用，数据从磁盘文件直接传输到 
   Socket 缓冲区，跳过应用缓冲区的拷贝。Java 通过 FileChannel.
   transferTo() 封装了此调用。

/**
 * RocketMQ 内置负载均衡策略
 */
// 1. 平均分配（AllocateMessageQueueAveragely）— 默认
// Queue: [Q0, Q1, Q2, Q3, Q4]
// Consumer A: [Q0, Q1, Q2]  ← 多一个
// Consumer B: [Q3, Q4]       ← 少一个

// 2. 轮询分配（AllocateMessageQueueAveragelyByCircle）
// Queue: [Q0, Q1, Q2, Q3, Q4]
// Consumer A: [Q0, Q2, Q4]  ← 每隔一个
// Consumer B: [Q1, Q3]

// 3. 一致性哈希（AllocateMessageQueueConsistentHash）
// 使用一致性哈希环分配，扩容减少重分配

// 4. 按机房（AllocateMessageQueueByMachineRoom）
// 按 Broker 所在机房分配

// 自定义策略
public class CustomAllocateStrategy implements AllocateMessageQueueStrategy {
    @Override
    public List<MessageQueue> allocate(String consumerGroup, String currentCID,
                                       List<MessageQueue> mqAll,
                                       List<String> cidAll) {
        // 自定义分配逻辑
        // 注意：所有消费者的分配结果需要一致
        List<MessageQueue> result = new ArrayList<>();
        int index = cidAll.indexOf(currentCID);
        for (int i = index; i < mqAll.size(); i += cidAll.size()) {
            result.add(mqAll.get(i));
        }
        return result;
    }

    @Override
    public String getName() {
        return "CUSTOM_STRATEGY";
    }
}

// 使用自定义策略
DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("group");
consumer.setAllocateMessageQueueStrategy(new CustomAllocateStrategy());

/**
 * Kafka 分区分配策略
 */
// 1. RangeAssignor（范围分配）— 默认
// Topic: T1(3 partitions), T2(4 partitions)
// Consumer A: T1-P0, T1-P1, T2-P0, T2-P1  ← 每个 Topic 均分
// Consumer B: T1-P2, T2-P2, T2-P3

// 2. RoundRobinAssignor（轮询）
// Consumer A: T1-P0, T1-P2, T2-P1, T2-P3
// Consumer B: T1-P1, T2-P0, T2-P2

// 3. StickyAssignor（粘性分配）
// 在 rebalance 时尽量保留已有分配，减少分区迁移
// 避免全部重新分配造成大量重复消费

// 4. CooperativeStickyAssignor（合作式粘性分配）
// Kafka 2.4+ 支持，增量式 rebalance
// 不再 stop-the-world，逐个分区迁移

// 配置策略
Properties props = new Properties();
props.put("partition.assignment.strategy",
    "org.apache.kafka.clients.consumer.CooperativeStickyAssignor");

/**
 * Rebalance 发生的场景：
 * 1. 消费者加入/退出消费组
 * 2. Topic 分区数变化
 * 3. 消费者超时未发送心跳
 */
public class RebalanceListenerExample {

    @Autowired
    private KafkaConsumer<String, String> consumer;

    /**
     * Kafka 监听 Rebalance 事件
     */
    public void consumeWithRebalance() {
        consumer.subscribe(Arrays.asList("topic"), new ConsumerRebalanceListener() {
            @Override
            public void onPartitionsRevoked(Collection<TopicPartition> partitions) {
                // Rebalance 前调用：提交偏移量
                log.info("分区被回收: {}", partitions);
                consumer.commitSync();
            }

            @Override
            public void onPartitionsAssigned(Collection<TopicPartition> partitions) {
                // Rebalance 后调用：新分区已分配
                log.info("新分配分区: {}", partitions);
                // 可以在这里 seek 到特定位置
            }
        });
    }

    /**
     * RocketMQ 监听 Rebalance（变化通知模式）
     */
    public DefaultMQPushConsumer createRocketMQConsumer() {
        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("group");
        // RocketMQ 默认自动处理 Rebalance
        // 可通过 setAllocateMessageQueueStrategy 控制策略
        return consumer;
    }
}

/**
 * 分区数如何确定
 */
public class PartitionDesign {

    // 经验公式：
    // 分区数 = max(生产 TPS / 单分区生产 TPS, 消费 TPS / 单分区消费 TPS)

    // 示例：
    // 预期生产 TPS = 50000，单分区生产 TPS = 10000
    // 预期消费 TPS = 20000，单分区消费 TPS = 5000
    // 分区数 = max(50000/10000, 20000/5000) = max(5, 4) = 5

    // 但还需要考虑：
    // 1. 分区太多 → 文件句柄多、Rebalance 慢
    // 2. 分区太少 → 扩展性差、消费者无法扩容

    // 推荐值：
    // - RocketMQ 默认 8，建议 8-32
    // - Kafka 每个 Topic 建议 3-10 个分区/Broker
}

## 面试核心考点

Q: 分区如何保证顺序？
A: 同一分区内消息有序，不同分区之间无序。
   保证业务的顺序性需要将同一业务 ID 的消息路由到同一分区。

Q: 全局有序如何实现？
A: 设置分区数为 1，但完全牺牲了并行性，不推荐。
   实际上只需要保证"业务有序"，不追求全局有序。

Q: 分区可以动态增加吗？
A: Kafka 可以增加分区，但无法减少。
   RocketMQ 可以通过 updateTopic 调整读/写队列数。

/**
 * 1. 生产者发送（DefaultMQProducer.send() 简化流程）
 */
public SendResult send(Message msg) {
    // 1. 获取 Topic 路由信息
    TopicPublishInfo topicInfo = tryToFindTopicPublishInfo(msg.getTopic());

    // 2. 选择消息队列（负载均衡）
    MessageQueue queue = topicInfo.selectOneMessageQueue(lastBrokerName);

    // 3. 发送消息
    SendResult result = sendKernelImpl(msg, queue, communicationMode);

    return result;
}

/**
 * 2. Broker 存储（CommitLog.putMessage() 简化流程）
 */
public PutMessageResult putMessage(MessageExtBrokerInner msg) {
    // 1. 获取当前文件（CommitLog）
    MapedFile mapedFile = this.mapedFileQueue.getLastMapedFile();

    // 2. 写入内存（PageCache）
    AppendMessageResult result = mapedFile.appendMessage(msg);

    // 3. 同步刷盘（如果配置了 SYNC_FLUSH）
    if (flushDiskType == FlushDiskType.SYNC_FLUSH) {
        flushCommitLog();
    }

    // 4. 构建 ConsumeQueue（异步）
    this.defaultMessageStore.storeStatsService
        .getPutMessageEntireTimeMax();

    return new PutMessageResult(PutMessageStatus.PUT_OK, result);
}

/**
 * 3. 消费者拉取（长轮询 PullMessageProcessor 简化流程）
 */
public RemotingCommand processRequest(ChannelHandlerContext ctx,
                                       PullMessageRequestHeader request) {
    // 1. 检查消息是否存在
    GetMessageResult getMessageResult =
        messageStore.getMessage(request.getConsumerGroup(),
            request.getTopic(), request.getQueueId(),
            request.getOffset(), request.getMaxMsgNums());

    // 2. 如果消息未到达，挂起请求（长轮询）
    if (getMessageResult.getStatus() == GetMessageStatus.NO_MESSAGE_IN_QUEUE) {
        this.pullRequestHoldService
            .suspendPullRequest(request.getTopic(), request.getQueueId(),
                new PullRequest(ctx, request));
        // 挂起默认 30 秒，有新消息到达时唤醒
    }

    return response;
}

/**
 * Kafka 生产者核心流程
 */
// 1. Producer.send() → 序列化 → 分区 → 加入批次
// 2. Sender 线程：批次打包 → 发送到 Leader 分区
// 3. Leader 写入日志 → Follower 同步 → ACK 返回

// 关键：批次发送（batch.size, linger.ms 控制攒批时间）
Properties props = new Properties();
props.put("batch.size", 16384);        // 批次大小（字节）
props.put("linger.ms", 5);             // 攒批等待时间
props.put("compression.type", "snappy"); // 批次压缩

// 底层：异步发送 + 批量确认
// 一个批次内的消息，一次网络请求完成

典型消息延迟分布（毫秒级）：

Producer 序列化:         0.01ms
Producer 网络发送:       0.5ms (同机房内网)
Broker 写入 PageCache:   0.01ms
Broker 刷盘(异步):       0.5ms (异步)
Broker 构建索引:         0.1ms
Consumer 拉取:           0.5ms (长轮询有等待)
Consumer 反序列化:       0.01ms
Consumer 业务处理:       10-100ms (主要耗时)
────────────────────────────────────
端到端延迟:              20-100ms 级别

Q1: 消息队列解决了什么问题，带来了什么问题？
A: 解决：解耦、异步、削峰
   带来：系统复杂度上升、消息丢失风险、重复消费、顺序问题、一致性难题

Q2: 如何保证消息不丢失？
A: 三环节保证：生产端(ACK+重试) → Broker(持久化+副本) → 消费端(手动ACK+DLQ)

Q3: 如何处理重复消息？
A: 接口幂等性：唯一 ID 去重、数据库唯一键、状态机校验

Q4: RocketMQ 事务消息原理？
A: 半消息 → 执行本地事务 → Commit/Rollback → 回查机制保证最终一致性

Q5: Kafka 为什么吞吐量高？
A: 顺序写入、零拷贝、批量压缩、分区并行

Q6: 消息顺序性如何保证？
A: 同一业务 ID 路由到同一分区 + 分区内单线程消费

Q7: 消息堆积了怎么办？
A: 
   1. 排查消费者是否故障
   2. 扩容消费者（确保队列数 ≥ 消费者数）
   3. 排查消费逻辑是否阻塞
   4. 紧急情况：重置偏移量跳过或临时 Topic 分流

Q8: 设计一个秒杀系统，如何使用 MQ？
A:
   1. 前端请求直接写入 MQ（削峰）
   2. 消费者以可控速率处理订单
   3. 结果异步通知用户

Q9: 100 万 QPS 的日志系统如何设计？
A:
   1. 使用 Kafka（高吞吐）
   2. 生产者侧批量发送+压缩
   3. 消费者侧批量写入 HDFS/ES
   4. 合理设置分区数（Broker 数 × 分区数）

Q10: 分布式事务场景：下单扣库存怎么保证一致性？
A:
   方案一：RocketMQ 事务消息
   方案二：本地消息表 + 定时任务补偿
   方案三：TCC（Try-Confirm-Cancel）框架

### ❌ 坑 1：自动 ACK
// 消费者收到消息立即 ACK，还没处理业务就确认了
// 消费者宕机 → 消息丢失

### ❌ 坑 2：消费者异常时吞掉异常
try {
    process(msg);
} catch (Exception e) {
    log.error(e); // 只打日志，不返回失败
    // 消息已 ACK → 丢失
}
// ✅ 正确：返回重试状态，让 MQ 重新投递

### ❌ 坑 3：积压时扩容消费者不管用
// 原因：队列数 < 消费者数，多余的消费者无队列可消费
// ✅ 创建 Topic 时就预留足够的队列（如 32 个）

### ❌ 坑 4：消息体过大
// 发送 100MB 的消息 → 网络拥塞 + GC 压力
// ✅ 限制消息体 < 1MB，大文件用 OSS 传递 URL

### ❌ 坑 5：广播模式用在关键业务
// 广播消费重启后偏移量不保留
// 广播模式下每个消费者收到的消息量 = 总消息数 × 消费者数

| 陷阱 | 影响 | 解决方案 |
|------|------|---------|
| 消息体过大（>1MB） | 网络拥塞、GC 压力 | 压缩或存 OSS 传路径 |
| Topic 数量过多 | 文件句柄耗尽、性能下降 | 控制 Topic 数量 |
| 分区数过多 | Rebalance 慢、文件碎片 | 建议 8-32 分区 |
| 消费者数 > 队列数 | 部分消费者闲置 | 队列数 ≥ 消费者数 |
| 消费逻辑中做耗时 IO | 消费速率下降、堆积 | 异步化消费逻辑 |
| 同步刷盘 + 低配磁盘 | TPS 降到几千 | 换 SSD 或改为异步刷盘 |