未记录事实核验 · 状态：维护中

记录仅说明核验时间与维护状态，不保证适用于所有环境；使用前请留意正文中的版本、适用条件与风险提示。

Kafka 最佳实践 - 生产环境经验总结

本文汇总 Topic 设计、消息规范、生产消费配置及运维安全的实践建议，供生产评审与故障分析练习使用。文中未提供可追溯的生产事故材料，因此不将这些建议表述为已证实的事故经验。

---

Topic 设计规范

命名规范

推荐格式：<domain>.<entity>.<event>

# 好的命名示例
order.payment.completed      # 订单域 - 支付实体 - 完成事件
user.profile.updated         # 用户域 - 档案实体 - 更新事件
inventory.stock.decreased    # 库存域 - 库存实体 - 减少事件

分区数黄金法则

分区数是不可减少的，只能增加，且增加后有副作用。

初始分区数估算公式：
  目标吞吐量 ÷ 单分区实测可持续吞吐量 = 吞吐维度的分区数估算值（向上取整）
  练习可假设单分区为 10-20 MB/s；该数值不是已验证上限，实际取值需按消息大小、硬件、压缩与确认配置测试

建议：
  初始值取估算结果的 2 倍（留扩展余量）
  单 Topic 分区数不超过 Broker 数量的 10 倍（避免元数据膨胀）

增加分区的副作用：

• 如果消息 Key 做了业务路由（如同一用户 ID 的消息必须同一分区），增加分区后路由规则改变，历史数据和新数据的分区分布不一致
• Consumer Group 会触发 Rebalance，短暂停止消费
• 分区数增加不会重新分布已有数据，只影响新写入的消息

复制因子与 min.insync.replicas

部署规模	推荐副本数	min.insync.replicas	说明
开发环境（1 Broker）	1	1	功能验证用
测试环境（3 Broker）	2	1	容忍 1 个 Broker 故障
生产环境（3+ Broker）	3	2	配合 acks=all，在 ISR 和副本分布满足条件时可容忍 1 个 Broker 故障；不构成无条件不丢数据保证
关键业务（5+ Broker）	3	3	配合 acks=all 要求三个 ISR 副本参与确认；ISR 少于三个时拒绝写入，需评估可用性代价

---

消息设计

Key 的选择

Key 决定了消息路由到哪个分区，设计不当会造成热点分区：

Key 设计	分布特征	适用场景
null（无 Key）	分布取决于客户端版本和分区器；默认策略可能采用粘性或自适应分区，不保证严格轮询或完全均匀	不要求按业务 Key 保序、侧重吞吐量
用户 ID（高基数）	分布均匀	需要保证同一用户的消息有序
地区（低基数，如 10 个省）	容易不均匀	不推荐，除非配合自定义分区器
订单 ID（UUID）	随机均匀	需要订单级别的消息顺序

热点分区的排查方式：

# 查看各分区副本的日志字节大小；它不是消息条数，差异需结合消息大小、保留策略及流量判断
kafka-log-dirs.sh --bootstrap-server kafka:9092 \
  --topic-list order-events \
  --describe | grep "size"

Value 格式选择

格式	优点	缺点	推荐场景
Avro + Schema Registry	紧凑、类型安全、Schema 演进	依赖 Registry 服务	生产环境首选
JSON	可读性好、无依赖	冗余字段多、无 Schema 约束	内部工具、低吞吐量场景
Protobuf	极致紧凑、跨语言	需要 .proto 文件管理	超高吞吐量、多语言混合
自定义二进制	最紧凑	维护成本极高	几乎不推荐

消息大小建议：

• 正常消息：< 100 KB
• 示例控制目标：< 1 MB；这不是不可调整的硬性上限，实际限制需协调 Broker/Topic 的消息大小限制及生产者、消费者相关配置，并评估批处理效率和资源开销
• 超大消息处理方案：消息里只存引用（S3/HDFS 路径），实际内容存对象存储

---

生产者最佳实践

核心配置原则

# 幂等性：Kafka Java 生产者自 3.0 起默认开启；此处显式启用，并需满足相关配置约束
enable.idempotence=true

# 必须与幂等性配合，等待所有 ISR 副本确认
acks=all

# min.insync.replicas 在 Broker/Topic 层面设置，不在生产者设置
# 生产者这边只负责 acks

# 重试配置（幂等性开启后重试是安全的）
retries=2147483647          # 最大值，交由 delivery.timeout.ms 控制总时长
delivery.timeout.ms=120000  # 2 分钟内必须成功，否则报错
retry.backoff.ms=100        # 初始重试间隔

# 批处理：平衡延迟与吞吐量
batch.size=65536             # 64 KB，增大可提升吞吐量
linger.ms=5                  # 等待 5ms 以积累批次，低延迟场景设 0
compression.type=lz4         # lz4 压缩比与 CPU 开销最均衡

# 缓冲区：发送请求积压时的本地缓冲
buffer.memory=67108864       # 64 MB
max.block.ms=60000           # send() 阻塞最长 60 秒（缓冲区满时）

避免消息丢失的检查清单

• [ ] acks=all + enable.idempotence=true
• [ ] Topic 的 min.insync.replicas >= 2
• [ ] 不使用 producer.send() 的无回调版本（fire-and-forget）
• [ ] 捕获 send() 回调中的异常，接入告警
• [ ] 生产者关闭时调用 producer.close() 而非强制终止进程（确保缓冲区消息发送完毕）

---

消费者最佳实践

Offset 提交策略

需要按业务处理结果控制提交位点时，应关闭自动提交并显式管理已处理位点；提交策略须与处理、重试和失败隔离方式配套。

// 提交顺序示例：假设已关闭自动提交；此片段未实现失败重试或隔离，不是完整可靠消费循环
ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(500));
try {
    for (ConsumerRecord<String, String> record : records) {
        process(record);  // 业务处理
    }
    consumer.commitSync();  // 处理完整批次后提交
} catch (Exception e) {
    // 处理失败后不提交并不会使下次 poll 自动重读；需显式重试或 seek 到失败位置，并防止后续提交越过失败记录
    log.error("处理失败，跳过提交", e);
}

// 错误：先提交再处理
consumer.commitSync();
process(records);  // 处理失败后 Offset 已提交，消息丢失！

处理 Rebalance

Rebalance 对消费的影响取决于分配协议和处理方式；正确管理已处理位点可减少重复消费，但不能消除处理成功而提交失败等窗口，业务处理仍需支持幂等或去重：

consumer.subscribe(topicList, new ConsumerRebalanceListener() {
    @Override
    public void onPartitionsRevoked(Collection<TopicPartition> partitions) {
        // Rebalance 前：提交当前已处理的 Offset
        // 这样新的消费者接管分区后，从正确位置开始
        consumer.commitSync(currentOffsets);
        log.info("Rebalance 触发，已提交 Offset: {}", currentOffsets);
    }

    @Override
    public void onPartitionsAssigned(Collection<TopicPartition> partitions) {
        // 接管新分区后的初始化逻辑（如预热缓存）
        log.info("分配到新分区: {}", partitions);
    }
});

Consumer Lag 监控告警

Consumer Lag（消费者积压）是最重要的消费者健康指标：

# 查看 Consumer Group 的所有分区积压情况
kafka-consumer-groups.sh --bootstrap-server kafka:9092 \
  --describe --group order-processor

# 输出字段含义：
# TOPIC         分区所属 Topic
# PARTITION     分区 ID
# CURRENT-OFFSET 消费者已提交的 Offset
# LOG-END-OFFSET  分区最新消息的 Offset
# LAG            积压量 = LOG-END-OFFSET - CURRENT-OFFSET
# CONSUMER-ID    消费该分区的消费者实例 ID

告警阈值建议：

• 关键业务 Topic：Lag > 1 万条，立即告警
• 普通 Topic：Lag 持续增长超过 30 分钟，告警
• Lag 突然变为 0：需检查是否已追平、发生位点重置或监控缺失被显示为零；消费者退出本身不会使 Lag 自动归零，应结合成员状态和指标有效性判断

---

运维最佳实践

Broker 层面的生产配置

# 禁止自动创建 Topic（生产环境必须）
auto.create.topics.enable=false

# 禁止删除 Topic（防止误操作）
delete.topic.enable=false

# 不允许不干净的 Leader 选举（防止数据丢失）
unclean.leader.election.enable=false

# 日志保留按业务需求设置，不要无限保留
log.retention.hours=168          # 7 天，大多数业务够用
log.retention.bytes=107374182400 # 每分区约 100 GiB 的保留限制，按日志段清理；不是集群或磁盘总上限

# Leader 不均衡时自动重平衡（减少热点 Broker）
auto.leader.rebalance.enable=true
leader.imbalance.check.interval.seconds=300

定期运维任务

任务	频率	命令
检查副本不足的分区	每天	kafka-topics.sh --describe --under-replicated-partitions
清理僵尸 Consumer Group	每周	kafka-consumer-groups.sh --delete --group <空组>
检查磁盘使用率	每天告警	kafka-log-dirs.sh --describe
验证 Topic 配置符合规范	每月	自定义审计脚本
更新 JVM 版本	每季度	滚动重启

---

安全最佳实践

生产环境必开的安全配置

# Broker 端：开启 SSL 加密传输
listeners=SASL_SSL://0.0.0.0:9092
ssl.keystore.location=/etc/kafka/ssl/kafka.keystore.jks
ssl.keystore.password=<keystore-password>
ssl.key.password=<key-password>
ssl.truststore.location=/etc/kafka/ssl/kafka.truststore.jks
ssl.truststore.password=<truststore-password>

# SASL 认证（推荐 SCRAM-SHA-256，比 PLAIN 更安全）
sasl.enabled.mechanisms=SCRAM-SHA-256
sasl.mechanism.inter.broker.protocol=SCRAM-SHA-256

# 开启 ACL 授权
# ZooKeeper 模式示例；KRaft 模式应按版本使用 org.apache.kafka.metadata.authorizer.StandardAuthorizer
 authorizer.class.name=kafka.security.authorizer.AclAuthorizer
allow.everyone.if.no.acl.found=false   # 默认拒绝无 ACL 的操作

ACL 最小权限原则

# 为生产者赋予最小权限（只允许写入特定 Topic）
kafka-acls.sh --bootstrap-server kafka:9092 \
  --add --allow-principal User:order-producer \
  --operation Write --operation Describe \
  --topic order-events

# 为消费者赋予最小权限（只允许读取特定 Topic 和 Group）
kafka-acls.sh --bootstrap-server kafka:9092 \
  --add --allow-principal User:order-consumer \
  --operation Read --operation Describe \
  --topic order-events \
  --group order-processor

# 禁止普通用户操作集群管理类 API
# 管理员账号单独管理，不与业务账号混用

---

避坑清单（10 个高频错误）

错误	后果	正确做法
生产者用 fire-and-forget（send 不处理回调）	消息丢失但无感知	始终添加回调，记录失败告警
enable.auto.commit=true	消息处理失败仍提交 Offset，导致丢失	手动 commitSync
acks=1 + 重要业务	Broker Leader 挂时丢消息	acks=all + min.insync.replicas=2
开启 auto.create.topics.enable	拼写错误的 Topic 名被自动创建，难以发现	生产环境关闭，统一走变更流程
分区数一次性设太大	元数据膨胀，Broker 内存压力大	先少后多，按需扩容
不设 max.poll.interval.ms	Consumer 处理慢时被 Group 踢出，触发 Rebalance 风暴	根据业务处理时长合理设置
Topic 保留时间设 infinite（-1）	磁盘被打满，Broker OOM	按业务周期设置，最长 30 天
Consumer Group 名随机生成	无法追踪 Lag，调试困难	每个应用使用固定、有意义的 Group ID
同时使用归一化后同名的点号和下划线 Topic 名	可能产生 JMX 指标名称冲突	可统一用 -，并检查全局命名冲突；使用 . 本身不必然造成冲突
跨机房不同 Kafka 版本混用	API 不兼容，MirrorMaker 出现奇怪问题	跨集群版本保持一致，按季度统一升级

---

掌握检查

• [ ] 能写出符合规范的 Topic 命名，并说出为什么避免用 . 作分隔符
• [ ] 理解分区数不可减少的原因，以及增加分区对有序消息的影响
• [ ] 能说出 acks=all 和 min.insync.replicas=2 必须配合的原因
• [ ] 能用代码实现手动 Offset 提交，并在 Rebalance 时正确处理
• [ ] 理解 Consumer Lag 的含义，知道如何查询和设置告警阈值
• [ ] 能说出 auto.create.topics.enable=false 的生产必要性

练习题

1. 一个电商系统的 order.payment.completed Topic 当前有 6 个分区，生产者使用 user_id 作为消息 Key。业务 TPS 增长需要扩容到 18 个分区，请评估这个扩容操作有什么风险，以及如何缓解。
2. 某消费者组 payment-processor 的 Lag 在最近 1 小时从 0 涨到了 50 万。请列出排查思路，以及分别对应什么根本原因。
3. 你负责一个 Kafka 集群的安全加固工作，现有状态是：auto.create.topics.enable=true、allow.everyone.if.no.acl.found=true、无 SSL。请给出加固方案，并说明每一步的操作顺序（注意不能影响线上业务）。

学习连接

前置知识: 故障处理与恢复 - 灾难恢复策略 下一步: 企业级项目实战 - 综合应用案例 相关概念: Schema Registry、多集群架构

---