JMeter压测Kafka：手写Java脚本攻克SASL认证难题

这周接了一个压测需求，要给某个内部Kafka主题灌一批业务数据，规模大概几十万条，消费端开了SASL认证，而且不是常见的PLAINTEXT全局认证，只是针对特定用户做了SASL_PLAINTEXT。我第一个反应是找现成的JMeter Kafka插件，结果试了一圈，要么版本和JMeter对不上，要么SASL参数写死没法灵活配置，最坑的是压测机上还不能随便改JMeter启动参数，得保证别人能直接打开工程。于是我只能老老实实自己写JMeter的Java脚本。这里说的“Java脚本”有两层意思：一个是JSR223 Sampler里跑Groovy脚本，看着像Java；另一个是把Java代码打成jar包后塞进JMeter的Java Request采样器。本文就从第一次动手的角度，把SASL认证配置、依赖包处理、代码写法、踩坑排查完整梳理一遍，给后面要干同样事情的同学省点时间。

1. 为什么要用JMeter脚本而不是Kafka插件灌数据

1.1 任务本身的约束条件

压测场景往往有几个硬性约束：不能动服务端配置、不能要求所有执行机都装额外客户端工具、不能把一次性脚本写死在一台机器上。这次的需求除了要往Kafka里传数据，还要求数据里包含特定的用户ID、订单号、时间戳，并且需要按照某个维度把消息分区，比如同一个userId的消息必须落到同一个partition。这种场景下，简单用kafka-console-producer逐行输入显然不现实，用现成的JMeter插件又遇到认证参数适配问题。

我看了一下当时可选的方案：一是用JMeter自带的JSR223 Sampler，在脚本里直接调Kafka客户端库往里发；二是写一个实现JavaSamplerClient接口的类，打包成jar放进JMeter的lib/ext目录，然后通过java Request采样器调用。两种方式本质上都是用Kafka客户端Producer API，区别在于脚本方式改动方便，Java Request方式对代码组织更友好、方便在IDE里调试。

为什么最终没选插件？除了认证适配问题，还因为插件本身的可维护性差。压测团队的人不一定会去看插件源码，出了问题没法快速定位。而自己写脚本，代码逻辑透明，出问题可以直接断点调试。尤其是在SASL认证这种环节，插件如果暴露出“认证类型不支持”的报错，你往往只能干瞪眼。

1.2 手写脚本适合什么场景

手写脚本并不是最优解，它更适合下面这些场景：

• 需要定制消息体结构，比如JSON、Avro，或者要根据JMeter参数动态生成。
• 压测环境中存在复杂的SASL/SSL认证，且不同环境配置还不一样。
• 需要精确控制发送批量大小、linger.ms、重试次数等Producer参数来模拟特定行为。
• 后续还要在CI/CD流水线里跑，脚本可以直接打包进Maven工程统一管理。

如果你的需求只是简单往一个不需要认证的topic里发一些固定字符串，那用插件或者KafkaProduceSampler这类现成组件就够了，没必要大动干戈。但如果跟我一样遇到奇奇怪怪的认证，还是老老实实自己写吧。

2. 环境准备：依赖、JAAS与SASL参数详解

2.1 需要的Kafka客户端依赖

在动手写代码之前，先把依赖准备齐全。JMeter本身不内置Kafka客户端，所以需要手动下载kafka-clients的jar包。要注意版本匹配问题，尽量选和你线上Kafka Broker版本接近的客户端版本。我当时用的是2.8.0的客户端打2.6.0的Broker，认证协议SASL_PLAINTEXT，跑起来没冲突。

实际需要的最小集合是：

• kafka-clients-2.8.0.jar
• slf4j-api-*.jar
• 一个slf4j绑定实现，比如slf4j-simple-*.jar或者logback-classic-*.jar（不强制，但Kafka客户端默认会用slf4j输出日志，没有绑定会打印警告）

依赖jar放哪里？JMeter的lib目录是共享库目录，会加载给所有测试计划用；lib/ext目录则用来存放JMeter扩展组件。如果只是给脚本用，建议放在lib目录下。但是要注意：如果lib目录里已经有某个旧版本的kafka-clients，可能引起版本冲突。因此更稳妥的做法是把这些jar放在一个独立目录，通过JMeter的类路径加载，或者在测试计划里用JSR223脚本设置ClassLoader。最简单还是丢进lib，但先检查一下有没有同名旧包。

2.2 SASL认证的核心：JAAS配置文件

Kafka的SASL认证靠JAAS（Java Authentication and Authorization Service）提供用户凭据。JMeter本身不管理Kafka的认证配置，所以必须显式告诉Kafka客户端用户名和密码，通常有三种方式：

1. 在JVM系统属性里指定java.security.auth.login.config指向一个JAAS文件。
2. 在Producer配置里直接传入sasl.jaas.config（这种方式最灵活，推荐）。
3. 手动设置Configuration对象（一般很少用）。

第2种方式最省心，因为JMeter里可以写死，也可以参数化。如果你用的是Java Request打包模式，还能随时从JMeter变量里取。示例配置片段：

props.put("security.protocol", "SASL_PLAINTEXT"); props.put("sasl.mechanism", "PLAIN"); props.put("sasl.jaas.config", "org.apache.kafka.common.security.plain.PlainLoginModule required username=\"testuser\" password=\"testpass\";");

注意这里username和password不是Kafka的明文账户，而是你SASL认证体系里的用户凭据。如果Kafka配置了SASL/SCRAM，那机制要换成SCRAM-SHA-256或SCRAM-SHA-512，sasl.jaas.config里的LoginModule也要换成ScramLoginModule。

2.3 Broker端参数和JMeter没有关系，但要了解

你可能需要和运维同学确认一下Broker端的listener.security.protocol.map和super.users，不是说非得知道，而是报错时方便排查方向。常见的是SASL_PLAINTEXT对应一个listener端口，例如PLAINTEXT://host:9092和SASL_PLAINTEXT://host:9093。但很多公司为了简化，会把所有端口都配成SASL_PLAINTEXT。如果连接失败，先telnet一下端口，看是不是防火墙问题。

我那次遇到的第一个报错就是org.apache.kafka.common.security.auth.Login，检查了很久发现是启动JMeter时没有加载JAAS配置。后来改成在代码里直接设置sasl.jaas.config，就没有这个问题了。

3. 方式一：JSR223 Sampler + Groovy脚本（最快路径）

3.1 为什么选Groovy而不是纯Java

JSR223 Sampler支持多种语言，Groovy是最主流的选择。主要原因是JMeter默认带Groovy引擎，不用额外装包，而且Groovy语法和Java几乎无缝兼容，Java代码可以直接粘进去。如果你在IDE里写过Java，那写Groovy只是少些分号、多些动态糖。

当然也有不少人坚持用JSR223的java语言，但注意JDK自带的JavaScript引擎是Nashorn，不是Java语言本身，你写不了Java源码。所以除非你自己实现了一个Java脚本引擎并注册到JMeter里，否则直接用java作为语言基本是不能用的。本文讲的“Java脚本”从实际落地看，就是用Groovy这种JVM语言完成原本Java代码该做的事。

3.2 在JMeter里添加JSR223 Sampler

操作非常简单：测试计划里右键添加，选择“取样器”里的“JSR223 Sampler”。语言选Groovy，脚本框里粘贴代码。关键点在于，JMeter会提供一个vars对象（访问JMeter变量），你可以把主题名、bootstrap.servers等作为JMeter变量传进来，这样脚本不用改代码就能切换环境。

一个完整的发送消息脚本示例如下：

import org.apache.kafka.clients.producer.KafkaProducer import org.apache.kafka.clients.producer.ProducerRecord import org.apache.kafka.clients.producer.ProducerConfig import java.util.Properties def props = new Properties() props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, vars.get("kafka_servers")) props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringSerializer") props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringSerializer") props.put("security.protocol", "SASL_PLAINTEXT") props.put("sasl.mechanism", "PLAIN") props.put("sasl.jaas.config", 'org.apache.kafka.common.security.plain.PlainLoginModule required username="' + vars.get("kafka_user") + '" password="' + vars.get("kafka_pass") + '";') def producer = new KafkaProducer(props) // 这里的userId和orderId可以从JMeter变量取，也可以自己在脚本里构造 def userId = vars.get("userId") def orderId = vars.get("orderId") def msg = "{\"userId\":\"" + userId + "\",\"orderId\":\"" + orderId + "\",\"ts\":\"" + System.currentTimeMillis() + "\"}" def record = new ProducerRecord(vars.get("kafka_topic"), userId, msg) producer.send(record) producer.flush() producer.close()

3.3 这段代码里容易被忽略的细节

每次执行要不要new Producer？

上面示例里每发一条消息就创建了一个Producer，这在低压单次测试里可以，但如果用JMeter的线程组跑压测，绝对不能这么做。每个线程应该只创建一个Producer，然后复用。正确做法是使用setUp线程组初始化Producer，存到props或vars里，循环线程只做发送。JSR223脚本没法直接存对象到vars（vars只放字符串），但可以放到props（JMeter的全局Properties）里，用props.put("producer", producer)保存，后续脚本里props.get("producer")取出。

这里要注意跨线程问题：props是全局的，多个线程都会访问同一个Producer对象。如果每个线程都要单独一个Producer，可以用Thread.currentThread().getName()作为变量名去存，避免线程间竞争。

要不要加flush()？

send()是异步的，数据会进Producer内部的缓冲区，如果脚本立刻结束而没有flush，可能有一小部分数据没来得及发出就随线程销毁了。单条发送时加一下flush()没问题，但批量压测时每一条都flush会严重降低吞吐量。我的建议是：压测时关闭autocommit提升性能，在线程组结束（teardown）时统一flush一次。

序列化器选择

如果只发字符串，用StringSerializer就可以了。如果消息体是JSON，可以先把对象用Gson或Jackson转成字符串再发。没有必要为JSON写自定义Serializer，除非你特别在意性能。另外key序列化器用StringSerializer，用userId作为key，可以保证相同userId进入同一分区。Kafka的消息分区是基于key的hash的，如果你希望所有消息都到同一个分区，key写死即可。

4. 方式二：Java Request采样器（工程化方案）

4.1 写一个实现JavaSamplerClient的类

当脚本逻辑越来越复杂，比如要读取CSV、拼接复杂结构体、重试、记录发送代价，塞在JSR223脚本框里就很痛苦。这时候建议写一个独立的Java类，继承AbstractJavaSamplerClient，在runTest方法里实现业务逻辑。

一个典型的类结构如下：

package com.example.jmeterkafka; import org.apache.jmeter.config.Arguments; import org.apache.jmeter.protocol.java.sampler.AbstractJavaSamplerClient; import org.apache.jmeter.protocol.java.sampler.JavaSamplerContext; import org.apache.jmeter.samplers.SampleResult; import org.apache.kafka.clients.producer.KafkaProducer; import org.apache.kafka.clients.producer.ProducerRecord; import java.util.Properties; public class KafkaProducerSampler extends AbstractJavaSamplerClient { private KafkaProducer<String, String> producer; @Override public Arguments getDefaultParameters() { Arguments args = new Arguments(); args.addArgument("BOOTSTRAP_SERVERS", "localhost:9092"); args.addArgument("TOPIC", "test"); args.addArgument("SASL_USERNAME", ""); args.addArgument("SASL_PASSWORD", ""); return args; } @Override public void setupTest(JavaSamplerContext context) { // 初始化Producer Properties props = new Properties(); props.put("bootstrap.servers", context.getParameter("BOOTSTRAP_SERVERS")); props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer"); props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer"); props.put("security.protocol", "SASL_PLAINTEXT"); props.put("sasl.mechanism", "PLAIN"); props.put("sasl.jaas.config", "org.apache.kafka.common.security.plain.PlainLoginModule required username=\"" + context.getParameter("SASL_USERNAME") + "\" password=\"" + context.getParameter("SASL_PASSWORD") + "\";"); producer = new KafkaProducer<>(props); } @Override public SampleResult runTest(JavaSamplerContext context) { SampleResult result = new SampleResult(); result.sampleStart(); try { String topic = context.getParameter("TOPIC"); String userId = context.getParameter("userId"); String orderId = context.getParameter("orderId"); String value = String.format("{\"userId\":\"%s\",\"orderId\":\"%s\"}", userId, orderId); producer.send(new ProducerRecord<>(topic, userId, value)).get(); result.setSuccessful(true); result.setResponseData("OK", "UTF-8"); } catch (Exception e) { result.setSuccessful(false); result.setResponseData(e.toString(), "UTF-8"); } finally { result.sampleEnd(); } return result; } @Override public void teardownTest(JavaSamplerContext context) { if (producer != null) { producer.close(); } } }

4.2 如何编译打包并让JMeter加载

推荐的构建方式是Maven或Gradle。有一个小坑：编译时需要把JMeter的ApacheJMeter_core和ApacheJMeter_java加到provided scope，但运行时不打包进去，因为JMeter自己会提供。你需要打成一个不包含依赖的普通jar，Kafka客户端依赖则放到JMeter的lib目录。如果你用maven-shade插件把Kafka依赖也打进去，反而容易和JMeter已有的jar冲突。

打包好后的jar文件放到JMETER_HOME/lib/ext目录下，重启JMeter，然后添加“Java请求”采样器，类名下拉框中就能看到这个类。

4.3 两种方式的对比

如果你是一个长期维护的压测平台，建议直接走Java Request方案；如果你只是应急用一次，JSR223脚本就够了。我这次是两者都写了：先用JSR223验证认证和topic连接，再把它改造成Java类放进工程，方便后续团队其他人使用。

5. 第一次实战遇到的坑和完整排查链路

5.1 认证报错：LoginModule没找到

第一次运行，JMeter日志直接抛了一大片异常：

ERROR o.a.j.p.j.s.JSR223Sampler: Problem processing JSR223 script javax.security.auth.login.LoginException: Could not login: the user 'testuser' does not have permission to login

第一反应是账号密码写错了，但仔细核对没有错。继续翻异常堆栈，看到Unable to load SASL mechanism之类的内容。后来才意识到，虽然props里设置了security.protocol和sasl.mechanism，但Kafka客户端找用户名密码时靠的是JAAS。如果通过sasl.jaas.config动态配置，是在客户端库内部动态创建LoginModule。这里的问题是，我传的字符串是单引号包裹的，Java字符串里包含双引号，拼接时没有转义，导致PlainLoginModule收到的参数是坏的。

修正方法很简单，把sasl.jaas.config的值改成用双引号拼起来，并且注意结尾的分号不能丢：

props.put("sasl.jaas.config", "org.apache.kafka.common.security.plain.PlainLoginModule required username=\"" + username + "\" password=\"" + password + "\";");

如果你在Groovy里写，也一样。还有个小技巧：如果用户名或密码带有特殊字符，比如分号或引号，最好对参数做base64编码后再塞进去，否则很容易破坏JAAS字符串语法。我当时特意让密码里没有特殊字符，才避开了这个坑。

5.2 依赖冲突：NoSuchMethodError

解决了认证问题之后，发送时又报NoSuchMethodError: org.apache.kafka.common.security.auth.SecurityProtocol之类的错。这大概率是JMeter的lib目录下有多个版本的Kafka依赖。比如有些插件会自带一个旧版kafka-clients，而我换上的新版客户端又没完全覆盖旧类。JMeter的类加载机制是经典的父类优先委托，先加载lib下的旧版本就不会再看新版本。

排查过程：

1. 先把JMETER_HOME/lib下所有带kafka关键字的jar列出来。
2. 用jar tf检查每个jar里的SecurityProtocol.class路径。
3. 如果发现旧版本，如kafka-clients-0.11.0.0.jar，移除它，保留新版本。
4. 重启JMeter，再次验证。

如果你不确定哪些jar是其他插件必须的，不要直接删，可以新建一个目录lib/ext2（不存在的目录），然后用JMeter启动参数预留类路径？其实最稳妥的方法是把你的实验环境独立出来，用干净JMeter + 自己的jar组合。我当时是专门在一台压测机上装了一个干净的JMeter，只放自己必要的jar，避开了各种历史遗留问题。

5.3 数据没带分区：乱序全堆到一个分区

代码跑通了，数据也发过去了，但消费端反馈说顺序不对。原因是我把batch.size调得很大，而key直接用了userId的字符串。Kafka分区计算是murmur2哈希，理论上相同key会进同一分区，但如果我的key是空字符串，全去随机分区了。检查代码时发现，从JMeter变量里拿userId时，变量名写错了，拿到了空值，于是所有消息都用空key发送。修正后强制要求每次发送前校验key非空。

5.4 高并发下性能上不去：瓶颈不是Producer而是脚本上下文

在压测时，单线程每秒只能发两百条左右，怎么调都上不去。后来发现瓶颈其实在JSR223脚本每次执行时解析Groovy代码，以及到处写日志。优化方式：

• 把Producer初始化和变量读全部挪到setUp脚本里，用props存储。
• 循环内脚本只做发送动作，不再重复创建对象。
• 关闭采样器的Enable Logging，避免打印每一条消息。
• 在JMeter线程组里把“一次循环执行多个取样器”变成只保留一个取样器，把多条消息合并到一个循环里发送，减少上下文切换。

优化后单线程吞吐提升到了每秒三千条左右，算是在压测可接受范围内。

6. 从单条脚本到压测场景：参数化、批量与监控

6.1 用CSV参数化模拟真实数据

压测不能老是发同样的消息，需要从外部文件读取参数。JMeter自带的CSV 数据集配置非常适合。你可以提前生成一个订单文件，三列：userId、orderId、amount。然后在线程组里引用${userId}等变量。这样JSR223脚本里的vars.get("userId")就能拿到当前线程的值。

还有个容易踩的坑：CSV配置中“线程共享模式”默认是所有线程。如果你希望每个线程都从第一行开始，或者错开读取，需要根据场景选择。像这种Kafka压测一般用“每个线程有自己的阅读器”，不然所有线程会读到同一行，数据重复率太高。

6.2 批量发送与异步回调

如果觉得每条消息都发一次太慢，可以开启批量模式。KafkaProducer本身自带linger.ms和batch.size，就算你只调send()，它也会自动批量传输。但如果你在send后立刻执行get()，那就和同步一样失去了批量优势。正确的压测写法是：

for (int i = 0; i < 10; i++) { producer.send(new ProducerRecord<>(topic, key, msg)) } producer.flush()

这样可以在一个循环里攒一批，最后flush一次。JMeter的循环控制器里可以控制每次发送几条。如果你需要知道每条消息是否都成功，不要用future.get()阻塞，而是注册一个回调：

producer.send(record, new Callback() { void onCompletion(RecordMetadata metadata, Exception exception) { if (exception != null) { // 记录失败 } } });

在JMeter脚本里，用回调可能不太好记录采样结果，因为脚本本身不知道发送成功与否。最简单的做法是批量发送后用get(5, TimeUnit.SECONDS)检查这批中的几条关键结果，不用全部检查。

6.3 观察消费端与监控

数据发完后，怎么确认都到了？一种方法是在压测脚本的teardown阶段查询Kafka的GetOffsetShell，统计topic的endOffset，对比发送总量。另一种更直接的方式，在JSR223脚本里用AdminClient的describeTopics和listOffsets，把结果写到日志或文件里。

我这里分享一个简单做法：在压测结束前写一个额外的JSR223脚本调用kafka.admin.AdminClient，获取partitions的offset，并打印到报告中。这个方法不用额外写Java类，只是脚本代码多一些。

import org.apache.kafka.clients.admin.AdminClient import org.apache.kafka.clients.admin.AdminClientConfig import org.apache.kafka.clients.admin.TopicDescription import org.apache.kafka.common.TopicPartitionInfo import java.util.concurrent.TimeUnit def adminProps = new Properties() adminProps.put(AdminClientConfig.BOOTSTRAP_SERVERS_CONFIG, vars.get("kafka_servers")) adminProps.put("security.protocol", "SASL_PLAINTEXT") adminProps.put("sasl.mechanism", "PLAIN") adminProps.put("sasl.jaas.config", "...") def admin = AdminClient.create(adminProps) def topics = admin.describeTopics([vars.get("kafka_topic")]).values() topics.get(vars.get("kafka_topic")).get().partitions().each { tp -> def offsets = admin.listOffsets( Collections.singletonMap(new org.apache.kafka.common.TopicPartition(vars.get("kafka_topic"), tp.partition()), org.apache.kafka.clients.admin.OffsetSpec.latest()) ).all().get() println("partition ${tp.partition()} latest offset: ${offsets.values()[0].offset()}") } admin.close()

注意这段只是验证角度的扩展，不需要每条都执行，放一个单独的取样器里跑一次就行。

6.4 压测过程中JMeter自身的内存与GC

如果压测线程多、发送频率高，JMeter堆内存要配置够。默认HEAP=-Xms512m -Xmx512m肯定不够，我一般改成-Xms2g -Xmx2g。如果脚本里大量创建字符串和Producer对象，内存会涨得很快。建议用JConsole连到JMeter进程观察老年代使用率，避免触发频繁GC导致TPS抖动。

另外，试跑时在jmeter.log里经常能看到“Could not connect to node”的警告，如果测试还能继续跑，可以忽略；但如果连接失败次数持续上升，就要检查网络和Broker负载了。

7. 经验之外的一点补充：SASL其他机制和跨环境配置

7.1 如果换成了SCRAM或GSSAPI

刚刚主要写了PLAIN机制，这是最简单也最常见的内网测试配置。但生产环境可能用SCRAM-SHA-256或Kerberos。SCRAM只需要改两处：

props.put("sasl.mechanism", "SCRAM-SHA-256"); props.put("sasl.jaas.config", "org.apache.kafka.common.security.scram.ScramLoginModule required username=\"user\" password=\"pass\";");

GSSAPI（Kerberos）就复杂得多，通常需要java.security.auth.login.config指向kafka_client_jaas.conf，还要配置java.security.krb5.conf，并且每个执行机都要有有效的TGT。如果不是特别必要，压测环境尽量别用Kerberos。

7.2 多环境配置管理

实际项目中，开发环境、测试环境、压测环境的Kafka地址和账号可能都不一样。不要把这些硬编码在脚本里。我在JMeter里用测试计划的“用户定义的变量”统一维护，脚本只读vars.get()。如果是Java Request模式，这些就是采样器参数面板里的字段。后续还可以用JMeter的-J参数从命令行覆盖变量，比如：

jmeter -n -t test.jmx -Jkafka_servers=192.168.x.x:9093 -Jkafka_user=load_test -Jkafka_pass=xxxx

这样工程文件里不会留下任何明文密码，也算规避了凭据泄露风险。

7.3 脚本里不要暴露敏感信息

你在JSR223脚本或者Java类里写了明文密码，jmx文件里就会体现。除非是纯内网临时测试，否则建议至少对密码做变量引用，而真正值通过命令行或环境变量注入。我这次因为测试环境的账号密码本来就是要统一暴露给压测执行机，所以没做太多花活，但后来考虑到可能有人从jmx文件里翻到信息，还是改成了环境变量方式。

压测和中间件打交道就是这样，往往不是功能难，而是环境配置和版本兼容折磨人。尤其是JMeter+Kafka+SASL这个组合，网上能搜到的资料不少，但大多只贴一个片段，缺少完整细节。希望这篇能把第一次走的弯路帮你提前踩平。后面我再遇到类似的需求，大概率会直接复用这套Java Request工程模板，再把配置项抽成外部文件，省下重复排查的时间。