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.jarslf4j-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认证靠JAASJava Authentication and Authorization Service提供用户凭据。JMeter本身不管理Kafka的认证配置所以必须显式告诉Kafka客户端用户名和密码通常有三种方式在JVM系统属性里指定java.security.auth.login.config指向一个JAAS文件。在Producer配置里直接传入sasl.jaas.config这种方式最灵活推荐。手动设置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-512sasl.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而不是纯JavaJSR223 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脚本没法直接存对象到varsvars只放字符串但可以放到propsJMeter的全局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 KafkaProducerString, 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自己会提供。你需要打成一个不包含依赖的普通jarKafka客户端依赖则放到JMeter的lib目录。如果你用maven-shade插件把Kafka依赖也打进去反而容易和JMeter已有的jar冲突。打包好后的jar文件放到JMETER_HOME/lib/ext目录下重启JMeter然后添加“Java请求”采样器类名下拉框中就能看到这个类。4.3 两种方式的对比维度JSR223 SamplerJava Request采样器上手速度快打开JMeter就能写慢需要建工程打包代码复用脚本难复用通常就在jmx里可以做成独立库多个jmx共用调试方便不方便只能靠日志可以在IDE里写main方法调试依赖配置直接依赖lib中的jar需要额外打依赖或放lib适合场景快速原型、压测执行机灵活变动复杂逻辑、需要单测和版本管理如果你是一个长期维护的压测平台建议直接走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下的旧版本就不会再看新版本。排查过程先把JMETER_HOME/lib下所有带kafka关键字的jar列出来。用jar tf检查每个jar里的SecurityProtocol.class路径。如果发现旧版本如kafka-clients-0.11.0.0.jar移除它保留新版本。重启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\;);GSSAPIKerberos就复杂得多通常需要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_servers192.168.x.x:9093 -Jkafka_userload_test -Jkafka_passxxxx这样工程文件里不会留下任何明文密码也算规避了凭据泄露风险。7.3 脚本里不要暴露敏感信息你在JSR223脚本或者Java类里写了明文密码jmx文件里就会体现。除非是纯内网临时测试否则建议至少对密码做变量引用而真正值通过命令行或环境变量注入。我这次因为测试环境的账号密码本来就是要统一暴露给压测执行机所以没做太多花活但后来考虑到可能有人从jmx文件里翻到信息还是改成了环境变量方式。压测和中间件打交道就是这样往往不是功能难而是环境配置和版本兼容折磨人。尤其是JMeterKafkaSASL这个组合网上能搜到的资料不少但大多只贴一个片段缺少完整细节。希望这篇能把第一次走的弯路帮你提前踩平。后面我再遇到类似的需求大概率会直接复用这套Java Request工程模板再把配置项抽成外部文件省下重复排查的时间。