说句实在话Kafka 的本地测试环境是我见过最容易被轻视但又最值得投入的一环。很多人觉得本地就是把 Kafka 下载下来解压启动然后跑一下生产消费就完事了。但等真的开始写业务代码、调参数、验数据的时候各种边界问题就全冒出来了。这套基于kafka_2.13-4.2.0的本地测试脚本就是冲着这个痛点去的把那些复读机一样的命令封装成一套干净利落、能反复执行、能出报告、还能当调试工具的本地测试体系。它能解决什么问题说白了就三件事环境拉起要够快、测试数据要可控可重复、问题定位要有抓手。无论你是在做消息队列的技术选型对比还是在排查线上消费延迟或者单纯想搞懂 Kafka 的分区、副本、位移这些概念这套东西都能给你省下大量重复劳动。我写这套脚本的时候目标对象很明确一个是刚接触 Kafka 的新手照着一跑就能看到消息从生产到消费的全过程另一个是已经写过生产级代码的开发者需要一个稳定、可控、可复现的本地环境来验证消费组重平衡、位移提交、吞吐量这些细节。整套脚本包含环境启动、主题管理、生产测试、消费测试、性能压测、清理回收几个模块全部命令行可跑没有任何图形界面依赖。1. 脚本设计的整体思路把“临场发挥”变成“标准流程”1.1 本地测试场景下的真实痛点如果你只是偶尔跑一两次 Kafka手动敲命令其实没有太大问题。但我敢说只要你连续调试三五天一定会遇到下面这些情况环境变量没有写进 shell 配置每次新开终端都要export PATH一遍烦到怀疑人生。生产端测试数据永远靠手敲测一次换一条想模拟批量消息或者带 key 的消息得现场拼 JSON。消费端日志刷得飞快要看某个分区的位移提交情况得从终端缓存里往上翻半天。集群重启之后上次创建的主题还在不在副本状态正常不正常得手动执行查询命令。最难受的是想对比acksall和acks1在本地环境的延时差异结果发现光改参数重启客户端就要折腾十分钟。这些零碎的痛点叠加起来就是你写脚本最原始的动力。这套脚本不是一拍脑袋写出来的而是把日常操作一步步拆解成固定模块之后的产物。1.2 脚本模块拆解与整体结构我推荐的方式是用一个统一的目录承载所有脚本目录命名清晰每个脚本只干一件事并且相互之间通过 shell 函数或者简单的参数透传来协作。整体结构长这样kafka-local-test/ ├── env.sh # 环境变量与环境检查 ├── start-kafka.sh # 启动 zookeeper kafka ├── stop-kafka.sh # 优雅停止 清理 ├── topic-mgr.sh # 主题管理的增删查改 ├── producer-test.sh # 生产端测试数据生成与发送 ├── consumer-test.sh # 消费端验证与位移查看 ├── perf-test.sh # 性能和延时压测 └── logs/ # 运行日志目录脚本自动创建每个脚本内部做的事情都很克制没有做成一个大而全的“瑞士军刀”。这个设计原则是我踩过坑之后总结出来的脚本之间职责单一你才能在出问题的时候快速定位。比如消费者组一直报 Rebalance你根本不用怀疑是不是启动脚本把某个 broker 参数改坏了。1.3 为什么选择 Shell 而不是其他语言这时候肯定有人问为什么不用 Python 或者 Go 来写测试工具答案很直接Kafka 自带的bin目录下已经有足够好用的命令行工具。kafka-console-producer.sh、kafka-console-consumer.sh、kafka-topics.sh、kafka-producer-perf-test.sh、kafka-consumer-perf-test.sh这些工具本身就是官方提供的最标准的测试入口。用 Shell 做一层薄封装最大的好处是零额外依赖只要机器上有 JDK脚本把环境变量设置好就能直接跑起来。Python 方案还得考虑confluent-kafka或者kafka-python的依赖安装和版本兼容问题反而把简单的事情变复杂。2. 环境准备与核心配置先把地基打好2.1 版本选型与前置依赖这次用的是kafka_2.13-4.2.0这个命名看着长其实拆开很直观2.13是 Kafka 编译时使用的 Scala 版本4.2.0是 Kafka 自身的版本号。选 Scala 2.13 编译版本是因为它兼顾了生态兼容性和新特性支持绝大多数开源组件目前都基于这个版本适配。前置环境只有两件事JDK 版本Kafka 4.x 要求 JDK 8 以上但为了稳妥和性能表现我建议直接用 JDK 17。实测下来JDK 17 在 G1 垃圾回收器的配合下本地启动速度和堆内存占用表现都比 JDK 8 要好一些。磁盘空间Kafka 的日志目录默认会保留log.retention.hours168也就是 7 天的数据。本地测试如果疯狂压测日志会快速膨胀。建议单独指定一个log.dirs并且把保留时间调短一点比如 24 小时。解压之后不要急着改任何配置先看一眼bin目录下的脚本确认有没有kafka-server-start.sh和zookeeper-server-start.sh。Kafka 4.x 虽然已经在推进 KRaft 模式取代 ZooKeeper但 4.2.0 这个版本对传统 ZooKeeper 模式的支持依然非常稳定本地测试用 ZooKeeper 模式是最不容易踩坑的选择。2.2 server.properties 关键参数解读很多新手拿到config/server.properties直接不改就启动这也能跑通但如果你想在本地模拟多分区、多副本的场景有些参数必须改。我建议关注的几个核心参数如下参数本地测试推荐值说明broker.id1单机测试固定即可listenersPLAINTEXT://localhost:9092本地测试不需要外网监听绑定 localhost 更安全也避免防火墙干扰log.dirs/tmp/kafka-logs统一放到临时目录测试完毕一键清理num.partitions3默认主题分区数3 个分区方便验证分区分配和消费并发offsets.topic.replication.factor1本地单 broker副本因子只能设 1设 2 或 3 会报错log.retention.hours24缩短日志保留时间避免压测之后磁盘爆炸auto.create.topics.enabletrue生产测试时如果主题不存在Kafka 会自动创建省一步手工操作这里特别强调offsets.topic.replication.factor。我见过不少人在本地测试时直接把生产环境的配置模板复制过来结果消费者组位移提交一直报错日志里刷Replication factor is larger than available brokers原因就是副本因子设了 3但本地只有一个 broker。新手在这个问题上卡一整天的情况太常见了。2.3 启动前的环境变量与 JVM 配置脚本里我写了一个env.sh统一处理环境变量和基础检查。这里有一个细节容易被忽视Kafka 自带的启动脚本会读取KAFKA_HEAP_OPTS环境变量如果没有显式设置默认堆内存可能只有 1G。本地测试虽然用不了多少内存但压测的时候如果堆内存太小GC 频率会直接影响测试结果的准确性。# env.sh 核心内容 export KAFKA_HOME/opt/kafka_2.13-4.2.0 export PATH$KAFKA_HOME/bin:$PATH export KAFKA_HEAP_OPTS-Xms512m -Xmx512m export KAFKA_JVM_PERFORMANCE_OPTS-server -XX:UseG1GC -XX:MaxGCPauseMillis20为什么堆内存设 512M 而不是更大因为本地测试场景下Kafka 本身用作消息中转256M 到 512M 完全够用。堆设得太大反而会让启动变慢而且垃圾回收时 Full GC 的停顿时间更长。这里的原则是测试环境要贴近最小可用配置而不是贴近生产配置否则你用本地环境测出来的参数到了生产环境根本不具备参考性。3. 核心测试脚本的实操实现3.1 统一启动与优雅停止启动脚本的思路很简单但有几个细节值得讲。Kafka 依赖 ZooKeeper 维护元数据所以启动顺序一定是先 ZooKeeper 后 Kafka。如果想知道服务是否真的起来了不能只看进程有没有还得看端口监听状态。更稳妥的做法是发一个描述 Broker 信息的请求确认 broker 已经注册到 ZooKeeper。# start-kafka.sh 核心逻辑 zookeeper-server-start.sh -daemon $KAFKA_HOME/config/zookeeper.properties sleep 3 kafka-server-start.sh -daemon $KAFKA_HOME/config/server.properties sleep 5 # 验证 broker 是否注册成功 kafka-broker-api-versions.sh --bootstrap-server localhost:9092 | head -20停止脚本这里要重点说一句。很多人图省事直接kill -9杀掉 Kafka 进程如果是在本地测试短时间看好像也没什么问题。但 Kafka 的日志目录里有未刷盘的 segment 文件强制杀死进程之后下一次启动时需要进行日志恢复。日志恢复期间broker 虽然起来了但分区可能处于Offline状态消费者根本拉不到数据。我推荐的方式是调用kafka-server-stop.sh它会向进程发送 SIGTERM让 Kafka 有足够的时间完成日志刷盘和优雅关闭。# stop-kafka.sh 核心逻辑 kafka-server-stop.sh sleep 5 zookeeper-server-stop.sh sleep 3 # 二次确认端口已经释放 if lsof -i:9092 -i:2181 | grep -q LISTEN; then echo Warning: port still in use, manual cleanup required fi脚本最后加了这个端口的二次检查看起来简单但实际价值很大。因为 Kafka 偶尔会出现进程虽然收到停止信号但 socket 没有完全释放的情况。如果下一次启动时端口还在被占用kafka-server-start.sh会直接报Address already in use而且这个报错信息藏在日志最后面容易漏看。3.2 生产端测试脚本从手工敲命令到参数化批量发送生产端脚本是我整个测试体系里用得最频繁的一个。裸用kafka-console-producer.sh的体验其实很一般因为它默认从标准输入读取数据你只能一条一条手敲想控制 key、分区、消息内容格式都比较费劲。而封装后的脚本可以做三件裸命令很难做的事生成带 key 的消息验证分区路由逻辑。控制消息发送条数和间隔模拟突发流量或平稳流量。支持从文件读取消息体为消费端准备可预期的数据。# producer-test.sh 核心逻辑 # 参数说明 # -t 主题名 # -n 消息条数 # -k 是否生成 key # -f 消息体文件可选 while [[ $# -gt 0 ]]; do case $1 in -t) TOPIC$2; shift ;; -n) NUM$2; shift ;; -k) WITH_KEYtrue ;; -f) BODY_FILE$2; shift ;; esac shift done for i in $(seq 1 $NUM); do KEYpartition-key-$((i % 3)) VALUE{\id\:$i,\ts\:$(date %s),\msg\:\test-message-$i\} if [ $WITH_KEY true ]; then echo $KEY::$VALUE else echo $VALUE fi done | kafka-console-producer.sh \ --bootstrap-server localhost:9092 \ --topic $TOPIC \ --property parse.keytrue \ --property key.separator:: \ --property acksall注意到ack参数我直接写到了 producer 脚本参数里了。为什么要用acksall因为本地测试的目的是验证逻辑正确性不是追求吞吐量。acksall表示 leader 分区副本收到消息并且所有 ISR 副本都同步完成之后才返回确认。单 broker 环境下ISR 只有一个副本acksall和acks1在延迟上几乎没有差别但如果你以后把脚本拿到多节点环境跑acksall能更真实地反映生产环境的确认链路。还有一个容易忽略的细节是parse.keytrue。加了这个配置之后producer 会把分隔符前面的部分当作消息 key后面的部分当作 value。没有这个配置整行内容都会被当作 value 发送。我刚开始测试的时候忽略了这一点导致消费端看到的消息 key 全是 null后续想按 key 做分区聚合分析时完全没法用。3.3 消费端测试脚本不只是看消息更要验证位移管理消费端测试脚本是整个体系里最能体现“深度”的部分。普通测试只需要看到消息打出来就行但真正做消息队列验证你还得回答这几个问题消息是从哪个分区消费的消费组当前的位移提交到了哪里消费组是否发生了重平衡消息是第一次消费还是从最早的位移重新拉取的# consumer-test.sh 核心逻辑 # 模式一从最新位移开始消费适合验证实时消息推送 kafka-console-consumer.sh \ --bootstrap-server localhost:9092 \ --topic $TOPIC \ --group $GROUP \ --from-latest \ --property print.keytrue \ --property print.partitiontrue \ --property print.offsettrue \ --timeout-ms 5000 # 模式二从最早位移重新消费适合验证历史数据完整性 kafka-console-consumer.sh \ --bootstrap-server localhost:9092 \ --topic $TOPIC \ --group $GROUP \ --from-beginning \ --property print.keytrue \ --property print.partitiontrue \ --property print.offsettrue--timeout-ms 5000这个参数很实用它让消费端在没有新消息到达时5 秒后自动退出而不是一直挂在那里刷空循环。这在脚本化测试里非常重要因为自动化流程不允许一个命令永远不结束。加了print.partitiontrue和print.offsettrue之后控制台输出的每一行都会带上分区号和位移号。这相当于在消费端看到了消息的“元数据轨迹”排查消息是否出现乱序、重复消费或者数据倾斜问题时这个输出格式是决定性的线索。消费端脚本还有一个配套手段就是单独查看消费组的位移情况。消费端的console-consumer退出之后不会自动展示 lag 信息。想看 lag得用kafka-consumer-groups.shkafka-consumer-groups.sh \ --bootstrap-server localhost:9092 \ --group $GROUP \ --describe输出结果里有CURRENT-OFFSET、LOG-END-OFFSET和LAG三列。LAG是 0 的时候说明消费端已经追平了所有消息LAG持续大于 0 且不下降说明消费逻辑存在瓶颈。我排查本地消费延迟问题时这个命令是使用频率最高的没有之一。3.4 性能压测脚本吞吐量和延时的量化标尺本地测试如果只是为了“能通”那 Console 生产者消费者已经够了。但如果你在做 Kafka 和 RabbitMQ 的选型对比或者想验证不同批量参数对吞吐量的影响就需要用到性能压测脚本。Kafka 自带的kafka-producer-perf-test.sh非常强大支持控制消息条数、消息大小、吞吐上限、确认机制等多个维度。# perf-test.sh 核心逻辑 kafka-producer-perf-test.sh \ --topic $TOPIC \ --num-records 100000 \ --record-size 1024 \ --throughput 5000 \ --producer-props \ bootstrap.serverslocalhost:9092 \ acksall \ linger.ms20 \ batch.size32768 \ buffer.memory67108864参数解释一下--num-records 100000总共发送 10 万条消息。数量太少统计误差大数量太多本地环境要等太久10 万条是性价比很高的选择。--record-size 1024每条消息 1KB。这个大小模拟的是比较典型的业务消息体如果测试 JSON 消息或者日志消息1KB 的设定是合理的。--throughput 5000限制发送速率最多每秒 5000 条。这个参数很关键不加的话生产端会以最大能力疯狂发送结果就是堆内存瞬间被打满压测的数值完全失真。限制了吞吐之后你可以观察在可控速率下消费端能不能跟上延迟是否稳定。linger.ms20生产端等待 20 毫秒再批量发送。这个参数对吞吐量的影响非常直接。linger.ms0时每条消息立即发送吞吐量全看网络往返linger.ms20时生产端会积攒 20 毫秒内的消息一起发送批量效率大幅提升同时 20 毫秒的额外延迟在本地测试中几乎无感。压测结束后脚本输出里会有50th、95th、99th的延迟分位数这些数据是本地测试最有参考价值的产出。举个例子如果 99 分位延迟比 50 分位高出好几倍说明存在明显的尾部延迟这种延迟通常会指向 GC 停顿或者批量发送的大小不均匀需要进一步排查。消费端压测用kafka-consumer-perf-test.sh但要注意消费者压测必须在消息已经准备好的前提下进行否则测出来的数据其实是“等待消息”的时间而不是真实的消费耗时。我的习惯是先跑一遍生产端脚本灌满数据再单独跑消费端压测这样才能得到干净的消费吞吐量指标。4. 常见问题与排查实录写脚本只是第一步真正让这套东西有价值的是把它在跑的过程中遇到的问题沉淀成排查手册。以下每一个问题都在我本地测试时真实触发过并且几乎都能在热搜词里找到对应的高频报错。4.1 生产者报 TimeoutException 或连接拒绝症状执行生产脚本后控制台刷出一堆org.apache.kafka.common.errors.TimeoutException或者Connection refused。排查顺序按下面这三步来先用lsof -i:9092确认端口到底有没有监听。如果端口没有监听看 Kafka 启动日志有没有报Address already in use这种情况多半是上一次的进程没杀干净端口被残留进程占用。我的脚本里特意加了二次确认的检查就是为了拦住这种情况。如果端口正常监听检查server.properties里的listeners配置。如果配置成了PLAINTEXT://0.0.0.0:9092而本地代码连接时用的是localhost:9092看起来也能通但如果代码里写的是机器名或者其它 IP就会因为监听地址不匹配而连接失败。本地测试统一改成PLAINTEXT://localhost:9092最省心。注意listeners和advertised.listeners是两回事。单机本地测试可以不配置advertised.listeners默认情况下它会继承listeners的地址。但如果你用 Docker 或者虚拟机跑必须在advertised.listeners里显式声明宿主机能访问的地址不然后果非常诡异——生产端能连上 broker但 broker 返回的元数据里写的是容器内网地址再往那个地址发数据就完全连不上。4.2 消费者启动后一直收不到消息这个问题出现频率极高而且往往不是环境问题而是消费参数问题。最常见的原因是消费者启动时带了--from-latest而生产脚本早就把数据发完了。--from-latest的意思是“只消费启动之后新到达的消息”之前已有的消息一条都不会拉取。此时的消息队列已经存在历史数据消费者却因为设置了--from-latest而看不到完全是预期行为不是故障。排查方法很简单先用kafka-consumer-groups.sh看这个消费组的CURRENT-OFFSET和LOG-END-OFFSET。如果LOG-END-OFFSET明显大于CURRENT-OFFSET且不是 0说明确实有历史消息堆积换成--from-beginning就能重新消费到。这个“坑”其实反映了消费位移的两种语义--from-beginning从分区最早可用的位移开始消费可以理解为把历史数据全部读一遍。--from-latest从消费者启动那一刻开始只消费新消息适合实时推送场景。本地验证历史数据时用--from-beginning验证推送链路时用--from-latest两者混用就会踩坑。4.3 消费组反复出现 Rebalance症状消费端日志里出现Revoking assigned partitions、Assigning new partitions之类的内容并且循环出现。消费端一直在拉取消息和处理消息之间反复横跳实际处理的消息量极低。Kafka 的消费组成员通过心跳机制向 group coordinator 汇报存活状态。session.timeout.ms是默认的会话超时时间如果消费者在超时时间内没有发心跳coordinator 就会判定它挂了触发重平衡把分区分给其它消费者。本地测试时如果消费者的逻辑里有一个耗时的数据处理过程比如写数据库或者调远程接口单条消息处理时间超过了心跳间隔就会触发一连串的 Rebalance。解决办法并不是把session.timeout.ms无限调大而是理解这两个参数max.poll.interval.ms两次poll()调用之间的最长间隔默认 5 分钟。max.poll.records单次poll()最多返回的消息条数默认 500 条。本地测试场景下我建议把max.poll.records调小到 100 条左右。这样单次拉取的数据量少了处理时间自然缩短不容易踩到心跳超时的边界。同时配合脚本参数验证多线程消费时的顺序性问题——如果业务代码里开了多个线程处理同一分区的消息Rebalance 的概率会直线上升这个问题在热搜词里排得很靠前我在本地测试里也花了不少时间才彻底搞清楚线程模型和分区模型的对应关系。4.4 OffsetOutOfRange 与位移丢失问题有时候消费者报OffsetOutOfRangeException分成两种情况。如果CURRENT-OFFSET大于LOG-END-OFFSET说明消费位移已经超过了分区里最新的消息位置。常见原因是消息设置了过期时间被删掉了或者日志目录被手动清理过。如果CURRENT-OFFSET小于最早的消息位移说明消费者尝试从已经过期的位移位置开始消费。本地测试时这个问题的根源经常是测试脚本在短时间内反复创建消费组导致位移提交到__consumer_offsets主题但清理日志又把位移给清了。要避免这个问题最简单的方式是每个独立的测试场景用不同的消费组名不要复用同一个 group id。脚本里我建议在 group 命名时加上时间戳或者场景名方便后面识别。5. 脚本的扩展方向与我的实操心得5.1 从脚本到自动化测试平台如果你觉得到这里就完事了那这套东西的潜力其实还没完全用上。脚本本身就是一层壳真正的价值在壳里面的测试逻辑。我后续做的扩展主要有两个方向。第一个方向是把压测脚本的输出解析出来生成一份简单的 HTML 报告。kafka-producer-perf-test.sh输出的是文本格式的性能数据你可以在脚本里用tee同时把输出写到文件再用简单的 awk 或者 Python 脚本把 50 分位、99 分位延迟抽出来拼到一个模板里。这一步虽然简陋但已经能支撑基本的选型对比汇报了。我拿这套脚本跑过 Kafka、RabbitMQ 和 RocketMQ 的对比测试三者各自的吞吐量、延迟特征、确认机制差异非常直观地体现在报告里。第二个方向是把主题创建的工作提前做掉。auto.create.topics.enable虽然默认是 true但自动创建的主题使用默认分区数和副本因子。如果你想测试 8 分区的消费并发表现必须显式创建主题。我的脚本里加了一个前置步骤每次生产测试前先检查主题是否存在不存在就用kafka-topics.sh --create创建并显式指定分区数和副本因子。# 主题创建前置检查 kafka-topics.sh --bootstrap-server localhost:9092 \ --describe --topic $TOPIC 2/dev/null | grep -q Topic: $TOPIC if [ $? -ne 0 ]; then kafka-topics.sh --bootstrap-server localhost:9092 \ --create --topic $TOPIC \ --partitions 3 --replication-factor 1 fi这个前置检查避免了我手动去记“哪些主题已经建过”也避免了自动创建带来的分区数不可控问题。5.2 我在实操中总结的几个习惯有些习惯看起来很琐碎但长期跑 Kafka 测试的收益非常大。第一日志文件一定按日期归档。Kafka 的启动日志默认是追加写到同一个文件跑几天之后文件会变得巨大。每次排查问题都要用tail -n 5000去翻。我在start-kafka.sh里加了一句日志重定向用当前日期命名日志文件kafka-server-start.sh -daemon $KAFKA_HOME/config/server.properties 21 | tee -a logs/kafka-$(date %Y%m%d).log如果有一天要复盘前天跑压测时 broker 内部发生了什么直接点开对应日期的日志文件就能看效率提升不是一星半点。第二生产端脚本的数据要可预期。我在脚本里生成的所有消息都带有id和ts字段。消息 ID 按序号递增消费端只要把收到的消息 ID 落下来就能直接在本地验证有没有乱序、有没有重复、有没有丢消息。这个做法做不了什么架构上的创新但它的确帮我发现过一次由于网络重试导致的消息重复投递问题。第三压测之前先清空目标主题。很多人都忽略了一个细节如果你往同一个主题里重复发送压测数据LOG-END-OFFSET会越堆越高消费端每次测试都要从头处理历史数据压测结果会受到历史数据的干扰。我写了一个清理主题的方法用kafka-topics.sh --delete删除之后立刻重建同名的主题这样每次压测都从零开始。# 清理主题并重建 kafka-topics.sh --bootstrap-server localhost:9092 --delete --topic $TOPIC sleep 2 kafka-topics.sh --bootstrap-server localhost:9092 \ --create --topic $TOPIC \ --partitions 3 --replication-factor 1这个操作看起来简单但要注意删除主题不是即时完成的Kafka 会把主题标记为删除后台异步清理。所以删除之后要加一个 2 秒的等待否则立刻创建同名主题可能会因为元数据没有完全刷新而报Topic exists already。第四理解脚本里的每个参数为什么要存在。这套脚本最后变成了我复盘 Kafka 原理的“活教材”。比如linger.ms和batch.size这两个参数不看压测数据你不会理解它们对吞吐量的影响到底有多大。acks参数不跑一遍对比你也不会直观地感知到“至少一次”和“最多一次”两种投递语义在实际消费体验里的差别。写脚本的过程本质上就是把这些散落的概念串成一条可执行的验证链。今天排查的消息重复问题明天就会成为你面试时回答“如何保证消息不丢失”的底气来源。我历来觉得测试脚本不是写完就丢的工具它应该跟着你的理解一起成长。一开始能满足“启动 收发”就够了跑熟了之后自然会往上加性能压测、位移监控、主题清理这些模块。这套基于kafka_2.13-4.2.0的脚本目前已经稳定陪我跑了好几个项目从选型对比到线上问题复现它都提供了一个足够快和足够确定的本地环境。后面如果我把消息读写最大吞吐和硬件配置之间的关系也梳理成一份测试模板大概率也是在这套脚本的基础上做的。所以说别小看本地测试脚本这件“小事”把它打磨顺手了你省下来的时间可以干很多更有价值的活儿。
