前提条件

已部署可在公网访问的 MQTT 服务。
已安装 JMeter 5.x 版本。

背景信息

MQTT 是专为移动互联网（Mobile Internet）、物联网（IoT）设计的超轻量级消息协议，用于连接移动端与云服务双向通信，广泛应用于各种应用领域，如端向云汇报状态、云向端推送消息、端向端发送消息（即时聊天）等场景。

现有使用 MQTT 发布广播消息，共有 5 个订阅客户端，每个消息订阅客户端都收到一份消息的场景，本文以此类场景为例介绍如何使用 PTS 的 JMeter 模式压测 MQTT 服务。

步骤一：安装 MQTT 插件

下载 mqtt-jmeter插件最新版本 JAR 包：mqtt-xmeter-1.0.1-jar-with-dependencies.jar。

拷贝插件 JAR 包到 JMeter 安装目录的 lib/ext/ 子目录下。

cp mqtt-xmeter-1.0.1-jar-with-dependencies.jar "${JMETER_HOME:?}"/lib/ext/

步骤二：准备客户端信息 CSV 文件

每个 MQTT 客户端需要提供以下信息：

用户名和密码。
为保证安全，MQTT 服务器通常需要认证，最常用的认证方式即用户名和密码认证。
客户端 ID（ClientId）。
客户端 ID 是全局唯一 ID，客户端唯一标识，通常与用户设备关联。如关联手机序列号，则可通过客户端 ID 变化检测用户是否更换了新手机。每个用户可能有一个或多个客户端 ID 。

客户端信息 CSV 文件应包含 UserName 、 Password 、 ClientId 这 3 列数据。例如期望压测 10000 台客户端设备同时在线的场景，则需要准备 10000 条客户端信息。

如客户端信息示例 CSV 文件 client.csv 内容如下：

UserName,Password,ClientId
pts_test,pts-pass1,client-test0001
pts_test,pts-pass1,client-test0002
pts_test,pts-pass1,client-test0003

说明

做压测时建议使用测试用户数据，避免泄露真实用户信息，避免真实用户产生脏数据。
使用预先准备的测试客户端 ID 方便服务器做客户端 ID 校验，同时方便跟踪排查问题。
手动编辑 CSV 文件很容易出错，推荐使用 EXECL、Numbers 等软件导出，或使用 Apache commons-csv程序生成。

步骤三：本地编辑 JMeter 脚本

本教程以 JMeter 5.x 英文图形界面为例。

读取客户端 CSV 数据文件。
1. 打开 JMeter，并新建脚本。
2. 在 JMeter 左侧文件目录中右键单击 Test Plan ，选择 Add > Threads (Users) > Thread Group。
3. 在 JMeter 左侧文件目录中右键单击 Test Plan ，选择 Add > Listener > > View Results Tree ，添加 View Results Tree 监听器，方便本地调试测试脚本。
4. 在 Test Plan 区域右键单击 Thread Group ，选择 Add > Logic Controller > Once Only Controller。JMeter 中一个线程模拟一个 MQTT 客户端设备，使用 Once Only Controller 保证一个线程仅读取一次客户端 CSV 数据文件，绑定一条客户端信息。
5. 在 Test Plan 区域右键单击 Once Only Controller ，选择 Add > Config Element > CSV Data Set Config。并在 CSV Data Set Config 对话框中配置以下信息。
  - Filename ：客户端信息 CSV 文件名 client.csv 。
    
    说明只输入文件名 client.csv ，不要包含文件路径，以实现文件最大的通用性。将 client.csv 文件保存到 JMeter 进程启动目录以保证能够被读取到。
  - File encoding：CSV 文件编码格式，本示例使用 UTF-8 。
  - Recycle on EOF：是否循环读取文件。本示例中选择 False，一条客户端信息只能被读取使用一次（与一个线程绑定），不允许循环读文件。

建立 MQTT 连接。

同样使用 Once Only Controller 控制一个客户端（对应一个线程）只需执行一次建连操作。

在 Test Plan 区域右键单击 Once Only Controller，选择 Add > Sampler > MQTT Connect。

在 MQTT Connect 对话框中配置以下信息。消息队列MQTT压力测试服务

配置	参数	说明
MQTT 连接配置	Server name or IP	MQTT 服务器公网地址。客户端设备通常使用公网访问 MQTT 服务。
	Port number	MQTT 服务器端口。例如 `1883`，即使用标准 TCP 端口。
	MQTT version	MQTT 版本。例如 `3.1.1`，目前主流 MQTT 服务器都支持 3.1.1 版本。
	Timeout(s)	超时秒数填写，即客户端建立连接、发送消息等相关操作的超时时间。例如 `10`。
	Protocols	连接协议。选择 `TCP`，即使用标准 TCP 连接协议。
MQTT 客户端配置	User name	从 CSV 文件读取 userName 字段。填写 `${userName}`。
	Password	从 CSV 文件读取 Password 字段。填写 `${password}` 。
	ClientId	从 CSV 文件读取 ClientId 字段。填写 `${clientId}` 。
	Add random suffix for ClientId	是否添加后缀。本例使用预先准备好的固定客户端 ID，不要添加后缀，则取消勾选。
	Keep alive(s)	活动心跳间隔秒数。例如 `300`，即连接空闲时，每 5 分钟发送一次活动心跳。

配置发布消息。
1. 在 Test Plan 区域右键单击 Thread Group，选择 Add > Sampler > MQTT Pub Sampler
2. 在 MQTT Pub Sampler 对话框中配置以下信息。
  - QoS Level：客户端向服务器发布消息的服务质量。本示例中选择 0，即只发送一次，丢失不重发，可按需选择其他级别。
  - Topic name：消息 topic 。MQTT topic 支持层次结构，使用 / 分割，类似文件路径，如 pts_test/jmeter 等。
  - Add timestamp in payload：是否添加消息头添加发送时间戳。一般勾选此项，方便测试时检查消息延迟。
  - Payloads：消息体。本示例中填写 Aliyun PTS From ${clientId}，即在消息体中添加客户端 ID，方便测试和调试检查。
订阅接收消息。
1. 在 Test Plan 区域右键单击 Thread Group，选择 Add > Sampler > MQTT Sub Sampler。
2. 在 MQTT Sub Sampler 对话框中配置以下信息。
  - QoS Level：服务器向客户端推送消息的服务质量。本示例选择 0，即只发送一次，丢失不重发，可按需选择其他级别。
  - Topic name：消息 topic 。应与发布消息的 topic 匹配，如本例中为 pts_test 。
  - Payload includes timestamp：是否在接收消息后从消息头解析发送时间。一般勾选此项，可计算出消息延迟，即从发布端、途经服务器、最后到达订阅端花费的总时间。
  - Sample on：选择 specified elapsed time (ms)，值填写 1000，表示持续接收消息 1000 毫秒。这段时间内，可能一条消息都接收不到，也可能接收到多条消息。
  - Debug response：一般勾选此项，用于记录接收到的消息内容，方便调试排查问题。正式执行性能测试时可取消该选项以优化性能和减少内存占用。
3. 在 Test Plan 区域单击 Thread Group，配置 Loop Count 为 3（循环执行 3 次）。
4. 在 JMeter 页面左上角单击保存，然后执行脚本。
  在 View Results Tree 页面查看脚本执行结果：
添加消息订阅线程组。
当前脚本配置下，执行一次消息订阅可能未收到消息但仍记录为 1 次结果，可能收到多个消息但不能拆分记录消息延迟，导致压测统计数据不准确。

为了解决此问题，MQTT Sub Sampler 支持另一种工作模式：每收到一条消息产生一个有效的执行结果，这样将更合理。但这种模式下，没有收到消息时线程将一直阻塞，为了避免阻塞整个线程，应将订阅消息拆分到不同的线程组。发布消息线程组移除订阅操作后，应添加等待时间避免单个客户端发布消息过快。实际场景通常是客户端数量多，但每个客户端的消息量少。
1. 在 Test Plan 区域右键单击 MQTT Pub Sampler ，选择 Add > Time > Constant Timer，设置延时 1000 毫秒，即每次发布消息后等待 1 秒。
2. 新建消息订阅线程组，从消息发布线程组复制读取 CSV 和 MQTT 连接配置，共享 CSV 文件。
  此时一个客户端（对应一条 CSV 记录）要么用于发布消息，要么用于订阅消息，不能两者同时使用。
3. 在新消息订阅线程组下的单击 MQTT Sub Sampler，然后在 MQTT Sub Sampler 对话框中配置 Sample on 为 number of received messages，值设为 1 ，即每收到一条消息产生一条执行结果。
4. 执行脚本。
  测试执行脚本结果如下：