折腾过 Kafka 的人都知道平时排查问题、验证配置、做快速测试最快的路径永远是敲命令行而不是先打开可视化控制台。这篇是 Kafka 系列的第三篇专门聊命令操作。我不打算逐个把 help 抄一遍而是把生产环境里真正高频、真正踩过坑的命令挑出来从参数含义到执行逻辑都讲明白看完你就能直接在集群上动手。如果你正在学 Kafka或者工作中需要维护 Kafka 集群这篇内容适合你。我会围绕 Topic 管理、生产消费链路、消费组与 Offset、配置动态调整、分区重分配、性能压测这几个方向展开最后单独整理一份问题排查清单。实操为主原理穿插着讲这样你用的时候才知道为什么这么敲。1. 环境准备与Kafka命令行全家桶1.1 Kafka命令行工具的存在形式Kafka 的二进制包解压之后bin 目录下面就是一排脚本。以前最常用的有 kafka-topics.sh、kafka-console-producer.sh、kafka-console-consumer.sh、kafka-consumer-groups.sh、kafka-configs.sh、kafka-reassign-partitions.sh后面又陆续增加了 kafka-producer-perf-test.sh、kafka-consumer-perf-test.sh、kafka-dump-log.sh 等辅助工具。这些脚本本身是 shell 包装核心逻辑都在 libs 目录下的 JAR 包里通过 kafka-run-class.sh 启动。所以你在看脚本内容的时候会发现它们基本都是同一套模板只是传入的主类不同。实际操作时我建议先把 Kafka 的 bin 目录加到 PATH 里或者像我一样在 ~/.bashrc 里写几个 alias省得每次敲全路径。比如export KAFKA_HOME/opt/kafka export PATH$KAFKA_HOME/bin:$PATH然后顺手做几个快捷命令alias ktkafka-topics.sh alias kgkafka-consumer-groups.sh alias kckafka-configs.sh alias kpkafka-console-producer.sh alias kcckafka-console-consumer.sh这样做主要是减少重复劳动排查问题的时候本来就急命令少敲几个字体验完全不一样。1.2 ZooKeeper与KRaft模式下命令差异Kafka 从 2.8 开始引入 KRaft 模式也就是去 ZooKeeper 的架构3.x 以后逐步成熟4.0 起 ZooKeeper 被彻底移除。这个变化直接影响命令行的连接参数。在 ZooKeeper 模式下很多管理命令需要指定 --zookeeper 参数比如老版本的 kafka-topics.sh 创建 Topic 时可以写成kafka-topics.sh --zookeeper localhost:2181 --create --topic test-topic --partitions 3 --replication-factor 1但在新版 Kafka 里官方已经不建议用 ZooKeeper 地址操作 Topic统一改成 --bootstrap-server从 2.2 开始命令行管理操作逐步迁移2.7 以后 kafka-topics.sh 对 --zookeeper 的支持进入弃用状态。我自己在 3.x 集群上已经完全不碰 ZooKeeper 参数了全部用 --bootstrap-server 指定 Broker 地址。这样不管底层是 ZooKeeper 还是 KRaft命令风格都是一致的。唯一要注意的是KRaft 模式下有些老 ZooKeeper 专属命令比如查看 /brokers/ids 的 zkCli 命令不再适用如果有这类查询需求要换成 Kafka 自带的管理命令或者 API。2. Topic生命周期与核心命令深度拆解2.1 创建Topic分区与副本不是拍脑袋Topic 是最核心的逻辑容器消息按 Topic 归类每个 Topic 分成多个分区分区是并行读写的基本单位。创建 Topic 的命令大家都熟悉kafka-topics.sh --bootstrap-server broker1:9092 --create \ --topic order-events \ --partitions 12 \ --replication-factor 3 \ --config cleanup.policydelete \ --config retention.ms604800000分区数怎么定很多人直接填 12 或者 24其实这是有讲究的。经验公式是目标吞吐量除以单分区吞吐量再结合消费者并发数向上取整。比如你的业务峰值每秒需要写入 24 MB实测单分区写入吞吐大约 2 MB/s那么理论上 12 个分区就够。但还要看消费者线程数如果消费者组里有 20 个线程分区少于 20 就会出现部分消费者空转所以分区数一般取 max(吞吐需求分区数, 消费者线程数)。副本因子更值得重视。3 副本可以容忍一台 Broker 宕机而不丢数据2 副本在某些异常场景下反而容易出现 ISR 收缩后只剩一个副本的情况。生产环境我通常建议至少 3 副本磁盘成本翻三倍但换来的可用性值这个价。创建 Topic 时 --config 可以带一些预置参数比如上面的 cleanup.policy、retention.ms。更常见的做法是先创建后面用 kafka-configs.sh 动态调整这样灵活一些。因为创建之后再改配置不涉及数据迁移风险小很多。2.2 查看与校验看懂--describe的输出先查一下 Topic 列表kafka-topics.sh --bootstrap-server broker1:9092 --list这个输出很简单就一行一个 Topic 名。如果是新接触一套集群我建议先 --list 看看规模再挑重点 Topic 做 --describe。kafka-topics.sh --bootstrap-server broker1:9092 --describe --topic order-events输出大致长这样Topic: order-events TopicId: XXXXXXXXXXXX PartitionCount: 12 ReplicationFactor: 3 Configs: cleanup.policydelete,retention.ms604800000 Topic: order-events Partition: 0 Leader: 2 Replicas: 2,1,3 Isr: 2,1,3 Topic: order-events Partition: 1 Leader: 1 Replicas: 1,3,2 Isr: 1,3,2这里面的 Leader 是当前负责该分区读写的主副本Replicas 是分区的全部副本列表Isr 是当前与 Leader 保持同步的副本集合。如果 Isr 数量长期小于 Replicas 数量说明有副本同步滞后或者宕机需要介入排查。另外--describe 还支持加 --under-replicated-partitions只看副本不同步的分区快速定位问题kafka-topics.sh --bootstrap-server broker1:9092 --describe --under-replicated-partitions这个命令在生产排障时非常省事不用一个 Topic 一个 Topic 去看一行就能暴露整个集群的异常分区。2.3 扩容配置与删除注意事项线上流量上涨Topic 分区不够用了需要扩容kafka-topics.sh --bootstrap-server broker1:9092 --alter \ --topic order-events \ --partitions 24这里有个铁律分区数只能增加不能减少。因为减少分区会涉及已有键值数据在分区间的重新分配逻辑上无法做到无损收缩。所以在创建 Topic 时分区数宁可稍微多留一点也不要卡着下限创建否则后面扩容只能往上走而且扩容只会把新增分区放在集群末尾 Broker 上很容易造成负载不均匀。删除 Topic 也不是删完就完事kafka-topics.sh --bootstrap-server broker1:9092 --delete --topic temporary-topic默认情况下Broker 需要开启 delete.topic.enabletrue 才能真正执行删除否则命令虽然返回成功Topic 实际还在。而且删除是异步操作删完之后 zookeeper 或 KRaft 元数据里先标记删除数据文件需要稍后才会清理干净。如果删除后立刻重建同名 Topic偶尔会看到老数据短暂存在这是正常现象等删除流程跑完就好。注意删除 Topic 是不可逆操作数据会全部丢失。我在测试环境经常用删除命令但生产环境执行前一定先确认名字拼写有没有问题最好再检查一下有没有消费者和生产者在连这个 Topic。3. 生产消费链路与消费组偏移实操3.1 命令行生产者的正确用法命令行最常用的数据验证方式就是用控制台生产者手动发消息kafka-console-producer.sh --bootstrap-server broker1:9092 --topic order-events执行后进入交互模式一行一条消息回车就发送。默认不指定 key消息会按轮询方式发到各个分区。如果我想指定 key让相同 key 的消息进同一分区需要这样写kafka-console-producer.sh --bootstrap-server broker1:9092 \ --topic order-events \ --property parse.keytrue \ --property key.separator:然后输入格式为 key:value比如order-1001:{orderId:1001,amount:299}指定 key 的价值在于如果消费者需要保证某个订单的所有事件按顺序处理必须让这些事件落到同一个分区。分区内有序分区间无序这个特性在生产设计里非常关键。命令行生产者默认的 acks 配置跟着 Producer 默认走新版 Kafka 里默认是 all也就是等待所有副本确认。控制台工具还支持直接指定重试次数、压缩类型等必要时可以加kafka-console-producer.sh --bootstrap-server broker1:9092 \ --topic order-events \ --compression-codec lz4 \ --batch-size 16384不过说实话控制台生产者主要用途是验证链路通不通、发几条消息看看效果批量压测或者正式接入还是写代码用 API 更合理。3.2 命令行消费者与消费组消费端最常用的命令是kafka-console-consumer.sh --bootstrap-server broker1:9092 \ --topic order-events \ --from-beginning--from-beginning 表示从分区最早的消息开始消费。如果不加这个参数默认从消费组当前提交的 Offset 开始消费。这里有一个非常容易踩的坑控制台消费者每次启动如果不指定 --group它会随机生成一个 group id。也就是说你在两个终端各起一次消费者不加 --group它们不会互相影响也看不到对方已经消费过的数据。想要两个终端组成同一个消费组共同分担分区消息必须显式指定相同的 group id# 终端A kafka-console-consumer.sh --bootstrap-server broker1:9092 \ --topic order-events \ --group console-group \ --from-beginning # 终端B kafka-console-consumer.sh --bootstrap-server broker1:9092 \ --topic order-events \ --group console-group \ --from-beginning这时你会发现A 和 B 收到的消息是不同分区的合起来才是完整数据流。这就是消费组负载均衡的核心机制一个分区的消息同一时间内只会被组内的一个消费者实例处理。还有一个日常排查特别实用的参数 --max-messages指定消费多少条后自动退出kafka-console-consumer.sh --bootstrap-server broker1:9092 \ --topic order-events \ --group console-group \ --max-messages 5这样不会一直挂在前台刷屏适合脚本里做数据抽样检查。3.3 消费组管理与Offset重置生产环境里消息积压、重复消费、数据重放都是靠消费组命令来处理的。先看消费组列表kafka-consumer-groups.sh --bootstrap-server broker1:9092 --list然后查看某个组的详细消费情况kafka-consumer-groups.sh --bootstrap-server broker1:9092 \ --describe --group console-group输出会列出每个分区、当前消费的 offset、log-end-offset、lag。lag 代表消费落后多少条lag 一直增长说明消费者处理能力跟不上或者消费者挂了。如果业务方要求重新消费一段时间内的消息需要重置 Offset。最常见的有三种场景把消费位置调整到最早kafka-consumer-groups.sh --bootstrap-server broker1:9092 \ --group order-consumer-group \ --topic order-events \ --reset-offsets --to-earliest --execute把消费位置调整到最新也就是跳过当前积压kafka-consumer-groups.sh --bootstrap-server broker1:9092 \ --group order-consumer-group \ --topic order-events \ --reset-offsets --to-latest --execute把消费位置按时间回退比如回到 2 小时前kafka-consumer-groups.sh --bootstrap-server broker1:9092 \ --group order-consumer-group \ --topic order-events \ --reset-offsets --to-datetime 2025-03-10T10:00:00.000 --execute执行 offset 重置时有个前提消费组必须处于无活跃消费者的状态否则会报错。如果消费者还挂着你可以先停掉消费者应用重置完再启动。另外--reset-offsets 默认是 dry-run只会打印将要执行的计划不会真正改看到结果没问题后再加 --execute。提示重置 Offset 是把双刃剑。往最早调会造成大量重复消费往往要把下游的幂等逻辑和缓存清理一并考虑往最新调会跳过积压消息数据丢失。操作前最好和上下游负责人确认一遍。4. 配置管理、分区重分配与性能测试4.1 动态修改Broker与Topic配置Kafka 的配置并不都需要重启 Broker 才能生效。kafka-configs.sh 可以动态修改 Broker、Topic、Client、User 级别的配置这是线上调优的重要入口。查看某个 Topic 当前的动态配置kafka-configs.sh --bootstrap-server broker1:9092 \ --entity-type topics \ --entity-name order-events \ --describe修改 Topic 的保留时间比如把 retention 改成 3 天kafka-configs.sh --bootstrap-server broker1:9092 \ --entity-type topics \ --entity-name order-events \ --alter \ --add-config retention.ms259200000查看 Broker 的敏感配置例如是否允许自动创建 Topickafka-configs.sh --bootstrap-server broker1:9092 \ --entity-type brokers \ --entity-default \ --describe如果只是临时调整某个 Broker 的配置可以在 --entity-name 里指定 Broker ID。这里要提一个常见误区很多人在看某个参数是否生效时习惯查 server.properties 文件。实际上动态修改的配置存在 ZooKeeper 或 KRaft 元数据里优先级高于静态文件配置所以必须用 kafka-configs.sh 才能看到完整状态。我遇到过几次“明明改了文件但没生效”的困惑后来养成习惯线上统一用命令改、命令查静态文件只做初始基线。4.2 分区重分配的完整执行流程分区重分配是运维里比较重的操作常用于以下几种情况新增节点后把旧节点分区迁移过去替换坏盘前把副本挪走均衡磁盘使用率。官方思路是用 kafka-reassign-partitions.sh 分三步走。首先写一个 JSON 文件描述每个 Topic 分区期望的副本分布{ version: 1, topics: [ { topic: order-events, partition: 0, replicas: [2, 1, 3] }, { topic: order-events, partition: 1, replicas: [1, 3, 2] } ] }不过手写这种 JSON 容易出错更稳妥的方式是先用 --generate 让 Kafka 自己生成移动方案kafka-reassign-partitions.sh --bootstrap-server broker1:9092 \ --topics-to-move-json-file topics-to-move.json \ --broker-list 1,2,3,4 \ --generatetopics-to-move.json 只写要迁移的 Topic 名即可{ topics: [ {topic: order-events}, {topic: payment-events} ], version: 1 }生成结果会同时打印 recommended-reassignment-evaluation.json 和当前分配方案我们取推荐方案保存为 reassignment.json然后执行kafka-reassign-partitions.sh --bootstrap-server broker1:9092 \ --reassignment-json-file reassignment.json \ --execute执行后数据会在副本之间拷贝期间查询执行进度kafka-reassign-partitions.sh --bootstrap-server broker1:9092 \ --reassignment-json-file reassignment.json \ --verifyverify 输出显示所有分区都处于 successfully migrated 状态整个流程才算结束。重分配期间流量会有波动集群 IO 和网络都可能打高建议在业务低峰期做并且提前评估目标 Broker 的磁盘剩余空间至少留够一倍的副本数据量因为新副本拷贝完成前不会自动删除旧副本。4.3 顺手好用的性能与存量查询命令集群上线前或者大促前做性能摸底是惯例。Kafka 官方提供了两个压测脚本。生产者压测kafka-producer-perf-test.sh --bootstrap-server broker1:9092 \ --topic perf-topic \ --num-records 1000000 \ --record-size 1024 \ --throughput 100000 \ --producer-props acksall linger.ms5 batch.size16384消费者压测kafka-consumer-perf-test.sh --bootstrap-server broker1:9092 \ --topic perf-topic \ --messages 1000000 \ --threads 4压测结束后会输出平均吞吐MB/s、平均延迟等关键指标。注意压测用的 Topic 尽量单独创建不要对着生产业务 Topic 压避免污染线上数据。压测完及时清理测试 Topic。还有一个查看某个 Topic 当前消息总数的快捷命令kafka-run-class.sh kafka.tools.GetOffsetShell \ --broker-list broker1:9092 \ --topic order-events --time -1输出每个分区的 log-end-offset。把各分区数值加起来就是总消息数近似值排查数据对账问题时我经常用。5. 命令行实践常见问题排查实录5.1 环境与连接类问题第一种常见报错是“Could not find or load main class kafka.tools.GetOffsetShell”或者直接提示类不存在。这类问题多半是脚本版本和安装方式不匹配或者 KAFKA_HOME 环境变量指错了路径。我一般先确认 which kafka-run-class.sh 指向的是不是预期目录再检查 libs 目录下有没有对应 JAR 包。第二种是连接超时Error while executing topic command : Replication factor: 3 larger than available brokers: 1这句话经常出现在新集群或容器化环境里意思是副本因子设了 3但可用 Broker 只有 1 个创建失败。生产环境有 3 台以上 Broker 就不会有这个错单机学习环境建议先设成 1。第三种是 bootstrap-server 地址写错。有人习惯把全部 Broker 都写进去其实没必要。bootstrap-server 只需要一个可用 Broker客户端会通过它拉取元数据。但要注意这个地址必须是客户端能真正访问的地址如果写的是 localhost而 Broker 部署在另一台机器上就会一直连不上。排查技巧是在客户端机器上用 telnet 测一下端口telnet broker1 9092能通再继续查应用配置。5.2 数据生产消费异常类问题消费不到数据是最常被问的问题。根据我的经验按以下顺序排查基本都能解决。先确认生产者有没有真的把消息发进去kafka-console-consumer.sh --bootstrap-server broker1:9092 \ --topic order-events --from-beginning --max-messages 10如果这个能查到数据说明数据链路和 Topic 本身没问题问题出在你自己的消费组身上。接着查消费组状态kafka-consumer-groups.sh --bootstrap-server broker1:9092 --describe --group your-group看 LAG 列。如果 LAG 为 0说明消息都没了可能被别的消费者消费过或者 Offset 被重置到最新了。如果 LAG 很大但又消费不到多半是消费者代码里过滤条件太严不是 Kafka 的问题。还有个容易忽略的坑同一个消费组下如果之前有消费者实例挂掉分区被重新分配给其他实例但新实例启动时从提交过的 Offset 继续消费。如果这个 Offset 已经大于当前 log-end-offset比如数据被删过或者 Topic 做了重建客户端会直接跳到最新表现为消息丢失。这种情况用 --reset-offsets --to-earliest 重置一次就能恢复。另外生产端如果发现消息发不进去优先查认证和配额。Kafka 开启了 ACL 之后命令行工具需要额外指定消费者或生产者权限比较省事的方式是在命令里加 --command-config 指向一个包含认证信息的配置文件kafka-console-producer.sh --bootstrap-server broker1:9092 \ --topic order-events \ --command-config client.propertiesclient.properties 里写好 security.protocol、sasl.mechanism、sasl.jaas.config 这些。不要在生产环境用明文密码裸敲命令命令历史里全是敏感信息。5.3 遗留坑位与习惯建议我在一台机器上并行操作多个 Kafka 集群时踩过不少坑。最典型的是环境变量没切干净同一个 shell 里 KAFKA_HOME 指向旧版本目录执行删除 Topic 命令时行为和新版完全不一样。现在我的做法是每个集群的维护都写成独立脚本脚本开头 source 对应的环境变量文件。再一个是命令中时间格式的坑。--to-datetime 的参数格式是严格匹配的不同版本要求“2025-03-10T10:00:00.000”这种带毫秒的格式少写 .000 有的版本能解析有的版本直接报错。如果时间格式不确定先执行 dry-run看到输出符合预期再 --execute。最后一个建议是习惯性把关键操作记录成操作文档。命令行工具没有图形界面也没有审计界面谁在什么时候执行过什么命令如果不留日志出了问题很难追溯。我在团队里约定删除 Topic、重置 Offset、分区重分配这类高危操作执行前必须把命令内容和影响范围发到群里确认执行后把输出截图存档。这套流程看起来麻烦但真正救过几次火。如果你刚开始接触 Kafka 命令行不用急着把所有工具都背下来。先把 kafka-topics.sh 和 kafka-consumer-groups.sh 这两个玩熟覆盖了绝大多数日常场景。剩下的命令用到的时候再查配合 --help 参数很快就能形成自己的命令手册。命令行这个东西多敲几次自然就顺手了。
