// 分布式消息队列架构：
//
// [Producer 1] [Producer 2] [Producer 3]
//       ↓           ↓           ↓
// ┌─────────────────────────────────────┐
// │         Broker Cluster              │
// │  ┌─────────┐ ┌─────────┐ ┌────────┐ │
// │  │ Broker 1│ │ Broker 2│ │Broker 3│ │
// │  │ P0(L)   │ │ P1(L)   │ │P2(L)   │ │
// │  │ P1(F)   │ │ P2(F)   │ │P0(F)   │ │
// │  │ P2(F)   │ │ P0(F)   │ │P1(F)   │ │
// │  └─────────┘ └─────────┘ └────────┘ │
// └─────────────────────────────────────┘
//       ↓           ↓           ↓
// [Consumer A1] [Consumer A2] [Consumer A3]
//        Group A (每个Consumer消费不同的Partition)
//
// 核心概念：
// Topic：消息的逻辑分类（如 order-events）
// Partition：Topic 的物理分片，每个 Partition 有序
// Broker：消息队列的服务器节点
// Producer：消息生产者
// Consumer Group：一组消费者共同消费一个 Topic
// Leader：Partition 的主副本，处理读写
// Follower：Partition 的从副本，同步数据

// 单 Partition 的局限：
// 1. 吞吐量受限于单机磁盘和网络
// 2. 消费者数量不能超过 Partition 数
// 3. 无法水平扩展

// 多 Partition 的优势：
// 1. 吞吐量线性扩展（N个Partition = N倍吞吐）
// 2. 消费者可以并行消费
// 3. 数据分布到多个 Broker

// Partition 分配：
// Topic: order-events, 3个Partition
// P0 → Leader on Broker 1, Follower on Broker 2, 3
// P1 → Leader on Broker 2, Follower on Broker 3, 1
// P2 → Leader on Broker 3, Follower on Broker 1, 2
//
// 每个 Broker 都有 Leader 和 Follower
// 负载均衡：读写分散到不同 Broker

// Producer 发送消息时指定 Partition：
//
// 方式1：指定 Partition
// producer.send({ topic: 'orders', partition: 0, value: 'msg' })
//
// 方式2：指定 Key（相同Key → 相同Partition）
// producer.send({ topic: 'orders', key: 'order-123', value: 'msg' })
// → hash('order-123') % 3 = 1 → Partition 1
//
// 方式3：轮询（默认）
// 消息依次发送到 P0, P1, P2, P0, P1, P2, ...
//
// Key 的重要性：
// 相同 Key 的消息进入同一 Partition → 保证顺序
// 例：同一订单的事件需要按顺序处理
// key = orderId → 同一订单的事件在同一 Partition → 有序

// Kafka 高性能的秘密：顺序写磁盘
//
// 磁盘访问速度：
// 随机写：~100 IOPS，每次 ~0.1ms
// 顺序写：~600MB/s，几乎无延迟
//
// Kafka 的设计：
// 每个 Partition 是一个追加写的日志文件
// 新消息总是追加到文件末尾 → 顺序写
//
// 存储结构：
// 00000000000000000000.log      (日志文件)
// 00000000000000000000.index    (偏移量索引)
// 00000000000000000000.timeindex (时间索引)
//
// 文件名是起始 Offset（20位数字）
// 当 Segment 达到 1GB 或 7天，创建新 Segment

// 消息在日志文件中的存储格式：
// Offset(8B) | MessageSize(4B) | CRC(4B) | Magic(1B) | Key | Value
//
// Offset：消息在 Partition 中的全局唯一序号（8字节）
// MessageSize：消息总大小
// CRC：校验和，检测数据损坏
// Key：可选，用于 Partition 路由
// Value：消息内容

// 偏移量索引：
// 将 Offset 映射到文件位置（position）
// 稀疏索引：不是每条消息都有索引条目
// 二分查找定位 Offset → 读取消息
//
// 查找过程：
// 1. 二分查找索引文件，找到最近的 Offset
// 2. 从文件位置开始顺序扫描
// 3. 找到目标 Offset 的消息

// 传统数据发送（4次拷贝，4次上下文切换）：
// 磁盘 → 内核缓冲区 → 用户缓冲区 → Socket缓冲区 → 网卡
// read(): 磁盘 → 内核 → 用户
// write(): 用户 → 内核 → Socket
//
// 零拷贝 sendfile()（2次拷贝，2次上下文切换）：
// 磁盘 → 内核缓冲区 → 网卡
// sendfile(): 磁盘 → 内核 → Socket（不经过用户空间）
//
// 性能提升：
// 传统方式：~200MB/s
// 零拷贝：~600MB/s（3倍提升）
//
// Kafka 的消费者拉取消息时使用零拷贝
// 消息从磁盘直接发送到网络，不经过应用层

// 每个 Partition 有 1 个 Leader 和 N 个 Follower
//
// 写入流程：
// 1. Producer 发送消息到 Leader
// 2. Leader 追加消息到本地日志
// 3. Follower 从 Leader 拉取消息（异步复制）
// 4. Follower 发送 ACK 给 Leader
// 5. Leader 更新 HW（High Watermark）
// 6. Producer 收到成功响应
//
// 读取流程：
// 消费者只能读取 Leader
// 只能读取 HW 之前的消息（已复制的消息）
//
// 术语：
// LEO（Log End Offset）：日志末尾的 Offset
// HW（High Watermark）：所有 ISR 中最小的 LEO
// 消费者只能看到 HW 之前的数据（保证一致性）

// ISR 是与 Leader 保持同步的副本集合
//
// ISR 的判断标准：
// Follower 的 LEO 与 Leader 的 LEO 差距 < 阈值
// 且最近一次同步时间 < 超时时间
//
// ISR 的作用：
// 1. 决定 HW：HW = min(ISR 中所有副本的 LEO)
// 2. Leader 选举：新 Leader 只能从 ISR 中选择
// 3. 平衡一致性和可用性：
//    - ISR 大 → 一致性好，但可用性低（等待更多副本）
//    - ISR 小 → 可用性好，但一致性差
//
// acks 配置：
// acks=0：Producer 不等待确认（最快，可能丢消息）
// acks=1：Leader 写入成功即返回（可能丢消息，如果 Leader 宕机）
// acks=all：所有 ISR 写入成功才返回（最安全，延迟最高）

// Leader 宕机时的处理：
//
// 1. Controller（特殊 Broker）检测到 Leader 宕机
// 2. 从 ISR 中选择新的 Leader
// 3. 通知所有 Broker 更新 Partition 的 Leader 信息
// 4. Producer 和 Consumer 连接到新 Leader
//
// 如果 ISR 为空：
// 1. 从所有副本中选择 LEO 最大的作为 Leader
// 2. 可能丢失少量数据（未复制到 ISR 的消息）
// 3. 或者等待 ISR 中的副本恢复（unclean.leader.election.enable=false）
//
// 选举时间：通常 5-30 秒

// Consumer Group 实现消息的负载均衡
//
// 规则：
// 1. 一个 Partition 只能被 Group 中的一个 Consumer 消费
// 2. 一个 Consumer 可以消费多个 Partition
// 3. 不同 Group 独立消费全部消息（广播）
//
// 示例：
// Topic: order-events, 3个Partition
//
// Group A（订单处理）：
//   Consumer A1 → P0, P1
//   Consumer A2 → P2
//
// Group B（库存处理）：
//   Consumer B1 → P0
//   Consumer B2 → P1, P2
//
// 两个 Group 独立消费全部消息
// 每个 Group 内部负载均衡

// Rebalance（重平衡）：
// 当 Consumer 加入或离开 Group 时触发
// 重新分配 Partition 给 Consumer
//
// 过程：
// 1. 所有 Consumer 向 Group Coordinator 发送心跳
// 2. Coordinator 检测 Consumer 变化
// 3. 触发 Rebalance
// 4. 所有 Consumer 暂停消费
// 5. Coordinator 重新分配 Partition
// 6. 通知所有 Consumer 新的分配
// 7. 恢复消费
//
// Rebalance 期间消费暂停 → 影响可用性
// 优化：Cooperative Sticky Assignment（增量式 Rebalance）

// Kafka 的高吞吐优化：
//
// 1. 顺序写磁盘（已述）
// 2. 零拷贝 sendfile（已述）
// 3. 批量发送：
//    Producer 积累一批消息后一次性发送
//    batch.size = 16KB（默认）
//    linger.ms = 0（默认，立即发送）
//    增大 linger.ms 可以提高批量大小
//
// 4. 数据压缩：
//    Producer 端压缩一批消息
//    支持 gzip、snappy、lz4、zstd
//    Broker 和 Consumer 不解压（保持压缩存储）
//    Consumer 消费时解压
//
// 5. 页缓存（Page Cache）：
//    利用操作系统的页缓存
//    消息先写入页缓存，异步刷盘
//    消费者读取时大概率命中页缓存
//    避免 JVM GC 问题（Kafka 不使用 JVM 内存缓存数据）

# Broker 配置
broker.id: 1
num.partitions: 6           # 默认 Partition 数
default.replication.factor: 3 # 默认副本数
log.segment.bytes: 1073741824 # 1GB per segment
log.retention.hours: 168     # 保留7天
log.retention.bytes: -1      # 不按大小删除
num.io.threads: 8            # IO 线程数
num.network.threads: 3       # 网络线程数

# Producer 配置
acks: all                   # 所有 ISR 确认
retries: 3                  # 重试次数
batch.size: 16384           # 批量大小 16KB
linger.ms: 5                # 等待5ms凑批
compression.type: lz4       # 压缩算法
max.in.flight.requests.per.connection: 5

# Consumer 配置
group.id: order-processor
enable.auto.commit: false   # 手动提交 Offset
auto.offset.reset: earliest # 从头开始消费
max.poll.records: 500       # 每次拉取最多500条
fetch.min.bytes: 1          # 最小拉取字节数
fetch.max.wait.ms: 500      # 最大等待时间

// 场景：订单事件需要按顺序处理
// OrderCreated → OrderPaid → OrderShipped → OrderDelivered
//
// 方案：使用 orderId 作为 Partition Key
// 同一订单的所有事件进入同一 Partition
// 同一 Partition 内消息有序
// 同一 Partition 只被一个 Consumer 消费
// → 保证同一订单的事件按顺序处理

// Producer:
producer.send({
    topic: 'order-events',
    key: order.id,  // 相同 orderId → 相同 Partition
    value: JSON.stringify({ type: 'OrderCreated', ... })
});

// 注意：
// 不同订单的事件可能在不同 Partition
// 不同 Partition 之间没有全局顺序保证
// 如果需要全局有序 → 只用一个 Partition（吞吐量受限）