books

《Java 消息中间件·Kafka 内核与客户端》前言：实验环境、集群与数据口径

本书所有实验都在同一台机器上完成：Temurin 25.0.5+8、colima 里的 Docker 29.5.2、apache/kafka:4.3.0 的 KRaft 集群。这篇前言给出单节点与三节点两套底座的完整启动命令、客户端工程的最小 pom、源码核实方法，以及「单次运行不作基准」和「未在本机执行」两条数据口径的具体用法。

Leeting Yan 2026-10-29 7 分钟阅读 3170 字

为什么先把环境写清楚

消息系统的书最容易出问题的地方，不是概念讲错，是结论和版本对不上。

三年写成的书里，「Kafka 需要 ZooKeeper」「用 --zookeeper 参数」「broker.id 是必填项」这类句子，在 4.x 上全是错的。读者照着敲，命令直接报「不认识的参数」。更隐蔽的是分区分配、再平衡这类行为，4.x 换了新协议，旧的描述看起来还能用，实际已经不对了。

所以这本书从第一页起就把版本钉死，并且每个实验都给出可以原样复制的命令。这一篇负责交代三件事：用什么机器、怎么起集群、数字怎么读。

机器与工具链

项目	值	备注
机器	Apple M1 Pro	10 核（8 性能核 + 2 能效核），32 GB 统一内存
操作系统	macOS 26.3
JDK	Temurin 25.0.5+8 LTS	`java -version` 输出见下
Maven	3.9.12
容器运行时	colima	Virtualization.Framework，aarch64，virtiofs
Docker	29.5.2	虚拟机是 Ubuntu 24.04.4 LTS，kernel 6.8，cgroup v2
容器资源	4 vCPU / 7.7 GiB	`docker info` 里的 `NCPU=4`、`MemTotal=8307167232`
编排	`docker-compose` 5.5.0	独立二进制，本机没有 `docker compose` 插件
Kafka 服务端	`apache/kafka:4.3.0`	镜像 678 MB
Kafka 客户端	`org.apache.kafka:kafka-clients:4.3.0`	fat jar，但 slf4j-api 需外部提供
流处理	`org.apache.kafka:kafka-streams:4.3.0`	仅卷三使用

JDK 的实际输出：

openjdk version "25.0.5" 2026-10-20 LTS
OpenJDK Runtime Environment Temurin-25.0.5+8 (build 25.0.5+8-LTS)
OpenJDK 64-Bit Server VM Temurin-25.0.5+8 (build 25.0.5+8-LTS, mixed mode, sharing)

有一点必须提前说：这台机器不是空闲的实验机，上面同时有别的任务在跑。写作期间实测 load average 在十几到二十几之间波动，2026-10-29 12:56 的一次采样是：

$ sysctl -n vm.loadavg
{ 14.86 18.37 23.02 }

不同时间段的差别可能很大，所有耗时数据都受这个前提影响，这也是后面数据口径的由来。

为什么用容器而不是装机

三个原因。

第一，可复现。把 Kafka 装在宿主机上，版本升级、配置文件位置、用户的 ulimit 都会影响结果。容器镜像把 broker 的 JVM 参数、日志目录、默认配置全部固定住，换台机器跑出来的行为是一致的。

第二，可破坏。本书要反复做「杀掉一个 broker」这类实验。在容器里 docker stop 是干净的，在宿主机上杀进程可能留下未清理的日志和锁文件，下一次启动的行为就变了。

第三，资源可限。三节点集群在笔记本上跑，必须给每个 broker 限内存。KAFKA_HEAP_OPTS=-Xmx512m 让三个 broker 加起来的堆不超过 1.5 GB，虚拟机的 7.7 GiB 才够用。

代价是容器网络比宿主机多一层 NAT。这一点直接影响了 listener 的配置，2.1 节会详细讲。

单节点底座

日常实验、验证一条命令、跑一个客户端，用单节点就够了。完整命令：

docker run -d --name kafka-lab \
  -p 19092:19092 \
  -e KAFKA_NODE_ID=1 \
  -e KAFKA_PROCESS_ROLES=broker,controller \
  -e KAFKA_LISTENERS='PLAINTEXT://0.0.0.0:9092,CONTROLLER://0.0.0.0:9093,EXTERNAL://0.0.0.0:19092' \
  -e KAFKA_ADVERTISED_LISTENERS='PLAINTEXT://127.0.0.1:9092,EXTERNAL://127.0.0.1:19092' \
  -e KAFKA_LISTENER_SECURITY_PROTOCOL_MAP='CONTROLLER:PLAINTEXT,PLAINTEXT:PLAINTEXT,EXTERNAL:PLAINTEXT' \
  -e KAFKA_INTER_BROKER_LISTENER_NAME=PLAINTEXT \
  -e KAFKA_CONTROLLER_LISTENER_NAMES=CONTROLLER \
  -e KAFKA_CONTROLLER_QUORUM_VOTERS='1@127.0.0.1:9093' \
  -e KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR=1 \
  -e KAFKA_TRANSACTION_STATE_LOG_REPLICATION_FACTOR=1 \
  -e KAFKA_TRANSACTION_STATE_LOG_MIN_ISR=1 \
  apache/kafka:4.3.0

启动后约 20–30 秒可用。镜像自带的启动脚本会在末尾打印一个汇总块，健康状态是这样：

   [error] 0 errors (0 unique)
   [warn] 0 warnings (0 unique)
   [info] 17 info messages

如果看到 Error connecting to node ...，先检查两件事：controller quorum voters 是不是用了 127.0.0.1 而不是容器名；EXTERNAL 的 advertised listener 端口有没有和 -p 映射对上。

单节点的能力边界要说清楚：它建不了 replication.factor > 1 的 topic。副本相关的实验必须上三节点，本书不会在单节点上「假装」三副本。

三节点底座

用 compose 文件起三个 broker，共享一个 controller quorum。文件放在 mq-ops/compose/docker-compose.yml，核心片段：

services:
  k1:
    image: apache/kafka:4.3.0
    container_name: k1
    ports: ["19091:19091"]
    environment:
      KAFKA_NODE_ID: "1"
      KAFKA_PROCESS_ROLES: broker,controller
      KAFKA_LISTENERS: PLAINTEXT://0.0.0.0:9092,CONTROLLER://0.0.0.0:9093,EXTERNAL://0.0.0.0:19091
      KAFKA_ADVERTISED_LISTENERS: PLAINTEXT://k1:9092,EXTERNAL://127.0.0.1:19091
      KAFKA_CONTROLLER_QUORUM_VOTERS: 1@k1:9093,2@k2:9093,3@k3:9093
      KAFKA_CONTROLLER_LISTENER_NAMES: CONTROLLER
      KAFKA_INTER_BROKER_LISTENER_NAME: PLAINTEXT
      KAFKA_LISTENER_SECURITY_PROTOCOL_MAP: CONTROLLER:PLAINTEXT,PLAINTEXT:PLAINTEXT,EXTERNAL:PLAINTEXT
      KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR: "3"
      KAFKA_TRANSACTION_STATE_LOG_REPLICATION_FACTOR: "3"
      KAFKA_TRANSACTION_STATE_LOG_MIN_ISR: "2"
      KAFKA_DEFAULT_REPLICATION_FACTOR: "3"
      KAFKA_MIN_INSYNC_REPLICAS: "2"
      KAFKA_HEAP_OPTS: -Xmx512m -Xms512m

k2、k3 复制同样的结构，只改 NODE_ID、容器名、端口和 advertised listener。三者在同一个 compose 网络里，所以 broker 之间可以用容器名 k1/k2/k3 互相解析——这是自定义网络和默认 bridge 的区别，2.1 节会展开。

启动命令：

docker-compose -f docker-compose.yml up -d

等 30–45 秒，用 broker 列表确认：

docker exec k1 /opt/kafka/bin/kafka-broker-api-versions.sh \
  --bootstrap-server 127.0.0.1:9092

返回的节点数应该是 3。三个容器的状态都应该是 Up，且 docker logs 里 ERROR 计数为 0。

宿主机访问走 EXTERNAL listener 的三个端口。上面的模板写的是 19091–19093，但你机器上这三组端口未必空闲——本机实测时就发现 19091 和 19093 已被别的进程占用，Docker 的端口映射声明成功却实际绑定失败，表现为「端口能连上、Kafka 请求超时」。换一组没被占用的端口（本机改用 29091–29093），ports、KAFKA_LISTENERS、KAFKA_ADVERTISED_LISTENERS 三处要同步改。

127.0.0.1:29091 → k1
127.0.0.1:29092 → k2
127.0.0.1:29093 → k3

bootstrap 写任意一个，客户端会通过元数据自动发现另外两个。附录 A 给出了一段十几行的 Python，用来确认某个端口后面真的是 Kafka 而不是别的进程。

客户端工程

所有 Java 实验共用一个最小 Maven 工程。pom.xml 只有三条关键配置：

<properties>
  <maven.compiler.release>25</maven.compiler.release>
  <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
</properties>
<dependencies>
  <dependency>
    <groupId>org.apache.kafka</groupId>
    <artifactId>kafka-clients</artifactId>
    <version>4.3.0</version>
  </dependency>
  <dependency>
    <groupId>org.slf4j</groupId>
    <artifactId>slf4j-api</artifactId>
    <version>2.0.20</version>
  </dependency>
  <dependency>
    <groupId>org.slf4j</groupId>
    <artifactId>slf4j-simple</artifactId>
    <version>2.0.20</version>
  </dependency>
</dependencies>

有一个容易踩的坑，而且网上很多说法是错的：kafka-clients 4.3.0 并不把 slf4j-api 打进去。只加 kafka-clients 这一个依赖，new KafkaProducer(...) 会直接抛：

Exception in thread "main" java.lang.NoClassDefFoundError: org/slf4j/LoggerFactory
	at org.apache.kafka.common.utils.Utils.<clinit>(Utils.java:118)

注意这不是「缺绑定实现」那种软失败——它连类都加载不到，程序起不来。所以 slf4j-api 是必需依赖。

再进一步：只加 slf4j-api 而不加绑定，程序能跑，但一行日志都不打（SLF4J 会退回 NOP 实现）。要看 broker 连接的细节，还要加 slf4j-simple（或 logback）。两个都加上，才是完整的。

本机实测的两个报错形态：

# 只有 kafka-clients
NoClassDefFoundError: org/slf4j/LoggerFactory

# kafka-clients + slf4j-api，无绑定
SLF4J(W): No SLF4J providers were found.
SLF4J(W): Defaulting to no-operation (NOP) logger implementation

一个能验证环境的最小程序：

public class Probe {
    public static void main(String[] args) {
        System.out.println("Runtime=" + System.getProperty("java.version"));
        System.out.println("kafka-clients="
            + org.apache.kafka.common.utils.AppInfoParser.getVersion());

        Properties p = new Properties();
        p.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "127.0.0.1:19092");
        p.put(ProducerConfig.MAX_BLOCK_MS_CONFIG, "1500");
        p.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
        p.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());

        try (KafkaProducer<String, String> producer = new KafkaProducer<>(p)) {
            System.out.println("producer constructed OK, metrics=" + producer.metrics().size());
        }
    }
}

本机实测输出：

Runtime=25.0.5
kafka-clients=4.3.0
producer constructed OK, metrics=77

请特别注意第二行：不连 broker 的时候，new KafkaProducer 也可能因为 max.block.ms 到期而抛 TimeoutException。这说明构造过程会去拉元数据，不是一个纯内存操作。这个细节在第 7.1 节会再出现一次。

数据口径

本书的数字分三类，正文里会明确标注。

第一类：本机实测。有命令、有原始输出、有日期。格式是：

本机实测（Temurin 25.0.5+8 / Kafka 4.3.0 KRaft 三节点 / 2026-10-29）：……

这类数字可以复现，但不能当作性能基准。原因前面说过，机器 load 常年在十几到二十几之间，同一条命令两次运行的耗时可能差一倍。所以正文只用它们做两件事：证明某个行为会发生（leader 切换了、写入被拒了），以及给出相对比例（A 比 B 慢多少倍）。

第二类：按公式推导。比如容量规划里「3 分区 × 100 MB/s × 7 天 = 多少 TB」。这类数字会写明输入假设，读者可以换成自己的数重算。

第三类：未在本机执行。本机只有一台机器，没有真实机架、没有多可用区、没有独立的监控系统。涉及这些的实验会写「未在本机执行」，并给出命令和预期输出形态，不会编一组数字。

第三类是这本书的信誉底线。一本书如果每个实验都声称做过，反而不可信。

源码核实方法

卷一第三部分（第 7–9 章）要读客户端源码。方法固定为四步：

# 1. 定位 sources jar（Maven 下载依赖后就在本地仓库）
ls ~/.m2/repository/org/apache/kafka/kafka-clients/4.3.0/

# 2. 解出目标文件
unzip -p ~/.m2/repository/org/apache/kafka/kafka-clients/4.3.0/kafka-clients-4.3.0-sources.jar \
  org/apache/kafka/clients/producer/internals/Sender.java > /tmp/Sender.java

# 3. 定位方法
grep -n "private void sendProducerData\|runOnce" /tmp/Sender.java

# 4. 读完整方法体，不要只读签名

sources jar 有 3.7 MB，关键类的大小如下，其中两个数字值得注意：

类	大小	说明
`KafkaProducer.java`	91.0 KB
`internals/Sender.java`	55.9 KB
`internals/RecordAccumulator.java`	61.5 KB
`internals/TransactionManager.java`	92.7 KB	事务细节留给卷二
`NetworkClient.java`	69.5 KB
`consumer/internals/AsyncKafkaConsumer.java`	115.4 KB	KIP-848 新实现
`consumer/internals/ConsumerCoordinator.java`	83.7 KB	classic 协议侧
`consumer/internals/AbstractStickyAssignor.java`	81.8 KB
`consumer/CooperativeStickyAssignor.java`	7.2 KB
`consumer/internals/Fetcher.java`	9.1 KB	重构后显著变小

这些数字用 wc -c 取字节数后除以 1024 得到。ls -lh 和 du 报的是块占用，会系统性偏大 2–3%（本机初稿就用 du 估过一遍，得到的 Fetcher.java 是 12K，实际 9.1 KB）。

最后一行是个陷阱。如果你按 3.x 的印象去描述 Fetcher 的职责，很可能会写错——4.x 的消费者重构之后，拉取管线的分工变了，Fetcher 只剩很小一部分。第 8.2 节会以实际源码为准，不按「我记得」来写。

复现顺序

想跟着做实验的读者，按这个顺序准备：

装 JDK 25（或任何 17+ 的 LTS），确认 java -version 是 25。
装 Docker 与 colima（或 Docker Desktop），确认 docker info 输出里有 Server Version: 29.x。
docker pull apache/kafka:4.3.0。
按上面的单节点命令起一个 broker，用 kafka-topics.sh --list 确认能连上。
建一个 Maven 工程，把 pom.xml 和上面的 Probe 跑通。
需要副本实验时，换成三节点 compose。

跑不通的时候，先看容器日志里的错误汇总块，再看 docker logs 里的连接错误。绝大多数问题是端口映射和 advertised listener 不一致，2.1 节有一张排错表。

环境交代完毕。第 1.1 节从最基本的问题开始：消息系统到底有几种形态，Kafka 为什么选了日志这一种。

继续阅读

探索更多技术文章

浏览归档，发现更多关于系统设计、工具链和工程实践的内容。

全部文章返回首页