如果需要压测 MQTT 服务,则需使用 MQTT 插件和 JMeter 生成 JMeter 脚本,再将脚本上传到 PTS 控制台进行压测。

前提条件

背景信息

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

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

步骤一:安装 MQTT 插件

  1. 下载 mqtt-jmeter插件最新版本 JAR 包:mqtt-xmeter-1.0.1-jar-with-dependencies.jar
  2. 拷贝插件 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 英文图形界面为例。

  1. 读取客户端 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 对话框中配置以下信息。ex_pg_csv_data_set_config
      • Filename :客户端信息 CSV 文件名 client.csv
        说明 只输入文件名 client.csv ,不要包含文件路径,以实现文件最大的通用性。将 client.csv 文件保存到 JMeter 进程启动目录以保证能够被读取到。
      • File encoding:CSV 文件编码格式,本示例使用 UTF-8
      • Recycle on EOF:是否循环读取文件。本示例中选择 False,一条客户端信息只能被读取使用一次(与一个线程绑定),不允许循环读文件。
  2. 建立 MQTT 连接。

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

    1. Test Plan 区域右键单击 Once Only Controller,选择 Add > Sampler > MQTT Connect
    2. MQTT Connect 对话框中配置以下信息。ex_pg_MQTT_connection
      配置 参数 说明
      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 分钟发送一次活动心跳。
  3. 配置发布消息。
    1. Test Plan 区域右键单击 Thread Group,选择 Add > Sampler > MQTT Pub Sampler
    2. MQTT Pub Sampler 对话框中配置以下信息。ex_pg_MQTT_pub_sampler
      • QoS Level:客户端向服务器发布消息的服务质量。本示例中选择 0,即只发送一次,丢失不重发,可按需选择其他级别。
      • Topic name:消息 topic 。MQTT topic 支持层次结构,使用 / 分割,类似文件路径,如 pts_test/jmeter 等。
      • Add timestamp in payload:是否添加消息头添加发送时间戳。一般勾选此项,方便测试时检查消息延迟。
      • Payloads:消息体。本示例中填写 Aliyun PTS From ${clientId},即在消息体中添加客户端 ID,方便测试和调试检查。
  4. 订阅接收消息。
    1. Test Plan 区域右键单击 Thread Group,选择 Add > Sampler > MQTT Sub Sampler
    2. MQTT Sub Sampler 对话框中配置以下信息。ex_pg_MQTT_pub_sampler2
      • 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 Count3(循环执行 3 次)。
    4. 在 JMeter 页面左上角单击保存,然后执行脚本。

      在 View Results Tree 页面查看脚本执行结果:

      ex_result_Samplerex_result_response_data
  5. 添加消息订阅线程组。

    当前脚本配置下,执行一次消息订阅可能未收到消息但仍记录为 1 次结果,可能收到多个消息但不能拆分记录消息延迟,导致压测统计数据不准确。

    为了解决此问题,MQTT Sub Sampler 支持另一种工作模式:每收到一条消息产生一个有效的执行结果,这样将更合理。但这种模式下,没有收到消息时线程将一直阻塞,为了避免阻塞整个线程,应将订阅消息拆分到不同的线程组。发布消息线程组移除订阅操作后,应添加等待时间避免单个客户端发布消息过快。实际场景通常是客户端数量多,但每个客户端的消息量少。

    1. Test Plan 区域右键单击 MQTT Pub Sampler ,选择 Add > Time > Constant Timer,设置延时 1000 毫秒,即每次发布消息后等待 1 秒。ex_pg_constant_timer
    2. 新建消息订阅线程组,从消息发布线程组复制读取 CSV 和 MQTT 连接配置,共享 CSV 文件。

      此时一个客户端(对应一条 CSV 记录)要么用于发布消息,要么用于订阅消息,不能两者同时使用。

    3. 在新消息订阅线程组下的单击 MQTT Sub Sampler,然后在 MQTT Sub Sampler 对话框中配置 Sample onnumber of received messages,值设为 1 ,即每收到一条消息产生一条执行结果。56-mqtt-sub-msg.png
    4. 执行脚本。

      测试执行脚本结果如下:

      57-result-sub-1

步骤四:使用 PTS 压测 MQTT 消息

在 PTS 控制台使用 JMeter 压测场景,上传 mqtt-jmeter 插件 JAR 包、JMeter 脚本和客户端信息 CSV 文件。详情请参见 JMeter 压测

ex_MQTT_JMeter

压测结果分析

开始压测后,压测过程中可看到实时并发数(发布消息客户端和订阅消息客户端总数),TPS 和 RT(消息延迟)等统计信息。

ex_MQTT_JMeter_report

本示例中发布消息为广播,每个消息订阅客户端都收到一份消息,总共有 5 个订阅客户端,因此可看到接收消息数约是发布消息数的 5 倍。压测结束后将生成压测报告,可查看场景并发、TPS、响应时间等变化趋势图和统计数据汇总等信息。如果出现错误,还可以结合请求采样日志和 JMeter 日志等进行排查。

ex_MQTT_JMeter_report1

常见问题

发布消息包含中文字符时,为什么查看接收到的消息包含乱码?由于 MQTT 消息体为二进制数据,发布消息时字符串使用系统编码(UTF-8 或 GBK)转为二进制,接收消息时默认使用 ISO-8859-1 将二进制转为字符串,两边编码不一致导致乱码现象。