JMeter脚本复用KafkaProducer对象时遭遇ClassCastException问题求助

问题根源分析 #

你遇到的ClassCastException本质是线程变量vars的存储范围不匹配，或是初始化逻辑未正确执行：

• vars是线程局部变量，仅在当前线程内有效。如果初始化采样器和发送采样器分属不同线程（比如Setup线程组和主线程组），vars无法跨线程传递对象。
• 若初始化采样器执行失败（比如缺少Kafka依赖包、变量未定义），producer变量会保留默认空字符串值，导致后续转换失败。

解决方案 #

1. 调整Producer初始化与存储方式 #

将Producer初始化放到Setup Thread Group（测试前仅执行一次），并通过全局属性props存储对象，确保所有线程都能复用同一个Producer。

步骤1：添加Setup Thread Group（线程数设为1）

在该组内创建JSR223Sampler，脚本如下：

import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.common.serialization.StringSerializer;
import java.util.Properties;

// 构建Producer配置
Properties propsConfig = new Properties();
propsConfig.put("bootstrap.servers", vars.get("bootstrapServer")); // 确保已定义bootstrapServer变量
propsConfig.put("transactional.id", "my-transactional-id");
propsConfig.put("acks", "all"); // 根据测试需求添加必要配置

// 初始化Producer并存储到全局属性
KafkaProducer<String, String> producer = new KafkaProducer<>(propsConfig, new StringSerializer(), new StringSerializer());
props.putObject("globalProducer", producer);
log.info("全局Kafka Producer初始化完成");

步骤2：主线程组的发送采样器脚本

import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.common.header.RecordHeader;

// 从全局属性获取Producer
KafkaProducer<String, String> producer = props.getObject("globalProducer") as KafkaProducer<String, String>;

// 构造消息体
String payload = "{\"data\":\"{\\\"assetSearchCriteria\\\":[{\\\"attribute\\\":\\\"category\\\",\\\"operator\\\":\\\"EQ\\\",\\\"value\\\":\\\"SYSTEM\\\"}]}\"}";
ProducerRecord<String, String> record = new ProducerRecord<>("REPO-Asset-Partner-QueryRequest", "", payload);

// 添加请求头（确保所有变量已定义）
record.headers().add(new RecordHeader("authorization", vars.get("bear").getBytes()));
record.headers().add(new RecordHeader("ce_id", vars.get("searchIndex").getBytes()));
record.headers().add(new RecordHeader("ce_type", "com.fico.repo.ms.asset.events.message.partner.query.AssetSearch".getBytes()));
record.headers().add(new RecordHeader("ce_time", vars.get("time").getBytes()));
record.headers().add(new RecordHeader("requestId", vars.get("searchCorelation").getBytes()));

// 异步发送消息（可选添加回调日志）
producer.send(record, (metadata, exception) -> {
    if (exception != null) {
        log.error("消息发送失败", exception);
    } else {
        log.info("消息已发送到Topic: {}，Partition: {}，Offset: {}", metadata.topic(), metadata.partition(), metadata.offset());
    }
});

步骤3：添加Teardown Thread Group（线程数设为1）

用于测试结束时关闭Producer，避免资源泄漏：

import org.apache.kafka.clients.producer.KafkaProducer;

KafkaProducer<String, String> producer = props.getObject("globalProducer") as KafkaProducer<String, String>;
if (producer != null) {
    producer.flush(); // 确保所有消息发送完成
    producer.close();
    log.info("全局Kafka Producer已关闭");
}

2. 关键注意事项 #

• 依赖包检查：确保JMeter的lib目录下已放入kafka-clients.jar及相关依赖包（如kafka-common.jar），版本需与代码兼容。
• 变量有效性：确认bootstrapServer、bear、searchIndex等变量已在User Defined Variables或其他配置元件中正确定义。
• 禁止中途关闭Producer：发送采样器中不要调用producer.close()，否则首次发送后Producer会被销毁，后续请求无法复用。

3. 同线程组内复用的替代方案 #

如果需要每个线程拥有独立的Producer，可在主线程组的初始化采样器中勾选Execute once only（仅执行一次），并继续使用vars存储：

// 初始化采样器脚本（勾选Execute once only）
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.common.serialization.StringSerializer;
import java.util.Properties;

Properties propsConfig = new Properties();
propsConfig.put("bootstrap.servers", vars.get("bootstrapServer"));
propsConfig.put("transactional.id", "thread-" + ctx.getThreadNum() + "-txn"); // 每个线程用唯一事务ID
KafkaProducer<String, String> producer = new KafkaProducer<>(propsConfig, new StringSerializer(), new StringSerializer());
vars.putObject("threadProducer", producer);

发送采样器中获取：

KafkaProducer<String, String> producer = vars.getObject("threadProducer") as KafkaProducer<String, String>;
// 后续发送逻辑同前

内容的提问来源于stack exchange，提问作者user3222327