kafka 源码与OS底层

            ┌─────────────────────────────────────────┐            │  bucket[0] bucket[1] ... bucket[n-1]    │  ← 每个 bucket 是一个 TimerTaskList（双向链表）            └─────────────────────────────────────────┘                ▲                │ currentTime（每 tickMs 转一格）tickMs = 1ms, wheelSize = 20任务 expirationMs = currentTime + delay桶位 = (expirationMs / tickMs) % wheelSize

Level 0: tickMs=1ms,    wheelSize=20  → 覆盖 0~20msLevel 1: tickMs=20ms,   wheelSize=20  → 覆盖 0~400msLevel 2: tickMs=400ms,  wheelSize=20  → 覆盖 0~8sLevel 3: tickMs=8s,     wheelSize=20  → 覆盖 0~160s... 按需向上溢出，新建更高层

def add(task: TimerTask): Boolean = {  val expiration = task.expirationMs  if (expiration < currentTime + tickMs) {    // 已经到期，直接执行    return false  } else if (expiration < currentTime + tickMs * wheelSize) {    // 当前层能容纳，放对应 bucket    val bucket = buckets((expiration / tickMs) % wheelSize)    bucket.add(task)    return true  } else {    // 溢出到上层    overflowWheel.add(task)  }}

┌─ Producer 收到 acks=all 请求 ────────────────────────────────────┐│                                                                  ││  ReplicaManager.appendRecords                                    ││      ↓                                                           ││  写本地 → 构造 DelayedProduce(timeout=requestTimeoutMs)           ││      ↓                                                           ││  producePurgatory.tryCompleteElseWatch(DelayedProduce, keys)     ││      │                                                           ││      ├─ tryComplete()：检查 HW 是否已推进过该 offset               ││      │     ├─ 是 → 立即响应                                       ││      │     └─ 否 → 加入对应 TopicPartition 的 watch list          ││      │                                                           ││      └─ 同时把任务塞进 SystemTimer 的时间轮                        ││            （超时兜底，避免 watch 永远不触发）                      ││                                                                  ││  当 Follower Fetch 上报 LEO，Leader 推 HW                         ││      ↓                                                           ││  partition.tryCompleteDelayedRequests()                          ││      ↓                                                           ││  从 watch list 取出 DelayedProduce.tryComplete() → 完成响应       │└──────────────────────────────────────────────────────────────────┘

RecordAccumulator├── ConcurrentMap<TopicPartition, Deque<ProducerBatch>>   // 每个 TP 一个队列├── BufferPool                                             // 内存池（buffer.memory）└── IncompleteBatches                                      // 跟踪未完成 batch（abort 时用）ProducerBatch├── MemoryRecordsBuilder      // 累积消息的可变 buffer├── List<Thunk>               // 每个 send 注册的回调├── createdMs / drainedMs     // 时间统计├── retryCount                // 重试次数└── producerId / sequence     // 幂等性元数据

// org.apache.kafka.clients.producer.KafkaProducerpublic Future<RecordMetadata> send(ProducerRecord<K, V> record, Callback cb) {    // 1. 拦截器 → onSend    // 2. 序列化 key/value    // 3. 分区器选 partition    // 4. accumulator.append(...)  ← 关键    RecordAccumulator.RecordAppendResult result = accumulator.append(        topicPartition, timestamp, serializedKey, serializedValue,        headers, callback, maxTimeToBlock /* max.block.ms */, ...);    if (result.batchIsFull || result.newBatchCreated) {        sender.wakeup();   // 唤醒 Sender 线程    }    return result.future;}

public RecordAppendResult append(TopicPartition tp, ...) {    Deque<ProducerBatch> dq = getOrCreateDeque(tp);    synchronized (dq) {           // 单 TP 队列粒度的锁        // 尝试追加到队尾 batch        RecordAppendResult result = tryAppend(...);        if (result != null) return result;    }    // 队尾 batch 装不下 → 申请新 buffer（可能阻塞 max.block.ms）    ByteBuffer buffer = bufferPool.allocate(size, maxTimeToBlock);    synchronized (dq) {        // 双重检查（DCL 模式）：可能其他线程已经创建了新 batch        result = tryAppend(...);        if (result != null) {            bufferPool.deallocate(buffer);            return result;        }        // 创建新 batch        ProducerBatch batch = new ProducerBatch(tp, recordsBuilder, now);        batch.tryAppend(...);        dq.addLast(batch);        return new RecordAppendResult(future, /*batchIsFull*/ false, /*newBatchCreated*/ true);    }}

// Sender 单线程循环void run() {    while (running) {        // 1. 检查哪些 broker 准备好了        ReadyCheckResult ready = accumulator.ready(metadata, now);        // ready 条件：batch 满 / linger 到 / metadata 可用 / 重试时间到        // 2. 按 broker 维度 drain（关键）        Map<Integer, List<ProducerBatch>> batches =             accumulator.drain(metadata, ready.readyNodes, maxRequestSize, now);        // 3. 处理过期 batch（delivery.timeout.ms 到了）        List<ProducerBatch> expired = accumulator.expiredBatches(now);        expired.forEach(b -> b.done(ERROR));        // 4. 构造 ProduceRequest 并发送        for (Map.Entry<Integer, List<ProducerBatch>> e : batches.entrySet()) {            sendProduceRequest(e.getKey(), e.getValue());        }        // 5. 处理 NetworkClient 的回调        client.poll(timeout, now);    }}

public Map<Integer, List<ProducerBatch>> drain(...) {    Map<Integer, List<ProducerBatch>> batches = new HashMap<>();    for (Node node : readyNodes) {        List<ProducerBatch> ready = drainBatchesForOneNode(node, ...);        batches.put(node.id(), ready);    }    return batches;}private List<ProducerBatch> drainBatchesForOneNode(Node node, int maxSize, long now) {    int size = 0;    List<ProducerBatch> ready = new ArrayList<>();    List<PartitionInfo> parts = metadata.partitionsForNode(node.id());    // 关键：从随机 offset 开始遍历，避免每次都从同一个 TP 抢资源（饿死后面的 TP）    int start = drainIndex.getAndIncrement() % parts.size();    int i = start;    do {        TopicPartition tp = ...;        Deque<ProducerBatch> dq = getDeque(tp);        synchronized (dq) {            ProducerBatch first = dq.peekFirst();            if (first == null || isMuted(tp)) continue;            if (size + first.estimatedSize() > maxSize && !ready.isEmpty()) break;            ProducerBatch batch = dq.pollFirst();            batch.close();   // 关闭 builder，禁止再追加            size += batch.estimatedSize();            ready.add(batch);        }        i = (i + 1) % parts.size();    } while (i != start);    return ready;}

                ┌──────────── Acceptor (1 个线程) ─────────────┐                │ ServerSocketChannel.accept() 接受新连接       │                │ 轮询分配给 Processor                          │                └───────────────────┬───────────────────────────┘                                    │                ┌───────────────────┼───────────────────┐                ▼                   ▼                   ▼         Processor[0]        Processor[1]        ...  Processor[N-1]         (num.network.threads 个 = 默认 3)         │ 每个有自己的 Selector         │ 处理 read / write 事件         └─→ RequestChannel.requestQueue (有界阻塞队列)                                    │                ┌───────────────────┼───────────────────┐                ▼                   ▼                   ▼         KafkaRequestHandler[0]  ...  KafkaRequestHandler[M-1]         (num.io.threads 个 = 默认 8)         │ poll requestQueue，调用 KafkaApis.handle(request)         │ 处理完写到 Processor 的 responseQueue         ▼         Processor 在下一轮 Selector 循环把 response 写回

KafkaProducer (Sender 线程)   ↓NetworkClient (单 Selector，多 broker)   ├── Map<NodeId, KafkaChannel>          ← 每个 broker 一个长连接   ├── InFlightRequests                    ← Map<NodeId, Deque<NetworkSend>>   │     - 每个 broker 一个 deque   │     - 上限 = max.in.flight.requests.per.connection   └── Selector (NIO)         ├── send queue         └── completed receives

LogManager├── Map<TopicPartition, UnifiedLog> currentLogs├── 后台线程：│   ├── log-cleaner-* (compact)│   ├── kafka-log-retention (delete by time/size)│   └── kafka-log-flusher (定期 fsync，默认 Long.MAX_VALUE 即不主动 fsync)└── 加载/卸载 Log 时的 lock

class UnifiedLog {    private val lock = new Object()                           // 大锁，写入串行    @volatile var nextOffsetMetadata: LogOffsetMetadata = ... // LEO，volatile 保证读不需要锁    var activeSegment: LogSegment = ...    val segments: ConcurrentNavigableMap[Long, LogSegment]    // 跳表，快速二分定位    def append(records: MemoryRecords): LogAppendInfo = {        lock.synchronized {            // 1. 校验消息（CRC、size、producer state）            // 2. 决定是否 roll segment            // 3. activeSegment.append(...)            // 4. 更新 LEO        }    }    def read(startOffset: Long, maxLength: Int): FetchDataInfo = {        // 读路径不加锁（segments 是 ConcurrentNavigableMap，LEO 是 volatile）        val segment = segments.floorEntry(startOffset).getValue        segment.read(...)    }}

应用 write(fd, buf, len)   ↓内核：把数据 copy 到 PageCache（脏页）   ↓write() 返回（此时数据还没落盘）   ↓内核后台 flush (kworker / pdflush)   ↓ 受以下参数控制   │ vm.dirty_background_ratio  → 后台开始刷盘的脏页占比（默认 10%）   │ vm.dirty_ratio              → 强制同步刷盘的脏页占比（默认 20% 后阻塞 write）   │ vm.dirty_expire_centisecs   → 脏页最大停留时间（默认 30s）   │ vm.dirty_writeback_centisecs → 后台 flush 周期（默认 5s）   ↓真正写到磁盘

应用 read(fd, buf, len)   ↓内核检查 PageCache   ├── HIT  → 从 PageCache 直接 copy 到用户 buf   └── MISS → 从磁盘读到 PageCache → 再 copy 到用户 buf              (会预读 readahead 后续 page)

磁盘 ──DMA──► PageCache (内核空间)            │            └─► copy ──► 应用 buffer (用户空间)                                  │                                  └─► copy ──► Socket buffer (内核空间)                                                       │                                                       └─DMA──► 网卡4 次拷贝 + 4 次上下文切换（user/kernel 来回）

sendfile(out_fd, in_fd, offset, count)磁盘 ──DMA──► PageCache (内核空间) ──► Socket buffer ──DMA──► 网卡                       │                       └─ 仅"描述符 + 长度"传递，数据不拷贝（DMA scatter-gather, 2.4+）2 次 DMA + 0 次 CPU 拷贝 + 2 次上下文切换

SSL 启用时：磁盘 ──DMA──► PageCache ──copy──► 用户 buffer                                       │                                       └─ SSL.encrypt() 加密                                       │                                       └──copy──► Socket buffer ──DMA──► 网卡退化成 4 次拷贝。

/dev/nvme0n1   /data/kafka   xfs   defaults,noatime,nodiratime,nobarrier   0 0

单层时间轮：  tickMs=1ms, wheelSize=20 → 覆盖 0~20ms分层时间轮（Kafka 实现）：  Level 0: tickMs=1ms,    wheelSize=20  → 0~20ms  Level 1: tickMs=20ms,   wheelSize=20  → 0~400ms  Level 2: tickMs=400ms,  wheelSize=20  → 0~8s  Level 3: tickMs=8s,     wheelSize=20  → 0~160s  Level 4: tickMs=160s,   wheelSize=20  → 0~3200s ≈ 53分钟每加一层，覆盖范围 ×20。1 小时（3600s）需要 5 层 → 5 × 20 = 100 个桶。若优化到 4 层（最高 tickMs=320s），共 80 个桶。对比 1 小时 1ms 精度的单层方案：3600,000 个桶 → 节省 4500x 内存。每个 task 添加：O(1)任务过期降级：上层 cascade 到下层 O(levels) ≈ O(1)配合 DelayQueue（只放 bucket 不放 task）→ DelayQueue 操作 O(log levels)

1) 先确认是 GC 抖还是其他   - jstat -gcutil <pid> 1000 看 GC 暂停 > 50ms   - JMX IsrShrinksPerSec / IsrExpandsPerSec 飙升2) JVM 调优   -Xms8g -Xmx8g                       (固定不动态扩缩)   -XX:+UseG1GC   -XX:MaxGCPauseMillis=20   -XX:G1HeapRegionSize=16m   -XX:InitiatingHeapOccupancyPercent=35   -XX:+ParallelRefProcEnabled   -XX:+AlwaysPreTouch                  (启动期预触摸内存)3) 不要用 ZGC（除非堆 ≥ 16GB）   - Kafka 大量数据走堆外（NIO/DirectBuffer + mmap）   - 堆里都是短命对象，G1 年轻代复制就够   - ZGC 读屏障开销让吞吐降 10%+4) 关闭 -XX:+DisableExplicitGC   - Kafka 内部依赖 System.gc() 触发 DirectBuffer 回收   - 关闭会导致堆外 OOM5) GC 抖根因   - 大消息 / 大 batch 进入老年代 → 调小消息上限   - DirectBuffer 累积 → 检查 Producer / Consumer SDK 是否泄漏   - 同时压测验证调优效果

现状：  传统 sendfile：磁盘 → PageCache → 网卡（2 次 DMA + 0 次 CPU 拷贝）  SSL 启用后：磁盘 → PageCache → 用户 buffer（CPU 加密）→ Socket buffer → 网卡            (4 次拷贝 + CPU 加密开销，吞吐降 30~50%)权衡方案：  1) 内网 PLAINTEXT，南北向 SSL     - Broker 间副本同步用 PLAINTEXT（机房内可控）     - 公网入口 SSL，加 broker 数量 / 网卡补吞吐  2) kTLS（Linux 4.13+，Kafka 3.0+）     - 加密下沉到内核，恢复零拷贝路径     - CPU 仍有加密开销，但少了一次内核-用户态拷贝  3) 应用层加密（端到端）     - Producer 加密 payload，Consumer 解密     - SSL 关闭，sendfile 恢复     - 业务复杂度高，Schema Registry 兼容差  4) 硬件加速（QAT / AES-NI）     - Intel QAT 卡专门做 SSL 卸载，CPU 占用下降 50%+     - 蚂蚁、阿里云已经在用实际选择：  - 金融场景：方案 1 + 方案 2 组合（内网明文 + kTLS 公网）  - 资损零容忍：方案 3（业务层加密）

1) 内存配比   机器 256GB → 堆 8GB → PageCache 248GB   单 Broker 数据热点 < 248GB → 几乎全命中2) Topic 数量控制   单 Broker partition 数 < 4000，避免文件描述符 / 内存映射开销   过多 Topic → segment 文件多 → PageCache 碎片化 → 命中率下降3) 消费节奏   Consumer Lag < 1 分钟（核心 Topic）→ 几乎只读最新数据 → PageCache 全命中   Lag > 1 小时 → 读冷数据 → PageCache miss + 磁盘 IO 突增4) Tiered Storage（KIP-405）   冷数据下沉 S3，本地只保留热数据 → PageCache 命中率提升5) 监控   sar -B 1 看 majflt/s（major page fault 次数）   命中率 = 1 - major_fault / total_read实际数据：   我们集群命中率从 92% 提到 98%，吞吐提升 30%，磁盘 IOPS 下降 60%。

mmap 的作用：  Kafka 把 .index / .timeindex 用 mmap 映射到内存  → 二分查找走指针运算，O(log n) 但在内存里mmap 的代价：  整个文件占进程虚拟地址空间  读时实际占物理内存（PageCache）  Topic 多 → 索引文件多 → mmap 数量爆炸优化方案：  1) 限制每分区 segment 数量     log.segment.bytes 调大（512MB → 1GB）→ segment 少 → 索引少  2) 加快 segment 回收     log.retention.hours 不要设过大 → 老 segment 早删 mmap 释放  3) 控制 Topic 数量     单 Broker partition 总数 < 4000  4) 监控 mmap 数量     ls /proc/<broker_pid>/maps | wc -l     超过 10000 警惕  5) 极端场景换 Pulsar     Pulsar 用 BookKeeper Ledger，索引在 BookKeeper 中     mmap 不会随 Topic 数量爆炸