企业官网建设流程全解析

先说结论：JMeter官方没有自带Kafka的Sampler，网上搜到的第三方插件看着方便，但一旦牵扯上SASL认证，真的要命。我这次要在压测环境里往Kafka灌数据，而且集群开了SASL_PLAINTEXT + PLAIN机制的用户名密码认证，折腾了整整三四天。期间试过第三方插件、JSR223脚本，最后老老实实写了一个Java Sampler，传参、认证、跑并发全搞定。

这篇把整个流程从零讲一遍，包括为什么选Java Sampler、依赖jar怎么准备、SASL认证到底卡在哪、代码怎么写、实际调试时遇到哪些坑。看完你应该能照着写一个自己的版本，不用再像我一样翻十几篇博客还凑不齐一个能跑的Demo。

1. 为什么绕了一圈，最后还是写Java Sampler

1.1 先说这个需求的真实场景

Kafka在压测链路里经常扮演“数据入口”的角色：上游系统往某个Topic里灌消息，后端的消费者再拉取处理。做压测或者联调的时候，我们需要模拟大量消息源源不断写入Kafka。这时候JMeter作为压测工具，就需要一个能发Kafka消息的Sampler。

难点在于，现在的Kafka集群基本不会裸奔，至少要开SASL认证。像我们这边的测试环境就是SASL_PLAINTEXT协议，走PLAIN机制，客户端要提供用户名密码才能连上去。如果只是往Kafka塞数据，最省事的方式是直接用命令行工具kafka-console-producer.sh，但那是脚本，不是压测工具，没法在JMeter里生成并发TPS，更没法跟JMeter的监听器结合统计响应时间。

1.2 网上流传的方案，我一个个排掉

先说第三方插件。JMeter的插件市场里确实有Kafka相关的Sampler，比如某稳定版插件，装上之后填bootstrap.servers、topic、payload就能发。但问题也很明显：插件对SASL的支持参差不齐，有些版本直接没做sasl.jaas.config的透传，有些默认只支持GSSAPI（Kerberos），想用PLAIN用户名密码根本没地方填。即便能填，插件本身的发包逻辑是封装死的，出问题没法自定义，只能干瞪眼。

然后是JSR223 + Groovy脚本。JSR223在JMeter里确实灵活，能直接写Java代码访问Kafka客户端API。但有个很现实的坑：如果每个线程每次迭代都new一个KafkaProducer，连接开销和元数据拉取会把性能拖垮。你可以把Producer定义成共享变量，但脚本作用域、线程安全、异常处理写起来都很别扭，而且Groovy脚本在高压并发下还存在JIT编译不稳定、排查困难的问题。

最后就是自定义Java请求（Java Sampler）。JMeter早就预留了扩展点，实现一个简单的类继承AbstractJavaSamplerClient，JMeter就会在它的线程组里调用你的方法。最爽的一点是生命周期可控：testStarted的时候创建Producer，runTest只负责发消息，testEnded负责关闭连接。这样Producer在整个压测期间复用，既省连接开销又灵活。

方案对比总结如下：

所以，写Java Sampler不是炫技，是灰姑娘试鞋，最后发现只有这双能穿上。

1.3 这么选的好处是什么

核心收益有两个。第一是可控：Producer怎么建、消息怎么发、失败怎么处理，代码全在自己手里。第二是可复现：jar包放到JMeter的lib/ext之后，任何环境只要配置好参数就能跑，不用每次CTRL+C、CTRL+V一堆脚本。特别是有SASL认证的环境，配置项直接暴露成Sampler参数，换环境只需要改JMeter里的参数值，不用重新写代码。

2. 动手之前，这些地基必须打好

2.1 版本选择：Kafka客户端和JMeter/JDK别乱配

写代码前先确认环境版本，不然编译出来跑不起来，白干。

JMeter建议用5.4以上版本，JDK用8或11都行。重点说Kafka客户端版本。Kafka客户端的兼容性逻辑是：客户端版本最好和服务端大版本保持一致，至少不要差的离谱。比如服务端是Kafka 2.8，客户端用2.8.x或者3.x一般都能连，但如果服务端是Kafka 3.x，客户端还在用老掉牙的0.10版本，加密协议、SASL机制都不一定匹配，很可能握手都过不去。

我用的是kafka-clients的2.8.2版本，足够覆盖常见的SASL_PLAINTEXT/PLAIN场景。没有特殊需求别追新，稳定优先。

提示：JDK版本别太老。kafka-clients3.x要求Java 8起步，部分新版本需要Java 11。如果你JMeter运行在JDK 8上，建议客户端版本卡在2.8.x附近，否则可能直接报UnsupportedClassVersionError。

2.2 依赖jar去哪找

写Java Sampler就得引Kafka客户端的库。一个最省事的办法：去Apache Kafka安装包里翻，里面自带lib目录，有一大堆jar。我们需要的是这些：

• kafka-clients-x.x.x.jar（核心）
• slf4j-api-x.x.jar（Kafka客户端打日志用）
• slf4j-simple-x.x.jar（日志简单绑定，调试方便）
• snappy-java、lz4-java、zstd-jni（消息压缩相关）
• （如果用到Kafka Connect之类的则不需要）

如果你用Maven/Gradle管理代码，更推荐直接在pom.xml里加上依赖，然后打包的时候用一个带依赖的插件（比如maven-assembly-plugin）打成一个胖jar。这样JMeter的lib/ext下只需要放一个jar，省去手动凑依赖的麻烦。

我当时的做法是：用Maven的assembly插件打成xxx-jar-with-dependencies.jar，扔到JMETER_HOME/lib/ext目录下，重启JMeter，搞定。

注意：jar放到lib/ext之后，必须重启JMeter才生效。刚开始我不知道，放进去就点“运行”，一直ClassNotFoundException，还以为是代码写错了，后来才发现是没重启。

2.3 SASL认证，一分钟快速理解

SASL全称是Simple Authentication and Security Layer，翻译过来是简单认证与安全层。Kafka用它做客户端认证，常见的机制有PLAIN（用户名+密码，明文传输）、SCRAM-SHA-256/512（带加盐和哈希的密码认证，更安全）、GSSAPI（基于Kerberos，一般企业内网用）。

PLAIN最少需要三样东西：

• security.protocol：指定传输层协议，比如SASL_PLAINTEXT表示明文传输+认证；SASL_SSL表示TLS加密传输+认证。
• sasl.mechanism：指定认证机制，这里就是PLAIN。
• sasl.jaas.config：这是Java认证授权服务（JAAS）风格的配置串，里面写死LoginModule和账号密码。

如果你之前没接触过JAAS，它其实就是一个Java标准里的“登录配置”。注意理解这两步：PLAIN认证需要把配置丢给Kafka客户端的LoginModule去加载，然后客户端带着这些凭证去Kafka broker握手认证。把这三个参数给对，认证就通了一半。

3. 写一个能跑的Java Sampler，代码拆开讲

3.1 类的骨架和生命周期

Java Sampler的规范是：继承AbstractJavaSamplerClient，实现几个核心方法即可。JMeter会按以下生命周期调用你：

• getDefaultParameters()：告诉JMeter这个Sampler有哪些参数，参数名、默认值都会显示在JMeter界面里。
• testStarted()：测试计划开始运行时调用，适合在这里创建Producer连接。
• runTest()：每次sample执行时调用，相当于JMeter里一个“请求”发生。返回SampleResult，JMeter通过这个对象判断成功与否、记录响应时间。
• testEnded()：测试计划结束时调用，在这里关闭Producer。

理解生命周期再写代码就很清楚：不要把Producer的创建放在runTest()里，那是性能灾难。也不要放在静态块里，JMeter在GUI模式下预览时也可能触发初始化，导致还没开始压测就连了一把Kafka。

3.2 完整代码示例（可复制）

下面是一个完整可用的类，我稍作简化，让它专注于发单条消息。你复制到自己的工程里，保证依赖没问题就能编译。

package com.example.jmeter.kafka; 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.Producer; import org.apache.kafka.clients.producer.ProducerRecord; import org.apache.kafka.common.serialization.StringSerializer; import java.util.Properties; public class KafkaSaslSampler extends AbstractJavaSamplerClient { private Producer<String, String> producer; private String topic; @Override public Arguments getDefaultParameters() { Arguments arguments = new Arguments(); arguments.addArgument("bootstrap.servers", "kafka1:9092,kafka2:9092"); arguments.addArgument("topic", "test-topic"); arguments.addArgument("message", "hello from jmeter"); arguments.addArgument("security.protocol", "SASL_PLAINTEXT"); arguments.addArgument("sasl.mechanism", "PLAIN"); arguments.addArgument("sasl.jaas.config", "org.apache.kafka.common.security.plain.PlainLoginModule required username=\"jmeter_user\" password=\"Jmeter@123\";"); return arguments; } @Override public void testStarted(JavaSamplerContext context) { Properties props = new Properties(); props.put("bootstrap.servers", context.getParameter("bootstrap.servers")); props.put("key.serializer", StringSerializer.class.getName()); props.put("value.serializer", StringSerializer.class.getName()); props.put("security.protocol", context.getParameter("security.protocol")); props.put("sasl.mechanism", context.getParameter("sasl.mechanism")); props.put("sasl.jaas.config", context.getParameter("sasl.jaas.config")); // 这两个参数是调吞吐用的，后面说明 props.put("linger.ms", 5); props.put("batch.size", 65536); producer = new KafkaProducer<>(props); topic = context.getParameter("topic"); } @Override public SampleResult runTest(JavaSamplerContext context) { SampleResult result = new SampleResult(); result.setSampleLabel("KafkaSASLProducer"); result.sampleStart(); try { String message = context.getParameter("message"); ProducerRecord<String, String> record = new ProducerRecord<>(topic, message); // send().get() 是同步发送，保证这条消息确实发成功了再返回 producer.send(record).get(); result.setSuccessful(true); result.setResponseMessage("OK"); result.setResponseData(("Sent: " + message).getBytes()); } catch (Exception e) { result.setSuccessful(false); result.setResponseMessage(e.getMessage()); result.setResponseData(("Error: " + e).getBytes()); } finally { result.sampleEnd(); } return result; } @Override public void testEnded(JavaSamplerContext context) { if (producer != null) { // flush确保缓冲里没发完的消息全部写出 producer.flush(); producer.close(); } } }

3.3 代码里几个值得说明的细节

第一个是StringSerializer.class.getName()。Kafka客户端配置里，key/value序列化器需要的是类名，不是Class对象，所以用getName()拼接字符串。有人喜欢写"org.apache.kafka.common.serialization.StringSerializer"，效果一样，但用getName()能避免手写拼错。

第二个是producer.send(record).get()。这是同步发送，意思是每条消息都等Kafka返回确认才结束sample。好处是响应时间真实，负载高时能直观看到发送耗时；坏处是吞吐上限受限于单条RTT（往返时延）。如果追求更高吞吐，可以改成异步发送加回调，后面专门聊。

第三个是linger.ms和batch.size。KafkaProducer默认会把小消息攒一批再发，linger.ms控制在缓冲区最多等多久，batch.size控制批次大小。这两个参数调大一些，压测时的TPS会比一条条同步发高很多，前提是你接受了“等一会才发出去”的延迟。这个延迟在测试数据灌入场景下通常可接受。

4. SASL认证：成也配置，败也配置

4.1 JAAS的两种提供方式，用哪个心里要有数

Kafka的SASL认证最终要落到JAAS配置上。常见有两种姿势：

第一种是JVM启动参数：-Djava.security.auth.login.config=/path/jaas.conf，jaas.conf文件内容大致如下：

KafkaClient { org.apache.kafka.common.security.plain.PlainLoginModule required username="jmeter_user" password="Jmeter@123"; };

这种方式的缺点是全局生效，一个JVM只能加载一个JAAS文件。如果你在同一个JMeter实例里同时要连多个不同账号的Kafka集群，这就不够用了。

第二种是在客户端Properties里直接设置sasl.jaas.config，例如代码里的：

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

Kafka客户端在较新版本已经支持直接从properties里读JAAS配置，完全不用动JVM参数。这个方式最大的好处是每个Sampler可以带不同的账号密码，互不干扰。我强烈推荐用第二种，因为它灵活，而且JMeter的Java Sampler可以把这个值作为参数在界面上配置，换环境不用重新编译。

4.2 在JMeter界面里参数到底怎么填

代码里getDefaultParameters()定义的那些参数，在你添加Java请求之后会出现在JMeter右侧的参数表里，每个参数都可以在GUI里覆盖默认值。实操时注意下面几个点。

SASL_PLAINTEXT环境的核心参数照着填：

特别注意sasl.jaas.config的值里有分号;和引号"。JMeter参数值本质上就是个字符串，直接填进去没问题，但千万别用${__property(...)}这类函数去拼接，容易把分号吃了解析失败。

我踩过一个具体坑：最开始我把sasl.jaas.config写成了Java字符串转义格式，也就是在双引号前加了一堆反斜杠，结果Kafka客户端拿到的是带反斜杠的配置串，一直报LoginModule找不到，后来去掉反斜杠就好了。JMeter界面里填的、和代码里直接props.put()的内容，都应该是Kafka客户端真正想要的那个原始字符串，不要再做一层转义。

4.3 认证失败时，怎么顺着报错找问题

SASL认证一旦失败，报错会出现在JMeter的SampleResult里，或者jmeter.log里。常见的几类，我一个个说。

第一类：认证直接失败

日志里的关键字一般是Authentication failed，有时候前面会跟一句Exception in thread "main" ...。核心信息多半是Login failed。这种情况优先检查用户名和密码，尤其是密码里如果带了&、@、空格这类特殊字符，在sasl.jaas.config字符串里要保持原样，别傻乎乎去URL编码，Kafka不认那套。

第二类：Topic权限不足

有时候认证本身是成功的，但发送时被broker以ACL（访问控制列表）拒绝，报Topic authorization failed或者Error while fetching metadata。这时候问题不在JMeter，而在Kafka侧给账号分Topic权限。找管理员给jmeter_user加上这个Topic的Write权限。

第三类：元数据拉取卡死或者一直超时

TimeoutException通常说明客户端根本够不着broker。按优先级排查：bootstrap.servers地址通不通、本机防火墙有没有放行9092端口、security.protocol有没有写错成PLAINTEXT。有时候测试环境broker明明开了SASL_PLAINTEXT，你配置里填成了SASL_SSL，两边握手对不上，报的错也是TimeoutException。

提示：Jameter里打开调试日志能帮你少熬几个夜。在jmeter.properties或log4j2.xml里把org.apache.kafka的日志级别调到DEBUG，会直接打出Kafka客户端与服务端握手的每一个步骤，认证失败的原因一目了然。

5. 跑起来之后的现场调试实录

5.1 日志到底去哪里看

Java Sampler运行时的日志分两路：一是你自己在Sampler里设置到SampleResult的响应数据，二是JMeter自身的日志输出。很多人只盯着JMeter界面右上角的绿色运行按钮，忽略了bin/jmeter.log，其实那个文件信息量更大。

我的习惯是先把Sampler的runTest()返回体写得非常详细：异常堆栈要拼进去，SampleResult.setResponseData()里塞一段文本；然后在控制台或者jmeter.log里看Kafka客户端自己的日志。双重保险，排错速度快很多。

5.2 我实际调试踩过的几个真实大坑

坑一：ClassNotFound，Kafka客户端压根没加载

放好jar重启JMeter之后，第一次runTest直接抛ClassNotFoundException: org.apache.kafka.clients.producer.KafkaProducer。原因就是我前面说的：忘记把依赖jar放到lib/ext。这里还有个隐藏细节：如果你打的是胖jar，里面合并了META-INF/services文件，可能跟JMeter自带的冲突，所以更稳妥的方案是只把kafka-clients及它独有的依赖放lib/ext，不要带一堆乱七八糟的库。

坑二：消息“发出去”了，但对端没收到

第一次压测完，看了下SampleResult全部成功，响应时间也正常，结果Kafka那边的消费者说一条都没有。查了半天发现，是我在testEnded里producer.close()之前忘了flush()。场景是这样的：KafkaProducer的send()如果没有get()，只是把消息放到缓冲区，靠linger.ms和batch.size触发实际发送。压测结束进程还在跑时看起来没问题，但JVM一退出，缓冲区里的消息没来得及flush就丢了。

解决方式很简单：在testEnded()里先producer.flush()，再producer.close()。这也是为什么上面的代码里我特意写了flush()。或者简单粗暴一点，在需要保证不丢消息的场景下，直接send().get()同步等结果。

坑三：JMeter日志被Kafka的日志刷屏

Kafka客户端依赖SLF4J，而JMeter自己是log4j2体系，两者如果绑定不对，启动之后日志会莫名其妙丢失或者打不出Kafka的DEBUG。这是典型的日志门面冲突。解决办法是统一用log4j-slf4j-impl这个桥接包，让SLF4J的调用落到log4j2上。具体做法：去JMeter的lib目录里看有没有log4j-slf4j-impl，如果有，就不要再往lib/ext里放slf4j-simple，否则会冲突。

5.3 一次典型的“第一次跑通”记录

我简单复盘一下当时的过程，给后来人一个心理预期。

第一步，在IDEA里建一个普通的Maven工程，只依赖kafka-clients，把上面的代码抄进去，编译通过。

第二步，用maven-assembly-plugin打出一个带全部依赖的jar，复制到JMETER_HOME/lib/ext/，重启JMeter。

第三步，在JMeter里新建一个线程组（先只开1个线程、循环1次），添加Java请求，选择刚写的Sampler类。JMeter会自动识别并填充参数，我在界面里覆盖了Kafka集群地址、账号密码、topic。

第四步，运行测试，查看SampleResult。第一次的结果是失败，报了个sasl.jaas.config的解析错误，我调整了引号转义之后再次运行，成功。

第五步，把线程数调到10、循环次数调到1000，真跑了一轮压测，通过Kafka的命令行消费者验证消息数量，确认topic收到了1万条。

整个过程如果没踩坑，半小时能跑通。踩了坑就是几天的命。

6. 进阶玩法：如何让消息内容动态化，让Producer扛住并发

6.1 消息内容用JMeter变量动态生成

压测场景下每条消息都一模一样没有意义，需要动态生成内容。两个常用办法：

一个是在JMeter的“用户自定义变量”或者“CSV数据文件”里准备一组数据，然后在Java Sampler的message参数值里引用变量，比如：

message = ${__time(yyyy-MM-dd HH:mm:ss,)}-${__threadNum}-${__Random(1000,9999,)}

Java Sampler的context.getParameter("message")取出来的时候，JMeter已经完成了变量替换，所以你代码里不需要特殊处理，直接拿到的是替换后的字符串。

另一个办法是在代码里自己造数据。比如根据JMeterContextService.getContext()拿到线程号和循环次数拼接字符串，或者干脆在runTest()里生成随机UUID。这样就不依赖JMeter的变量替换机制，处理复杂格式时更自由。

我当时用的是CSV文件方式，因为需要贴近业务字段的真实分布，后面还接了JSR223预处理来造JSON格式消息。注意，Java Sampler里runTest()拿到的参数是在每次执行时重新取值的，所以JMeter变量能够实时变化。

6.2 并发场景下，Producer要不要每个线程新建一份

不需要。KafkaProducer官方说明是线程安全的，多个线程可以共享同一个实例。上面的代码在testStarted()里只new了一次Producer，JMeter线程组里不管开多少线程，runTest()都是拿同一个Producer去send。这是正确姿势，既节省了连接资源，也避免每次迭代重复创建Producer的开销。

但要注意线程安全的分界线：producer.send()本身是线程安全的，但你自己写的代码要小心共享变量。比如不要在多个线程里同时改同一个topic字段再拿去send，虽然ProducerRecord是每个请求内new的，问题不大，但养成局部变量的习惯更安全。

6.3 吞吐上不去的几个优化点

如果发现TPS卡在某个值上不去，先检查是不是同步发送。send().get()在局域网内单次RTT可能只要几毫秒，但1万并发就是1万次等待，吞吐自然受限。改成下面的异步发送加回调：

producer.send(record, (metadata, exception) -> { if (exception != null) { // 记录发送失败的消息 } });

这样runTest()会在消息进入缓冲区后立即返回，JMeter统计到的响应时间其实变成了“入队时间”，TPS会高很多。但代价是响应时间不再代表真实发送延迟，对结果解读有影响，压测时心里要有数。

还有几个Kafka客户端参数对吞吐影响很大：

• linger.ms：默认0，即来一条发一条。调大一点（比如5~20ms）可以聚合更多消息，减少网络往返，但会增加端到端延迟。
• batch.size：默认16KB，调大到64KB或128KB，单个批次能装更多消息。
• buffer.memory：Producer缓冲区总大小，默认32MB，高并发下如果消息量大于处理速度，可能报BufferExhaustedException，这时需要调大。
• compression.type：设为snappy或lz4，能显著降低带宽占用，CPU换带宽，局域网里收益一般，跨机房传输收益大。

我实际压过一组对比：同步send+默认参数，单线程TPS大概在几千；改成linger.ms=5、batch.size=131072，同样的机器能到两三万，并发线程加到20后还能往上走。不过别只看数字，关键还是被测系统对消息延迟是否敏感，压测前想清楚你的目标。

6.4 如果Kafka开了SASL_SSL，要加什么

SASL_SSL比SASL_PLAINTEXT多了一层TLS加密传输，配置也多点。除了前面几个SASL参数，还要在Properties里加：

props.put("ssl.truststore.location", "/path/to/truststore.jks"); props.put("ssl.truststore.password", "changeit");

truststore文件可以从Kafka集群的管理员那里拿，或者从broker的证书导出来。这里有个常见误区：SASL_SSL与SSL认证不同，它只是传输加密，客户端身份认证还是靠SASL。所以用户名密码那套配置一点都不能少。

7. 常见问题速查表，总有一个你会碰上

下面这张表，是我在写这个Sampler和后续维护过程中总结的高频问题，建议收藏：

排查顺序建议这样走：先看jmeter.log有没有Kafka客户端的异常堆栈，没有再看SampleResult的响应信息；这两个信息定位不了，就把org.apache.kafka的日志级别调到DEBUG，看握手过程。

还有一个小经验：写这个Sampler的时候，不要把sasl.jaas.config写死在代码里。把它留在getDefaultParameters()里，测试环境的账号密码本身就是敏感信息，有人会从外部配置文件读取，你可以做成JMeter系统属性，再在参数默认值里用${__P(sasl.jaas.config,)}读取，这样代码仓库里就不会出现明文密码。

我实际用下来的体会是，Java Sampler第一次写确实有门槛，但一旦跑通了，后面再改业务字段、调认证方式、换集群，都只是改参数的事。KafkaClient官方本身就是一个稳定的库，大部分时间不需要跟它较劲，要较劲的往往是配置细节和部署环境的集成。遇到问题别慌，按着日志一层层剥，总能找到根因。

最后分享一个不相关但很实用的技巧：如果压测过程中需要频繁改动Kafka连接参数，完全可以在JMeter里同时配置多个Java Sampler副本，每个副本连不同的Kafka集群，然后用JSR223或切换控制器动态选择。这样在联调环境之间切换，连代码都不用动。