简介MQTT自动化测试场景下这款基于C# WinForm开发的自动发消息工具主要面向需要模拟消息高峰但仍不具备编码能力的测试人员。通过简单配置MQTT连接、发送间隔和总次数即可自动生成日期、序列号、Mac地址、整数、浮点数等随机消息并支持实时日志查看帮助验证业务程序在消息冲击下的健壮性整个配置与运行过程无需编写额外代码。压缩包共35个文件约5.47MB包含可直接运行的主程序、依赖dll库、xml配置文件、pdf与md说明文档、log运行日志及本地数据库文件其中dll用于支撑程序运行xml用于保存配置文档则提供详细使用与更新说明结构完整便于本地部署和二次调试。目前已有698人学习下载适合软件测试、物联网开发及运维人员用于自动化测试场景下的MQTT消息模拟与验证工具当前为V1.0.8版本更新说明与使用手册均在包内可帮助使用者快速上手并降低消息构造与发送的门槛。1. 把MQTT自动发送消息做成可控脚本是自动化测试MQTT的第一步把MQTT客户端点开手动输入topic和payload选好QoS再点发布这是很多人对MQTT自动化测试的初印象。一条两条可以手点但200条不同设备、不同payload规则的消息就点不动了更不用说反复复现丢包、乱序、离线消息。做自动化测试MQTT首先落地的通常就是这个MQTT自动发送消息软件。这类软件的实质还是一个MQTT客户端只是它把发布动作变成可控脚本按topic与payload规则批量发按QoS和频率控制行为并且能和订阅端一起完成断言。常见落地做法是用Python paho-mqtt写一套发布/订阅脚本再用mqttx这类命令行MQTT工具做快速校验。它解决的不是“怎么连上broker”而是“怎么在测试里稳定地制造符合场景的MQTT消息”。适合物联网功能测试、协议测试、嵌入式设备联调、后端链路联调的工程师。下面从协议参数开始讲再做最小脚本落到自动化断言最后给常驻使用技巧。2. MQTT协议详解自动发送前要定下的连接与消息参数写自动发送脚本之前必须先过一遍MQTT协议里影响消息行为的参数。很多人以为发消息就是把topic和payload拼好再publish真实环境里决定消息能不能送达、会不会重复、会不会在订阅时突然冒出一条旧消息的往往是QoS、retain、clean_session这些容易被忽略的默认值。2.1 MQTT协议详解影响自动发送消息结果的5个参数2.1.1 QoS级别QoS 0 和 QoS 2 在自动化测试里分别怎么用QoS 0是发送端把消息交给TCP就不管了不等待任何确认适合高频遥测和吞吐测试QoS 1会有一个PUBACK确认保证至少到达一次但可能重复QoS 2则走PUBREC/PUBREL/PUBCOMP四次握手严格只到达一次单条消息消耗的资源明显更大。自动化测试里常见错误是全程固定用QoS 0结果弱网下消息丢了分不清是broker故障还是网络抖动。我一般会把QoS也当作被测项同一组消息分别以qos0、1、2跑一遍记录丢包率、重复率、端到端延迟。这样自动发送消息软件本身不屏蔽协议层差异测试结果才有区分度。2.1.2 retain与clean_session测离线消息与清理历史残留retain保留消息是自动化测试环境里的头号污染源。broker上保留消息不会因为发送端断开而消失新的订阅者一订阅就立刻收到旧payload。如果新用例断言“收到第一条消息”很可能拿到的是上一轮测试留下的脏数据。清理办法是向该topic发布一条空payload且retainTrue的消息clean_client.publish(device/001/status, payloadb, qos1, retainTrue)空payload加retainTrue的语义是清除该topic的保留消息。在用例的setup阶段先执行这一步再开始自动发送能避免大量诡异失败。clean_session则决定连接断开后broker是否保留该客户端的会话状态。测离线消息补发时要用clean_sessionFalse并且在paho-mqtt 2.x里配合session_expiry_interval一起设置否则broker可能很快清理会话离线消息根本没有机会补发。2.1.3 keepalive与遗嘱消息用自动发送脚本模拟异常下线keepalive告诉broker“多久没收到报文就把我判死”。自动化脚本里设30秒通常够用如果要测断线重连可以故意设成5秒然后断开网络观察broker判定死链的时间。遗嘱消息是异常下线通知。客户端可以在connect时指定will_topic和will_payload进程被kill、网络断开时broker代发遗嘱。自动发送脚本要测这个场景不能优雅disconnect而是直接os._exit()或kill掉进程再订阅遗嘱topic验证broker是否按预期发布了遗嘱消息。2.2 客户端选型paho-mqtt、mqttx 与 Java/Node 客户端的取舍客户端语言/形态适用场景注意点paho-mqttPython库自动化测试脚本主力2.x需显式传callback_api_versionmqttxCLI/桌面临时验证topic和payload支持MQTT 5.0适合快速排错paho.mqtt.javaJava库Spring Boot/Netty服务自测和业务代码同语言调试方便mqtt.jsNode/浏览器前端联调与mock事件驱动适合轻量测试Python侧paho-mqtt是自动化测试最常用的选择脚本短、好维护也能和pytest深度集成。Java接口自动化测试框架如果对接的是Spring Boot 3.x Netty MQTT这类物联网服务paho.mqtt.java是自然选择但测试团队通用做法仍然是Python脚本Java工程只做服务端自测。mqttx我一般放在排错场景先命令行发一条消息确认链路通再回脚本里跑完整用例。2.3 先准备一个本地broker自动化测试脚本的验证环境自动发送脚本最好跑在本地临时broker上而不是直接打共享测试环境。常见做法是用docker起一个mosquittodocker run -d --name mqtt-test -p 1883:1883 eclipse-mosquitto:2.0本地broker的优势是可以随时清掉重建还能通过docker logs mqtt-test看连接与发布日志。确认脚本逻辑稳定后再把broker地址换成真实测试服务器。这样自动发送消息软件本身出的问题不会和网络环境问题混在一起。3. 用Python paho-mqtt实现最小MQTT自动发送消息脚本3.1 最小脚本连接本地broker并按topic批量发布下面这段代码是自动化测试MQTT的最小闭环用paho-mqtt 2.x运行import json import time import paho.mqtt.client as mqtt client mqtt.Client( callback_api_versionmqtt.CallbackAPIVersion.VERSION2, # paho-mqtt 2.x 必需 client_idauto-send-001, clean_sessionTrue, ) client.connect(127.0.0.1, port1883, keepalive30) client.loop_start() time.sleep(0.2) # 等后台线程完成 CONNACK 处理 for seq in range(20): payload json.dumps({seq: seq, state: running}) info client.publish(device/001/status, payloadpayload, qos1, retainFalse) print(fseq{seq} rc{info.rc} mid{info.mid}) time.sleep(0.1) time.sleep(0.5) # 等 QoS1 的 PUBACK 处理完 client.loop_stop() client.disconnect()connect建立TCP连接并完成MQTT握手但它不会自动读取后续报文所以紧接着要loop_start()开启后台网络循环。sleep(0.2)是为了让CONNACK已经被后台线程处理connect本身失败时不抛异常错误会在on_connect回调或后续publish时暴露。publish返回的info.rc是关键参数rc等于0表示消息被paho本地接收如果是MQTTErrorCode里的其他值说明连接状态异常。脚本最后sleep(0.5)再loop_stop是为了让QoS 1的PUBACK有时间回来避免进程退出时消息还在发送队列里。3.2 连接参数配置表client_id、keepalive、clean_session、session_expiry_interval、重连延迟参数推荐值影响自动化测试场景client_id按用例唯一同id会互踢多设备模拟必须不同idkeepalive30控制broker判定死链时间测断线重连时调小到5clean_sessionTrue是否保留会话测离线消息时设Falsesession_expiry_interval不设或300MQTT 5.0会话保留时长配合clean_sessionFalse使用reconnect_delay_set(1, 30)控制自动重连退避验证网络恢复后订阅是否还在reconnect_delay_set是paho提供的方法设置最小和最大重连延迟client.reconnect_delay_set(min_delay1, max_delay30)它控制paho在断线后自动重连的退避策略。注意clean_sessionTrue时重连后订阅关系不保留必须在on_connect回调里重新subscribeclean_sessionFalse时broker会按session_expiry_interval保留订阅重连后不用重新订阅。3.3 用mqttx命令行工具交叉验证自动发送逻辑脚本跑通后还需要一条独立路径验证broker状态。mqttx是常用的MQTT客户端工具命令行模式适合快速验证mqttx pub -h 127.0.0.1 -p 1883 -t device/001/status \ -m {seq:21,state:running} -q 1如果你本机装了mosquitto也可以用等效的mosquitto_pubmosquitto_pub -h 127.0.0.1 -p 1883 -t device/001/status \ -m {seq:22,state:running} -q 1这套交叉验证的价值在于Python脚本、mqttx、mosquitto_pub三者发送相同topic与payload如果只有Python脚本失败问题在paho的配置如果三个工具都失败问题在broker或网络。自动发送脚本本身写得多好都不如两条独立命令能更快定位故障边界。4. 自动化测试MQTT实战从“发出去了”到“结果对不对”4.1 模拟多设备并发上报如何组织topic与客户端物联网测试里常见的场景是大量设备同时上报状态。topic按设备编号组织比如device/001/status、device/002/status。模拟多设备时每个设备必须使用独立client_id和独立连接不能在一个连接里轮流改topic模拟多个设备因为broker视角看到的是同一个客户端。多线程实现时paho的client对象不是线程安全的不要让多个线程共享同一个client实例import threading import time import paho.mqtt.client as mqtt def device_worker(device_id: str): worker mqtt.Client( callback_api_versionmqtt.CallbackAPIVersion.VERSION2, client_idfsim-{device_id}, clean_sessionTrue, ) worker.connect(127.0.0.1, 1883, 60) worker.loop_start() for i in range(10): payload f{{device: {device_id}, seq: {i}}} worker.publish(fdevice/{device_id}/status, payload, qos1) time.sleep(0.2) worker.loop_stop() worker.disconnect() threads [ threading.Thread(targetdevice_worker, args(f{i:03d},)) for i in range(10) ] for t in threads: t.start() for t in threads: t.join()每个线程创建独立连接client_id为sim-001到sim-010broker会看到10个独立客户端。线程数量控制在20个以内比较稳妥太多线程同时连接会让本地broker的连接线程池打满测试结果里出现大量unexpected disconnect容易误判为代码问题。4.2 在一个测试脚本里同时订阅与发布用断言验证MQTT响应自动发送只是前半段自动化测试MQTT的完整闭环必须包含“发出去后验证响应”。在同一个Python进程里开两个client一个负责订阅响应topic一个负责发指令import queue import time import paho.mqtt.client as mqtt responses queue.Queue() def on_response(client, userdata, msg): responses.put({topic: msg.topic, payload: msg.payload, ts: time.time()}) sub mqtt.Client( callback_api_versionmqtt.CallbackAPIVersion.VERSION2, client_idtest-assert-sub, ) sub.on_message on_response sub.connect(127.0.0.1, 1883, 60) sub.subscribe(device//response, qos1) sub.loop_start() time.sleep(0.3) pub mqtt.Client( callback_api_versionmqtt.CallbackAPIVersion.VERSION2, client_idtest-assert-pub, ) pub.connect(127.0.0.1, 1883, 60) pub.publish(device/001/command, {cmd:reboot}, qos1) try: resp responses.get(timeout5) except queue.Empty: raise AssertionError(5 秒内未收到 MQTT 响应消息) assert bcode:0 in resp[payload], resp[payload]queue.get(timeout5)是超时断言的关键5秒内没有响应就抛异常。订阅topic用了通配符匹配device/001/response、device/002/response等任意设备号。响应payload需要断言具体字段比如code为0。这类字段断言在opc ua转mqtt的链路里尤其有用topic不变但payload结构由转换规则决定结构一调整断言立刻暴露问题。4.3 自动化测试MQTT时最容易踩的3个坑4.3.1 retain消息残留污染断言订阅同一个topic的用例第一轮成功第二轮开始收到旧payload这是retain残留。处理方式有两个用例setup阶段向该topic发空retain消息清理或者在on_message里过滤第一条retain消息if msg.retain 1 and is_first_message: return4.3.2 payload超过broker限制导致连接被断开broker对单条消息大小有上限比如mosquitto默认的max_packet_size。payload过大时publish本地返回正常但broker可能直接断开连接。自动发送前先检查字节长度payload_bytes payload.encode(utf-8) if len(payload_bytes) broker_max_packet_size: raise ValueError(fpayload too large: {len(payload_bytes)})4.3.3 QoS 2重复投递与“只收到一次”的错误断言QoS 2虽然协议上保证只到达一次但实际broker实现或客户端重发机制不完善时仍可能观察到重复消息。断言不要写死“这条消息只能出现一次”改为记录消息序号并去重统计seen set() if msg_id not in seen: seen.add(msg_id) else: duplicate_count 1这样既能验证消息到达又能单独统计重复率。对QoS 2做重复率断言时阈值按业务要求设而不是默认0。5. 把自动发送脚本升级成常驻MQTT测试工具5.1 用TLS加密链路验证证书与双向认证兼顾STM32设备场景设备端无论是STM32还是4G模块上云时MQTT TLS加密通信几乎是必选项。自动发送脚本要模拟这个链路得在paho里配置CA证书和客户端证书import ssl client.tls_set( ca_certsca.crt, certfileclient.crt, keyfileclient.key, tls_versionssl.PROTOCOL_TLS_CLIENT, ) client.connect(mqtt.internal, 8883, 60)tls_set之后connect走TLS握手证书链或域名校验失败会直接抛异常。测试脚本里要区分两种失败握手失败是证书问题握手成功但认证失败是用户名密码或client_id权限问题。自签名证书调试时可以临时用tls_insecure_set(True)但正式回归用例必须关掉否则证书过期这类线上风险测不出来。5.2 发送二进制payload处理固件包与原始字节流部分自动化测试需要发送原始字节流比如OTA固件包或协议自定义的二进制帧。paho的publish直接支持bytes类型payloadwith open(firmware.bin, rb) as f: data f.read() crc zlib.crc32(data) 0xFFFFFFFF client.publish(ota/device/001, data crc.to_bytes(4, big), qos1)注意不要把bytes先转成str再发这样会破坏原始数据。二进制payload断言时先比较长度再校验末尾CRC长度一致且CRC正确才能判定消息完整。5.3 参数化入口与CI接入让自动发送脚本成为可重复执行的回归用例最后把脚本做成命令行入口broker地址、topic、QoS、消息数量全部参数化python mqtt_smoke.py --broker 127.0.0.1 --port 1883 \ --topic device/001/command --qos 1 --count 200 --expect-reply脚本内部所有断言失败时返回非0退出码成功返回0Jenkins或GitHub Actions直接当作一个普通测试任务执行。消息量要克制单个用例控制在200条以内消息间隔不低于0.1秒并发设备不超过20个否则测试环境先被压垮断言结果失去参考意义。建议把QoS、payload模板、topic前缀抽成一个JSON配置CI回归用例只需改配置文件不需要动Python代码。本文还有配套的精品资源点击获取