这份模板怎么用
四个部分：集群、工程、客户端代码、运维脚本。每一段都可以独立复制。
版本基线固定为 Kafka 4.3.0 + JDK 25。换成 3.x 会有一批参数不存在（broker.id、--zookeeper、ZooKeeper 部署），不要混用。
单节点：一条命令
日常验证、跑单个客户端、看一个行为，用这个就够：
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 \
-e KAFKA_HEAP_OPTS='-Xmx512m -Xms512m' \
apache/kafka:4.3.0
三个要点：
CONTROLLER_QUORUM_VOTERS用127.0.0.1而不是容器名。默认 bridge 网络不解析容器名，写kafka-lab:9093会启动失败，日志里是 controller 连接超时。ADVERTISED_LISTENERS必须是宿主机能访问的地址。写127.0.0.1:19092配合-p 19092:19092才能让宿主机客户端连上。- 单节点建不了
replication.factor > 1的 topic，会直接报InvalidReplicationFactorException。
启动约 20–30 秒后可用。健康标志是日志末尾的汇总块：
[error] 0 errors (0 unique)
[warn] 0 warnings (0 unique)
三节点：compose 文件
每个节点必须显式写全 KAFKA_LISTENERS。这一条是本书踩过的坑：如果用 YAML 锚点继承再只覆盖 ADVERTISED_LISTENERS，容器内监听的端口和宿主机映射的端口会对不上，宿主机客户端报 topic not present in metadata，而 topic 本身完全正常。
services:
k1:
image: apache/kafka:4.3.0
container_name: k1
ports: ["29091:29091"]
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:29091
KAFKA_ADVERTISED_LISTENERS: PLAINTEXT://k1:9092,EXTERNAL://127.0.0.1:29091
KAFKA_LISTENER_SECURITY_PROTOCOL_MAP: CONTROLLER:PLAINTEXT,PLAINTEXT:PLAINTEXT,EXTERNAL:PLAINTEXT
KAFKA_INTER_BROKER_LISTENER_NAME: PLAINTEXT
KAFKA_CONTROLLER_LISTENER_NAMES: CONTROLLER
KAFKA_CONTROLLER_QUORUM_VOTERS: 1@k1:9093,2@k2:9093,3@k3:9093
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_GROUP_INITIAL_REBALANCE_DELAY_MS: "0"
KAFKA_HEAP_OPTS: -Xmx512m -Xms512m
k2:
image: apache/kafka:4.3.0
container_name: k2
ports: ["29092:29092"]
environment:
KAFKA_NODE_ID: "2"
KAFKA_PROCESS_ROLES: broker,controller
KAFKA_LISTENERS: PLAINTEXT://0.0.0.0:9092,CONTROLLER://0.0.0.0:9093,EXTERNAL://0.0.0.0:29092
KAFKA_ADVERTISED_LISTENERS: PLAINTEXT://k2:9092,EXTERNAL://127.0.0.1:29092
KAFKA_LISTENER_SECURITY_PROTOCOL_MAP: CONTROLLER:PLAINTEXT,PLAINTEXT:PLAINTEXT,EXTERNAL:PLAINTEXT
KAFKA_INTER_BROKER_LISTENER_NAME: PLAINTEXT
KAFKA_CONTROLLER_LISTENER_NAMES: CONTROLLER
KAFKA_CONTROLLER_QUORUM_VOTERS: 1@k1:9093,2@k2:9093,3@k3:9093
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_GROUP_INITIAL_REBALANCE_DELAY_MS: "0"
KAFKA_HEAP_OPTS: -Xmx512m -Xms512m
k3:
image: apache/kafka:4.3.0
container_name: k3
ports: ["29093:29093"]
environment:
KAFKA_NODE_ID: "3"
KAFKA_PROCESS_ROLES: broker,controller
KAFKA_LISTENERS: PLAINTEXT://0.0.0.0:9092,CONTROLLER://0.0.0.0:9093,EXTERNAL://0.0.0.0:29093
KAFKA_ADVERTISED_LISTENERS: PLAINTEXT://k3:9092,EXTERNAL://127.0.0.1:29093
KAFKA_LISTENER_SECURITY_PROTOCOL_MAP: CONTROLLER:PLAINTEXT,PLAINTEXT:PLAINTEXT,EXTERNAL:PLAINTEXT
KAFKA_INTER_BROKER_LISTENER_NAME: PLAINTEXT
KAFKA_CONTROLLER_LISTENER_NAMES: CONTROLLER
KAFKA_CONTROLLER_QUORUM_VOTERS: 1@k1:9093,2@k2:9093,3@k3:9093
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_GROUP_INITIAL_REBALANCE_DELAY_MS: "0"
KAFKA_HEAP_OPTS: -Xmx512m -Xms512m
启动与检查：
docker-compose -f docker-compose.yml up -d
sleep 45
docker-compose -f docker-compose.yml ps
docker exec k1 /opt/kafka/bin/kafka-broker-api-versions.sh \
--bootstrap-server k1:9092,k2:9092,k3:9092
本机只有独立的 docker-compose（5.5.0），没有 docker compose 插件。如果你的环境装了插件，把命令里的连字符去掉。
端口为什么要写 2909x：本机 19091 和 19093 被别的进程占着，Kafka 的端口映射虽然声明成功但实际绑定失败。换一组没被占用的端口最省事。选之前先查：
lsof -nP -iTCP:29091 -sTCP:LISTEN
Maven 工程
<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 时抛：
NoClassDefFoundError: org/slf4j/LoggerFactory
不加 slf4j-simple 则能跑但不打印任何日志，连 broker 连不上都看不见。
生产者最小模板
import org.apache.kafka.clients.producer.*;
import java.util.Properties;
import java.util.concurrent.TimeUnit;
public class MinimalProducer {
public static void main(String[] args) throws Exception {
Properties p = new Properties();
p.put("bootstrap.servers", "127.0.0.1:29091,127.0.0.1:29092,127.0.0.1:29093");
p.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
p.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");
p.put("acks", "all");
p.put("enable.idempotence", "true");
p.put("linger.ms", "5");
p.put("delivery.timeout.ms", "30000");
try (KafkaProducer<String, String> producer = new KafkaProducer<>(p)) {
for (int i = 0; i < 10; i++) {
RecordMetadata md = producer
.send(new ProducerRecord<>("demo", "k" + i, "v" + i))
.get(10, TimeUnit.SECONDS);
System.out.printf("partition=%d offset=%d%n", md.partition(), md.offset());
}
}
}
}
send().get() 会阻塞，适合实验和验证，不适合生产。生产环境用回调：
producer.send(new ProducerRecord<>("demo", key, value), (metadata, ex) -> {
if (ex != null) {
System.err.println("send failed: " + ex.getMessage());
}
});
注意回调里不要做耗时操作，它在 Sender 线程上执行，会阻塞整个生产者的发送。
消费者最小模板
import org.apache.kafka.clients.consumer.*;
import org.apache.kafka.common.serialization.StringDeserializer;
import java.time.Duration;
import java.util.List;
import java.util.Properties;
public class MinimalConsumer {
public static void main(String[] args) {
Properties p = new Properties();
p.put("bootstrap.servers", "127.0.0.1:29091,127.0.0.1:29092,127.0.0.1:29093");
p.put("group.id", "demo-group");
p.put("key.deserializer", StringDeserializer.class.getName());
p.put("value.deserializer", StringDeserializer.class.getName());
p.put("enable.auto.commit", "false");
p.put("auto.offset.reset", "earliest");
p.put("max.poll.records", "100");
try (KafkaConsumer<String, String> consumer = new KafkaConsumer<>(p)) {
consumer.subscribe(List.of("demo"));
while (true) {
ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(500));
for (ConsumerRecord<String, String> r : records) {
System.out.printf("%s-%d@%d key=%s val=%s%n",
r.topic(), r.partition(), r.offset(), r.key(), r.value());
}
if (!records.isEmpty()) {
consumer.commitSync();
}
}
}
}
}
两个必须自己处理的地方：
enable.auto.commit=false，在处理完成后手动提交。自动提交在poll返回时就可能提交位点，处理失败的消息会丢（8.3 节）。max.poll.interval.ms（默认 5 分钟）必须大于单批处理时间。超过就会被踢出组，触发再平衡，且这批消息会被重复消费。
优雅关闭要配合关闭钩子：
Runtime.getRuntime().addShutdownHook(new Thread(consumer::wakeup));
用 wakeup() 打断 poll，让主循环自然退出，而不是直接杀进程。
指标打点
打印一组关键指标：
producer.metrics().forEach((name, metric) -> {
String n = name.name();
if (n.startsWith("buffer-") || n.startsWith("record-")
|| n.startsWith("batch-size") || n.startsWith("request-latency")
|| n.equals("record-error-total") || n.equals("record-send-total")) {
System.out.printf("%s{client=%s} = %s%n",
n, name.tags().get("client-id"), metric.metricValue());
}
});
metrics() 返回约 77 项。真正需要盯的几项：
| 指标 | 正常 | 反常的含义 |
|---|---|---|
buffer-available-bytes | 接近 buffer.memory | 持续接近 0 = 生产快于发送，send() 会阻塞 |
bufferpool-wait-time-total | 平稳 | 持续增长 = 分配内存被阻塞，是背压的确切信号 |
record-queue-time-avg | 小 | 变大 = 消息在缓冲里等太久，批次/网络有问题 |
record-error-total | 0 | 大于 0 = 有发送失败，必须看日志 |
request-latency-avg | 稳定 | 抖动 = broker 侧压力或网络问题 |
消费端的 lag 不要只看客户端指标：客户端只在 poll 时更新，进程停了指标也就不动了。积压以 kafka-consumer-groups.sh --describe 的 LAG 列为准。
运维脚本
按 follower → leader 的顺序滚动重启，每次等 ISR 恢复：
#!/bin/sh
set -e
K=/opt/kafka/bin
B=k1:9092,k2:9092,k3:9092
wait_isr() {
for i in $(seq 1 30); do
isr=$(docker exec k1 $K/kafka-topics.sh --bootstrap-server $B \
--describe --topic demo 2>/dev/null | grep -o 'Isr: [0-9,]*')
[ "$isr" = "Isr: 1,2,3" ] && return 0
sleep 2
done
echo "ISR 未恢复，停止" >&2
return 1
}
for n in 2 3 1; do
docker restart k$n
wait_isr
done
顺序不是随意的：先重启 follower，最后重启 leader，全程只发生一次 leader 选举。如果先重启 leader，每重启一个都可能触发一次选举和一轮 ISR 重建（4.3 节）。
三条底线
复制这些模板时，有三件事不要省：
- 每个节点显式写全
KAFKA_LISTENERS，不要用锚点继承再部分覆盖。 enable.auto.commit=false，除非你能接受丢消息。- 滚动重启前先确认
min.insync.replicas和当前 ISR 的大小，ISR 已经等于min.insync.replicas时再停一个，写入会立刻被拒。
集群起不来时，按这个顺序查：容器日志的 [error] 汇总块 → kafka-broker-api-versions.sh 是否三个节点都通 → 端口是否被别的进程占用 → advertised 地址和映射端口是否一致。四步里通常第一步就能定位。
继续阅读
探索更多技术文章
浏览归档，发现更多关于系统设计、工具链和工程实践的内容。