只往文件末尾追加，缓存和搬运都交给内核

# 段大小设成允许的最小值 1 MiB，4 MB 数据就能切出好几段
bin/kafka-topics.sh --bootstrap-server localhost:9092 --create \
  --topic order-demo --partitions 1 --config segment.bytes=1048576

# 每条 200 字节；batch.size=1024 让每批只装 4 条，方便看稀疏索引
bin/kafka-producer-perf-test.sh --topic order-demo \
  --num-records 20000 --record-size 200 --throughput -1 \
  --producer-props bootstrap.servers=localhost:9092 batch.size=1024 linger.ms=0

Created topic order-demo.
20000 records sent, …

# logDir 就是分区目录的上一级；解压版默认是 /tmp/kraft-combined-logs
bin/kafka-log-dirs.sh --bootstrap-server localhost:9092 --describe --topic-list order-demo
ls -l <logDir>/order-demo-0/

{"partition":"order-demo-0","size":4485000, … ,"logDir":"…/data"}

    1864  00000000000000000000.index
 1047696  00000000000000000000.log
     708  00000000000000000000.timeindex
    1864  00000000000000004672.index
 1047696  00000000000000004672.log
      56  00000000000000004672.snapshot
     456  00000000000000004672.timeindex
          …（9344、14016 两段同上）
10485760  00000000000000018688.index      ← 活跃段：索引预分配 10 MiB
  294216  00000000000000018688.log
      56  00000000000000018688.snapshot
10485756  00000000000000018688.timeindex
       8  leader-epoch-checkpoint
      43  partition.metadata

cd <logDir>/order-demo-0
<kafka>/bin/kafka-dump-log.sh --files 00000000000000004672.index | head -4
<kafka>/bin/kafka-dump-log.sh --files 00000000000000004672.log  | head -16

offset: 4695 position: 4485
offset: 4715 position: 8970        ← 不超过 4730 的最大一条
offset: 4735 position: 13455

baseOffset: 4672 lastOffset: 4675 count: 4 position: 0     size: 897
baseOffset: 4676 lastOffset: 4679 count: 4 position: 897   size: 897
…
baseOffset: 4692 lastOffset: 4695 count: 4 position: 4485  size: 897   ← 第 1 条索引指着这批
…
baseOffset: 4712 lastOffset: 4715 count: 4 position: 8970  size: 897   ← 从这开始扫
baseOffset: 4716 lastOffset: 4719 count: 4 position: 9867  size: 897
baseOffset: 4720 lastOffset: 4723 count: 4 position: 10764 size: 897
baseOffset: 4724 lastOffset: 4727 count: 4 position: 11661 size: 897
baseOffset: 4728 lastOffset: 4731 count: 4 position: 12558 size: 897   ← 4730 在这一批

# 段文件有多少在页缓存里（RES）
fincore <logDir>/order-demo-0/*.log

# 盯着 Broker 的 sendfile；另开一个终端从头消费
sudo strace -f -y -e trace=sendfile -p <Broker 进程号>
bin/kafka-console-consumer.sh --bootstrap-server localhost:9092 \
  --topic order-demo --from-beginning --max-messages 20000 > /dev/null

  RES PAGES    SIZE FILE
   1M   256 1023.1K order-demo-0/00000000000000000000.log
   1M   256 1023.1K order-demo-0/00000000000000004672.log
   …
 288K    72  287.3K order-demo-0/00000000000000018688.log

sendfile(243<socket:[10561219]>, 247<…/order-demo-0/00000000000000004672.log>, [12558] => [264460], 1035138) = 251902
sendfile(243<socket:[10561219]>, 247<…/order-demo-0/00000000000000004672.log>, [264460] => [497755], 783236) = 233295
…

import java.io.*;
import java.lang.management.ManagementFactory;
import java.net.*;
import java.nio.channels.*;
import java.nio.file.*;

public class ZeroCopyDemo {
    public static void main(String[] args) throws Exception {
        Path file = Path.of("big.bin");
        if (Files.notExists(file)) {                          // 准备一个 1 GiB 的文件
            try (OutputStream out = Files.newOutputStream(file)) {
                byte[] mb = new byte[1 << 20];
                for (int i = 0; i < 1024; i++) out.write(mb);
            }
        }
        for (int round = 1; round <= 3; round++) {
            send("read+write", file, false);
            send("transferTo", file, true);
        }
    }

    static void send(String name, Path file, boolean zeroCopy) throws Exception {
        try (ServerSocket server = new ServerSocket(0)) {
            Thread sink = new Thread(() -> {                  // 接收端：读到就扔
                try (Socket s = server.accept(); InputStream in = s.getInputStream()) {
                    byte[] b = new byte[64 * 1024];
                    while (in.read(b) != -1) { }
                } catch (IOException e) { throw new UncheckedIOException(e); }
            });
            sink.start();
            var mx = ManagementFactory.getThreadMXBean();
            long cpu0 = mx.getCurrentThreadCpuTime(), t0 = System.nanoTime();
            if (zeroCopy) {                                   // 零拷贝：FileChannel.transferTo → sendfile
                try (FileChannel ch = FileChannel.open(file);
                     SocketChannel sock = SocketChannel.open(new InetSocketAddress("localhost", server.getLocalPort()))) {
                    long size = ch.size();
                    for (long pos = 0; pos < size; ) pos += ch.transferTo(pos, size - pos, sock);
                }
            } else {                                          // 老办法：读进 byte[]，再写给 socket
                try (InputStream in = Files.newInputStream(file);
                     Socket sock = new Socket("localhost", server.getLocalPort());
                     OutputStream out = sock.getOutputStream()) {
                    byte[] buf = new byte[64 * 1024];
                    for (int n; (n = in.read(buf)) != -1; ) out.write(buf, 0, n);
                }
            }
            sink.join();
            System.out.printf("%s  发送线程 CPU %4d ms  总耗时 %4d ms%n", name,
                    (mx.getCurrentThreadCpuTime() - cpu0) / 1_000_000, (System.nanoTime() - t0) / 1_000_000);
        }
    }
}

read+write  发送线程 CPU  552 ms  总耗时 1235 ms   ← 第一轮有 JIT 预热
transferTo  发送线程 CPU   57 ms  总耗时  484 ms
read+write  发送线程 CPU  415 ms  总耗时  717 ms
transferTo  发送线程 CPU   57 ms  总耗时  299 ms
read+write  发送线程 CPU  421 ms  总耗时  590 ms
transferTo  发送线程 CPU   60 ms  总耗时  310 ms