Kafka 核心使用

命令行能做的事

装好之后，日常 90% 的操作都不需要打开代码：建 Topic、看分区、调参数、删数据，kafka-topics.sh 和 kafka-configs.sh 全包了。这一篇先把命令过一遍，再讲客户端参数。

Topic 生命周期

# 创建：3 分区 2 副本（Broker 数必须 >= 2）
bin/kafka-topics.sh --bootstrap-server localhost:9092 \
  --create --topic order-created \
  --partitions 3 --replication-factor 2

# 查看详情
bin/kafka-topics.sh --bootstrap-server localhost:9092 --describe --topic order-created

# 只看有问题的分区，巡检时用
bin/kafka-topics.sh --bootstrap-server localhost:9092 --describe \
  --under-replicated-partitions

# 扩容分区：3 → 6，只能增不能减
bin/kafka-topics.sh --bootstrap-server localhost:9092 \
  --alter --topic order-created --partitions 6

# 删除
bin/kafka-topics.sh --bootstrap-server localhost:9092 \
  --delete --topic order-created

四个动作里有三个坑：

分区不能减少。Kafka 没有「合并分区」这个操作，减分区等于重建 Topic 再迁数据。
分区一扩，key 的落点就变了。原来 hash(key) % 3，扩到 6 之后变成 hash(key) % 6，同一个 key 可能落到新分区，分区内有序的保证被破坏。有顺序要求的 Topic，扩容前必须停写或者接受乱序。
副本因子不能直接改。--alter 没有 --replication-factor 参数，改副本要走分区重分配（第 5 篇讲）。

改配置不用重建

Topic 级别的配置是动态的，改完立即生效，不用重启也不用重建：

# 把保留时间改成 3 天，同时限制单条消息最大 2MB
bin/kafka-configs.sh --bootstrap-server localhost:9092 \
  --alter --entity-type topics --entity-name order-created \
  --add-config retention.ms=259200000,max.message.bytes=2097152

# 查看当前生效的配置（带来源：DYNAMIC_TOPIC_CONFIG / DEFAULT_CONFIG）
bin/kafka-configs.sh --bootstrap-server localhost:9092 \
  --describe --entity-type topics --entity-name order-created

# 删掉自定义配置，回落到 broker 默认
bin/kafka-configs.sh --bootstrap-server localhost:9092 \
  --alter --entity-type topics --entity-name order-created \
  --delete-config retention.ms

--describe 里的 Configs 列会标出来源，DYNAMIC_TOPIC_CONFIG 是 Topic 级覆盖，STATIC_BROKER_CONFIG 是 server.properties 里写死的，DEFAULT_CONFIG 是代码默认值。排查「为什么保留时间和我配的不一样」时，先看这个来源。

分区数与副本因子怎么定

场景	分区数	副本因子	说明
单机学习	1 到 3	1	没有冗余，挂了就没了
一般业务	6 到 12	3	消费者并行度留余量，能扛一台机器
高吞吐埋点	按目标吞吐推算	3	单分区 10MB/s 保守估，再乘安全系数
强顺序交易	等于 key 分组数	3	顺序性和并行度直接冲突，宁可少分区

分区数一旦定下来，后期调整代价很大。经验值：单 Broker 控制在 2000 个分区以内，单集群控制在 20 万以内，超过之后控制器故障恢复时间会明显变长。

生产者关键参数

生产者要回答两个问题：消息会不会丢、吞吐够不够。参数基本都围绕这两点。

acks：什么时候算写成功

# 最安全：ISR 里的副本都写入才算成功
acks=all
# 折中：Leader 写入即返回，Leader 挂了可能丢最近一批
acks=1
# 完全不管：发出去就返回，几乎不用
acks=0

acks=all 是生产环境的默认选择，配合 Broker 侧 min.insync.replicas=2，才能在挂一台机器时不丢数据。代价是每次写入要多等一次副本同步，延迟会稍高。

retries 与 delivery.timeout.ms：重试的边界

# 默认就是 Integer.MAX_VALUE，基本不用改
retries=2147483647
# 重试间隔，指数退避到上限
retry.backoff.ms=100
retry.backoff.max.ms=1000
# 总超时：从 send 到最终成功或失败的上限，默认 2 分钟
delivery.timeout.ms=120000
# 请求超时
request.timeout.ms=30000

很多人以为 retries 设大点就能一直重试，其实真正的天花板是 delivery.timeout.ms。它会从发送那一刻开始计时，超时后无论还剩多少次重试都直接失败。三个值的关系是 delivery.timeout.ms >= linger.ms + request.timeout.ms，调的时候别拆开单独改。

而且重试本身会引入重复：第一次写成功但响应丢了，客户端重试就写了两条。所以光靠 retries 保证不了「不重不漏」。

幂等与事务

# 3.0 起默认 true，不用手动开
enable.idempotence=true
# 幂等的隐含要求，改了会直接报配置冲突
acks=all
max.in.flight.requests.per.connection=5

幂等生产者给每条消息带上生产者 ID 和序列号，Broker 侧去重，解决的是单分区、单会话内的重复。进程重启后 PID 变了，去重能力就断档了。

要跨分区、跨会话的精确一次，得上事务：

transactional.id=order-tx-01

事务消息必须配消费者侧 isolation.level=read_committed，否则未提交的消息也会被读到。事务的开销比幂等大不少，只在「扣库存 + 发消息」这类必须原子的场景用。

吞吐三件套

# 批大小：16KB 是默认值，吞吐优先可以调到 64KB 到 256KB
batch.size=65536
# 攒批等待时间：默认 0，调大到 5 到 20ms 能显著提升吞吐
linger.ms=10
# 压缩：lz4 综合最优，zstd 压缩率更高但 CPU 更贵
compression.type=lz4
# 发送缓冲区，攒批 + 异步发送都靠它
buffer.memory=67108864

batch.size 和 linger.ms 是「攒批」的两个维度，任一条件满足就发送。两者都调大，吞吐上去、延迟也上去，是个纯权衡。业务允许 10ms 延迟的话，linger.ms=10 基本是无脑收益。

发送失败的兜底

send() 是异步的，返回的 Future 里藏着异常。只调 send() 不看回调，等于把失败静默丢掉——这是「消息莫名少了」的头号原因。要么 .get() 同步等结果，要么写 callback，没有第三种正确姿势。

消费者关键参数

auto.offset.reset：没位点的时候从哪读

这个参数只在消费组没有已提交位点或者位点越界时生效，不是「每次都从头读」：

# 默认：只消费启动之后的新消息
auto.offset.reset=latest
# 从头读：新组首次消费、数据重放时用
auto.offset.reset=earliest
# 不自动重置，直接抛异常，适合要求严格位点的场景
auto.offset.reset=none

有个经典误会：觉得设了 earliest 就能重新消费历史。实际上只要组里还有有效位点，这个参数根本不会被触发，想重放得先 --reset-offsets（第 5 篇有这个命令）。

enable.auto.commit：自动提交的陷阱

# 默认开启，每 5 秒提交一次
enable.auto.commit=true
auto.commit.interval.ms=5000

自动提交方便，但语义是「poll 到就提交」，不是「处理完才提交」。消费者拉了一批消息，业务处理到一半进程挂了，位点已经提交过，这批消息就永久丢了。也就是说自动提交默认给的是 at-most-once。

要 at-least-once（宁可重复不可丢），关掉自动提交，处理完再手动提交：

enable.auto.commit=false

// 手动提交的标准骨架：先处理、后提交
while (running) {
    ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(500));
    for (ConsumerRecord<String, String> record : records) {
        process(record);            // 业务处理，失败就抛异常跳出
    }
    consumer.commitSync();          // 整批处理完再提交，至少一次
}

commitSync() 阻塞直到 Broker 确认，失败会重试；commitAsync() 不阻塞但没有重试，适合追求吞吐且能容忍偶发重复的场景。一个常用组合是平时 commitAsync、关闭时 commitSync，兼顾性能和正确性。

超时与拉取

# 会话超时：超过这个时间没心跳就踢出组，默认 45 秒
session.timeout.ms=45000
# 心跳间隔，一般是会话超时的 1/3
heartbeat.interval.ms=3000
# 两次 poll 的最大间隔，超了也踢出组，默认 5 分钟
max.poll.interval.ms=300000
# 单次 poll 最多拿多少条，批量处理时调小避免超时
max.poll.records=500
# 每分区每次最多拉多少字节
max.partition.fetch.bytes=1048576

session.timeout.ms 和 max.poll.interval.ms 是两类完全不同的超时，前者是「心跳断了」，后者是「业务处理太慢」，第 6 篇排 rebalance 会反复用到。

分区策略

默认：key 哈希，无 key 走粘性

key 不为 null  →  murmur2(key) & 0x7fffffff % 分区数  →  固定分区
key 为 null    →  随机选一个分区，攒满一批再换下一个（粘性分区）

murmur2 是 Kafka 自己的哈希实现（Java 的 hashCode 不能跨语言一致），所以别的客户端算出来的分区和 Java 客户端一致，这一点对多语言混部很重要。

无 key 时的粘性策略是从 2.4 开始的行为：不再一条一条轮询打散，而是选一个分区连续写，攒满 batch.size 再换。这样每批只打一个分区，broker 侧写入更连续，吞吐比老版轮询高不少。

分区数变了，key 的落点就变了

hash(key) % 分区数 里的除数变了，所有 key 的映射都会重排。这意味着：给有 key 的 Topic 扩分区，会破坏既有的顺序性假设。如果下游按 key 做状态计算，扩容那一刻要准备好处理「同一 key 出现在两个分区」的乱序窗口。

自定义分区器

有些业务需要按规则路由，比如大客户单独走一个分区、或者按地域分：

partitioner.class=com.example.VipFirstPartitioner
# 3.3+ 的开关：忽略 key，强制均匀打散（调试和压测常用）
partitioner.ignore.keys=false

自定义实现只需要实现 Partitioner 接口：

public class VipFirstPartitioner implements Partitioner {
    @Override
    public int partition(String topic, Object key, byte[] keyBytes,
                         Object value, byte[] valueBytes, Cluster cluster) {
        int total = cluster.partitionsForTopic(topic).size();
        // Vip 开头的 key 优先落到 0 号分区，其余按默认哈希
        if (key != null && key.toString().startsWith("Vip")) {
            return 0;
        }
        return Math.floorMod(Utils.murmur2(keyBytes), total);
    }

    @Override
    public void close() { }

    @Override
    public void configure(Map<String, ?> configs) { }
}

自定义分区器排障时很坑：客户端代码和 Broker 无关，出问题只能翻业务日志。所以除非确有需求，优先用「key 设计」而不是「自定义分区器」来控路由。

顺序性的正确姿势

想要顺序，不需要自定义分区器，只需要保证同一个业务实体用同一个 key（订单号、用户 ID、设备号）。Kafka 天然保证单分区有序，你要做的是别让同一个实体的消息散到多个分区。

小结

Topic 管理四件事：创建、描述、扩分区、删；分区只能增不能减，扩容会改变 key 的散列落点。
Topic 级配置用 kafka-configs.sh --alter 动态改，改完立即生效，--describe 能看到配置来源。
生产者抓三组参数：可靠性（acks/retries/幂等）、攒批（batch.size/linger.ms/压缩）、超时（delivery.timeout.ms）。
消费者先分清 auto.offset.reset 只管「没位点」的情况；要 at-least-once 就关自动提交、处理完再 commitSync。
分区策略默认就是「key 哈希 + 无 key 粘性」，顺序性靠 key 设计解决，自定义分区器是最后手段。

到这里单机已经能干活了，但一台机器谈不上高可用。下一篇《部署与高可用》把 Broker 扩成 3 节点 KRaft 集群，顺带把分区规划和磁盘保留策略算清楚。

命令行能做的事​

Topic 生命周期​

改配置不用重建​

分区数与副本因子怎么定​

生产者关键参数​

acks：什么时候算写成功​

retries 与 delivery.timeout.ms：重试的边界​

幂等与事务​

吞吐三件套​

消费者关键参数​

auto.offset.reset：没位点的时候从哪读​

enable.auto.commit：自动提交的陷阱​

超时与拉取​

分区策略​

默认：key 哈希，无 key 走粘性​

自定义分区器​

小结​