在进行物联网（IoT）系统性能测试时，JMeter 是第一选择工具之一。但原生的 JMeter 并不支持 MQTT 协议，因此需要借助插件或自行开发。本文将从现有插件使用、自定义插件开发、以及大规模物联网压力测试实战三个方面，提供一套完整的解决方案。

一、MQTT插件快速上手

最常用的是根据Eclipse Paho库开发的mqtt-jmeter插件。

1. 安装

从GitHub - emqx/mqtt-jmeter的Releases页面下载最新mqtt-jmeter-xxx.jar。

将jar放入$JMETER_HOME/lib/ext/ 目录。

重启JMeter，在Sampler中能看到MQTT Connect、MQTT Pub Sampler、MQTT Sub Sampler 和 MQTT DisConnect。

2. 创建基本测试场景

一个典型的设备行为是：建立连接 - 订阅主题 - 发布消息 - 断开连接。

线程组：设置线程数模拟设备数量。

MQTT Connect：

Broker URL: tcp://broker.emqx.io:1883

ClientId: 使用 JMeter 函数 ${__threadNum} 或 ${__UUID} 保证唯一。

Keep Alive: 60s，勾选 Clean Session。

MQTT Sub Sampler（可选）：订阅 test/topic，QoS 设为 1，并设置超时时间。常用作接收下行指令的证实。

MQTT Pub Sampler：主题 test/topic，QoS 1，消息体填入模拟遥测数据的 JSON。

MQTT DisConnect：释放连接。

优势：开箱即用，界面配置。

局限：无法实现复杂的连接状态机（如断线重连后的行为）、自定义负载加密、或特殊的 WebSocket 连接参数。此时就需要自定义开发。

二、自定义MQTT插件开发

开发自定义插件可实现和业务更紧密的耦合，如模拟设备影子更新、OAT升级流程等。

1. 技术选型和架构

客户端库：推荐Eclipse Paho Java（稳定、异步支持）或HiveMQ MQTT Client（性能更优，支持背压）。

开发方式：

轻量级：实现 JavaSamplerClient 接口，直接以“Java 请求”形式运行，无需编写 GUI。

专业级：继承 AbstractSamplerGui + AbstractSampler，打包成标准 JMeter 组件。

设计-连接复用：不能每个sampler都新建连接。可通过JMeterContext级别的变量存储MqttClient实例，使同一线程内的采样器共享一条TCP长连接。

2. 实战：根据JavaSamplerClient实现发布采样器

代码MqttPubClient.java：

java

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.eclipse.paho.client.mqttv3.*;

import org.eclipse.paho.client.mqttv3.persist.MemoryPersistence;

public class MqttPubClient extends AbstractJavaSamplerClient {

    // 用于线程内共享连接的 key

    private static final String CTX_CLIENT_KEY = "MQTT_CLIENT";

    private MqttClient client;

    // 定义采样器参数

    @Override

    public Arguments getDefaultParameters() {

        Arguments params = new Arguments();

        params.addArgument("broker", "tcp://127.0.0.1:1883");

        params.addArgument("clientIdPrefix", "perf_");

        params.addArgument("topic", "test/device");

        params.addArgument("qos", "1");

        params.addArgument("message", "{\"temp\":25}");

        params.addArgument("cleanSession", "true");

        return params;

    }

    @Override

    public void setupTest(JavaSamplerContext context) {

        // 从线程上下文获取已存在的连接，避免重复创建

        client = (MqttClient) context.getJMeterContext().getVariables().getObject(CTX_CLIENT_KEY);

        if (client == null || !client.isConnected()) {

            String broker = context.getParameter("broker");

            String clientId = context.getParameter("clientIdPrefix") + Thread.currentThread().getName();

            try {

                client = new MqttClient(broker, clientId, new MemoryPersistence());

                MqttConnectOptions options = new MqttConnectOptions();

                options.setCleanSession(Boolean.parseBoolean(context.getParameter("cleanSession")));

                options.setConnectionTimeout(10);

                options.setKeepAliveInterval(60);

                // 可在此加入 SSL、用户名密码等配置

                client.connect(options);

                // 将连接存入线程上下文，供后续采样器复用

                context.getJMeterContext().getVariables().putObject(CTX_CLIENT_KEY, client);

            } catch (MqttException e) {

                log.error("MQTT连接失败", e);

            }

        }

    }

    @Override

    public SampleResult runTest(JavaSamplerContext context) {

        SampleResult result = new SampleResult();

        result.sampleStart();

        try {

            String topic = context.getParameter("topic");

            int qos = Integer.parseInt(context.getParameter("qos"));

            String payload = context.getParameter("message");

            MqttMessage message = new MqttMessage(payload.getBytes());

            message.setQos(qos);

            client.publish(topic, message);

            result.setSuccessful(true);

            result.setResponseData("Published to " + topic, "UTF-8");

        } catch (MqttException e) {

            result.setSuccessful(false);

            result.setResponseData(e.toString(), "UTF-8");

        } finally {

            result.sampleEnd();

        }

        return result;

    }

    @Override

    public void teardownTest(JavaSamplerContext context) {

        if (client != null && client.isConnected()) {

            try {

                client.disconnect();

                client.close();

                context.getJMeterContext().getVariables().remove(CTX_CLIENT_KEY);

            } catch (MqttException e) { }

        }

    }

}

配置和运行：

将 Paho 等依赖库一同打入 jar，放入 lib/ext。

在 JMeter 中添加Java 请求，选择 MqttPubClient。

可结合 SetUp Thread Group 执行一次 MQTT Connect，再用普通线程组循环调用此发布采样器。

3. 扩展实现带回调的订阅采样器

订阅采样器需在runTest中使用client.subscribe() 并传入IMqttMessageListener，利用CountDownLatch等待指定数量的消息到达或超时，以此测量端到端下行延迟。

三、物联网压力测试实战：

1. 设计压测场景

物联网压力测试的重要标准包括：并发连接数、消息吞吐量（In/Out）、消息延迟（P99）、连接建立速率 以及Broker资源消耗。

典型场景分解：

设备上线风暴：短时间内大量设备同时发起 CONNECT。使用 Synchronizing Timer 实现集合点爆破。

心跳维持：设备连接后，每 30~60 秒发送一次小包（如 PINGREQ 或业务心跳）。用 While Controller + Constant Timer 模拟。

上行遥测：设备周期上报传感器数据。重视 Pub 吞吐和 Broker 的 messages.received。

下行指令：控制端下发命令给设备，通过共享订阅或通配符转发。注意Sub端到端延迟。

2. JMeter 压测计划创建技巧

线程组选择：Concurrency Thread Group 或 Stepping Thread Group 可以精细控制“启动-保持-停止”的节奏，如：100线程每步，每30秒增加1000，最后到达10万并发，观察系统拐点。

参数化：使用 CSV Data Set Config 为每个虚拟设备分配独立的 证书、ClientID、上报主题，模拟真实异构设备。

长连接保持：在线程组内使用 Loop Controller 无限循环，在循环内部放置 MQTT Pub 和一个 Constant Timer，保证连接永不释放。

结果收集：

开启 Aggregate Report 查看 TPS、平均/最大延迟。

使用 Backend Listener（如 InfluxDB + Grafana）实时监控，添加 mqtt 自定义度量。

如果自行开发，可在 SampleResult 中设置 Latency 为消息往返时间。

3. 压测环境调优

JMeter 客户端侧（压力机）

JVM 参数：-Xms4g -Xmx4g -XX:MaxMetaspaceSize=256m

操作系统限制：Linux 下 ulimit -n 1048576，调大临时端口范围 net.ipv4.ip_local_port_range = 1024 65535，开启 tcp_tw_reuse。

分布式：单台压力机一般能支撑 5~10 万 MQTT 连接（受限于端口和内存）。超过此数量必须使用 JMeter 分布式集群（Master + Slaves）。

Broker侧（如EMQX）

开启Dashboard的Prometheus统计。

监控 CPU、内存、Erlang 进程数、connections.count、socket.used。

Broker 同样需要调大 ulimit，调优 TCP backlog 等内核参数。

4. 分析短板

连接上不去：检查 JMeter 端 TIME_WAIT 端口耗尽，或 Broker 的 max_connections 限制。

消息延迟增大：查看 Broker 的 mailbox 积压，可能是下游订阅者消费过慢。

Broker 崩溃：一般是内存溢出，需检查客户端是不是异常导致消息风暴（如无限制的离线消息队列），可在 Broker 侧限制 最大飞行窗口 和 消息队列长度。

四、避坑指南和实践

Clean Session 的灵活运用：压测上连接时建议使用 true 减少 Broker 状态维护；测离线消息持久化时，第一个连接用 false 订阅并断开，第二个同 ClientID 再连接证实接收。

SSL/TLS 压测：必须在 JavaSamplerClient 的 setupTest 中加载 SSLSocketFactory，并注意每个连接都加载证书会造成很大的 CPU 开销，可考虑在类级别预加载 SSLContext。

消息体大小：物联网设备消息一般很小（几百字节），压测时不要只测 1KB 小包，也要按实际比例混合 10KB 固件 OTA 报文，观察 Broker 的 TCP 粘包/拆包性能。

WebSocket 模拟：某些 IoT 平台使用 MQTT over WebSocket，Paho 库原生支持，只需将 Broker URL 改为 ws://... 即可。

通过成熟的mqtt-jmeter插件可以解决80%的通用物联网压测需求，而面对私有协议认证、复杂设备影子交互等定制场景，根据Paho的JavaSamplerClient开发能提供极大的灵活性。将两者和JMeter强大的定时器、参数化、分布式能力结合，就能创建出从边缘到云端的可靠物联网性能测试方案。