Ubuntu下Mosquitto订阅端性能测试：Python3实现与QoS吞吐量分析

1. 从一个订阅端压测需求说起做物联网平台的朋友大概率都绕不开 MQTT 这个协议。它轻量、省流量、支持一对多发布订阅几乎是设备上云的默认选项。但真正把设备接进来之后很多人会卡在同一个问题上服务端到底能扛多少连接、多少消息吞吐这个问题不测清楚上线就是赌博。我最近在做一个设备接入层的容量评估需要摸清一台普通云主机上 Mosquitto 的订阅端极限。发布端好测随便写个循环就能压订阅端反而麻烦因为它要维持长连接、要处理回调、还要统计延迟和丢包。标题里提到的“Ubuntu 下 Mosquitto python3 实现的订阅端性能测试”说的就是这件事——在 Ubuntu 环境里用 Python3 写一个 MQTT 订阅客户端对 Mosquitto 做订阅侧的性能压测。这篇文章适合三类人看一是刚接触 MQTT、想搞明白订阅端到底怎么写的开发者二是已经在用 Mosquitto、但没做过系统压测的运维或后端三是需要给物联网项目做容量规划的技术负责人。我会把整个测试的设计思路、代码实现、参数计算、踩过的坑全部摊开讲代码可以直接抄去改。核心关键词就几个Mosquitto、Python3、订阅端、性能测试、QoS、吞吐量后面会反复出现。先说结论方向订阅端压测的难点不在“连上”而在“稳定地连上很多个、并且准确统计每个消息的到达情况”。很多人第一次写压测脚本跑几百个连接就崩了问题往往出在文件描述符、线程模型和回调阻塞上而不是 Mosquitto 本身不行。2. 测试方案的整体设计与选型考量2.1 为什么订阅端压测比发布端更难发布端压测的逻辑很线性建立连接然后疯狂 publish统计发送速率就行。订阅端的复杂度高一个量级原因有三个。第一订阅端要维持连接的生命周期。一个订阅客户端从 CONNECT 到 SUBSCRIBE 到持续接收 PUBLISH中间任何一步出问题都会导致统计失真。连接数一多资源消耗是叠加的。第二订阅端要处理回调。Python 的 MQTT 库比如 paho-mqtt是基于回调的on_message 里做的事情如果太重会直接拖慢整个客户端的消息处理速度测出来的“吞吐”其实是你的回调处理速度不是 Mosquitto 的推送能力。第三订阅端要准确统计。延迟怎么算丢包怎么算消息乱序怎么处理这些统计逻辑本身会引入开销设计不好就会污染测试结果。所以订阅端压测的核心矛盾是既要模拟真实订阅行为又要让统计开销尽可能小。这个矛盾贯穿整个方案设计。2.2 工具选型为什么是 Mosquitto paho-mqtt Python3服务端选 Mosquitto 没什么争议。它是 Eclipse 基金会维护的开源 MQTT broker体积小、依赖少、单机性能在同类里属于第一梯队Ubuntu 的 apt 源里直接就有装完就能用。对于容量评估这种场景Mosquitto 足够代表一类轻量 broker 的典型表现。客户端库选 paho-mqtt理由也直接它是 Python 生态里最成熟的 MQTT 客户端API 稳定支持 MQTT 3.1/3.1.1/5.0社区资料多出问题好查。虽然它在极限性能上不如用 C 或 Go 写的客户端但我们的目标是测 Mosquitto 的订阅侧承载能力用 Python 客户端反而更贴近大多数真实业务场景——毕竟大量物联网后端就是用 Python 写的。Python3 的版本建议 3.8 以上主要是为了 asyncio 和更好的多线程支持。不过这里有个关键决策用多线程还是多进程还是单线程事件循环这个选择直接决定了能压到多少连接。2.3 线程模型的关键取舍我一开始想用单线程 事件循环因为 paho-mqtt 自带 loop_start() 会起一个后台线程跑网络循环理论上单进程能撑不少连接。实测下来单进程跑到 800 个连接左右就开始出现明显的消息延迟抖动CPU 单核跑满。原因是 paho 的回调都在同一个网络线程里执行连接一多回调排队就严重。后来改成多进程模型每个进程负责一批连接进程内用多线程或者 loop_start 管理。这样能利用多核但进程间统计汇总又成了新问题。最终的方案是多进程 每进程内多客户端实例 共享内存统计具体在第四节展开。这里先给个经验值在一台 4 核 8G 的 Ubuntu 云主机上纯订阅不做业务处理的场景Mosquitto 单机撑 5000 到 10000 个订阅连接是比较现实的区间具体取决于 QoS 和消息频率。超过这个量级瓶颈往往先出现在客户端侧而不是 broker 侧。2.4 测试指标的定义压测之前必须把指标定义清楚否则数据没法解读。我关注四个核心指标连接建立成功率发起 N 个连接成功建立 M 个M/N 就是成功率。低于 99% 说明系统参数或 broker 配置有问题。消息吞吐量msg/s单位时间内订阅端收到的 PUBLISH 消息总数。注意是“收到”不是“发出”。端到端延迟从发布端发出到订阅端收到的时间差。这需要发布端在 payload 里打时间戳。消息丢失率发布总数减去接收总数再除以发布总数。QoS 0 下丢包是正常的QoS 1/2 下丢包就说明有问题。这四个指标里延迟最难测准因为它依赖发布端和订阅端的时钟同步。如果两端在同一台机器上可以用 monotonic 时钟跨机器就必须做时间同步否则测出来的延迟没有意义。3. 环境准备与 Mosquitto 服务端配置3.1 Ubuntu 下安装 Mosquitto 的完整步骤安装本身很简单但有几个细节不注意会埋坑。sudo apt update sudo apt install -y mosquitto mosquitto-clients装完之后 Mosquitto 默认会以服务形式启动。先确认状态sudo systemctl status mosquitto默认配置下Mosquitto 只监听 127.0.0.1:1883且不允许匿名连接较新版本默认 local_only 且需要认证。做压测必须改配置否则客户端连不上。配置文件在/etc/mosquitto/conf.d/下新建一个比如stress.conflistener 1883 0.0.0.0 allow_anonymous true max_connections -1 max_queued_messages 10000 max_inflight_messages 100逐条解释一下为什么这么配listener 1883 0.0.0.0监听所有网卡方便从其他机器压测。如果只在本机测用 127.0.0.1 更安全。allow_anonymous true压测环境图省事生产环境千万别这么干。max_connections -1不限制最大连接数让系统资源自己决定上限。max_queued_messages每个客户端排队消息的上限。压测时如果订阅端处理慢消息会堆积这个值决定了堆积多少之后开始丢弃。max_inflight_messagesQoS 1/2 下同时处于“已发送未确认”状态的消息数。这个值影响吞吐和内存的平衡。改完重启sudo systemctl restart mosquitto3.2 系统级参数调优这一步最容易被忽略Mosquitto 本身配置好了但 Linux 系统的默认限制会先把你卡死。压测前必须调这几个参数。文件描述符限制。每个 MQTT 连接占一个 fd默认的 1024 根本不够。查看当前值ulimit -n临时改当前 shell 有效ulimit -n 65535永久改要编辑/etc/security/limits.conf加上* soft nofile 65535 * hard nofile 65535注意systemd 服务不受 limits.conf 控制Mosquitto 作为服务运行时要在 service 文件里单独配LimitNOFILE65535或者用systemctl edit mosquitto加 override。TCP 相关参数。大量短连接或频繁重连时TIME_WAIT 会堆积。编辑/etc/sysctl.confnet.core.somaxconn 32768 net.ipv4.tcp_max_syn_backlog 32768 net.ipv4.tcp_tw_reuse 1 net.ipv4.ip_local_port_range 1024 65535tcp_tw_reuse允许复用 TIME_WAIT 状态的连接对压测场景很有用。改完sudo sysctl -p生效。提示这些参数在生产环境要谨慎尤其是tcp_tw_reuse某些场景下可能引发连接异常。压测环境可以放开生产环境建议只调 somaxconn 和 backlog。3.3 Python 环境与依赖安装Python3 一般系统自带。装 paho-mqttpip3 install paho-mqtt建议用虚拟环境隔离避免污染系统 Pythonpython3 -m venv mqtt-test source mqtt-test/bin/activate pip install paho-mqttpaho-mqtt 的版本要注意1.x 和 2.x 的 API 有差异。2.x 版本里Client()的构造参数变了回调签名也调整过。我用的 1.6.1稳定且资料多。如果装的是 2.x代码里的client mqtt.Client()要改成mqtt.Client(mqtt.CallbackAPIVersion.VERSION1)才能兼容旧写法。确认版本python3 -c import paho.mqtt.client as mqtt; print(mqtt.__version__)4. 订阅端压测脚本的完整实现4.1 单客户端订阅的最小可用代码先把最小骨架跑通再谈压测。一个能收消息的订阅端长这样import paho.mqtt.client as mqtt import time def on_connect(client, userdata, flags, rc): print(fconnected with result code {rc}) client.subscribe(test/topic, qos1) def on_message(client, userdata, msg): print(freceived: {msg.payload.decode()} on {msg.topic}) client mqtt.Client(client_idsub_test_001) client.on_connect on_connect client.on_message on_message client.connect(127.0.0.1, 1883, 60) client.loop_forever()这段代码能跑但直接拿来做压测有三个致命问题print会拖慢回调、loop_forever阻塞主线程没法做统计、单个 client_id 重复连接会被 broker 踢掉。压测脚本必须解决这些。4.2 压测脚本的核心结构设计我的压测脚本分成四层连接管理层、消息处理层、统计层、控制层。连接管理层负责批量创建客户端、分配唯一的 client_id、处理重连。消息处理层就是 on_message 回调但里面只做最轻量的操作——记录时间戳、计数绝不做字符串格式化或 IO。统计层用共享内存或原子计数器汇总数据。控制层负责启动、停止、输出报告。先看 client_id 的生成。Mosquitto 对 client_id 有长度限制默认 23 字节可配置且同一个 client_id 重复连接会踢掉前一个。压测时用sub_{进程号}_{序号}的格式保证唯一client_id fsub_{os.getpid()}_{index}再看 on_message 的轻量化。压测时最忌讳在回调里做重活。我的做法是回调里只做两件事从 payload 里解析出发布端打的时间戳然后更新计数器。def on_message(client, userdata, msg): recv_ts time.monotonic() # payload 格式: seq:timestamp try: seq_str, send_ts_str msg.payload.decode().split(:) send_ts float(send_ts_str) latency recv_ts - send_ts stats.record_latency(latency) stats.incr_received() except Exception: stats.incr_error()注意这里用的是time.monotonic()不是time.time()。monotonic 不受系统时钟调整影响测延迟更准。前提是发布端和订阅端在同一台机器上或者两端时钟已经严格同步。4.3 多进程 多连接的具体实现单进程撑不住大量连接所以用多进程。每个进程负责一批客户端进程数一般设成 CPU 核数。import multiprocessing as mp import paho.mqtt.client as mqtt def worker(proc_id, conn_count, broker, topic, qos, duration, result_queue): clients [] for i in range(conn_count): cid fsub_{proc_id}_{i} client mqtt.Client(client_idcid) client.on_connect make_on_connect(topic, qos) client.on_message make_on_message(proc_id) client.connect(broker, 1883, 60) client.loop_start() clients.append(client) time.sleep(duration) for c in clients: c.loop_stop() c.disconnect() result_queue.put((proc_id, get_local_stats())) if __name__ __main__: result_queue mp.Queue() processes [] for pid in range(PROC_COUNT): p mp.Process(targetworker, args(pid, CONN_PER_PROC, ...)) p.start() processes.append(p) for p in processes: p.join()这里有几个关键点。loop_start()会为每个客户端起一个后台网络线程连接数一多线程数就爆炸。更省资源的做法是多个客户端共享一个网络循环但 paho 不直接支持。折中方案是控制单进程的连接数比如每个进程 200 到 500 个连接进程数按核数来。make_on_connect和make_on_message用闭包生成是为了让每个客户端能拿到自己的上下文比如进程 id、统计对象避免全局变量污染。4.4 统计数据的汇总与延迟计算跨进程统计是个麻烦事。简单方案是每个进程结束后把本地统计通过 Queue 传回主进程汇总。但如果要实时看吞吐就得用共享内存。我用的是multiprocessing.Value做原子计数器from multiprocessing import Value received_count Value(L, 0) # unsigned long error_count Value(L, 0) def on_message(client, userdata, msg): with received_count.get_lock(): received_count.value 1get_lock()保证多进程写入安全。但注意加锁本身有开销如果消息频率极高比如每秒几万条锁竞争会成为瓶颈。这时候可以每个进程维护本地计数定期比如每秒汇总一次到共享变量用空间换时间。延迟统计不能只存平均值因为平均值会掩盖长尾。我记录 P50、P95、P99 三个分位数。实现上每个进程维护一个延迟列表最后汇总排序取分位。如果内存吃紧可以用直方图近似。def percentile(data, p): if not data: return 0 data_sorted sorted(data) idx int(len(data_sorted) * p / 100) return data_sorted[min(idx, len(data_sorted) - 1)]4.5 发布端配合脚本订阅端要测吞吐必须有发布端持续发消息。发布端脚本相对简单关键是 payload 里要带序号和时间戳import time def publish_loop(client, topic, rate, duration): start time.monotonic() seq 0 interval 1.0 / rate while time.monotonic() - start duration: ts time.monotonic() payload f{seq}:{ts} client.publish(topic, payload, qos1) seq 1 time.sleep(interval)rate是目标发送速率。如果要测极限吞吐就去掉 sleep用循环狂发但要注意发布端自己别成为瓶颈。发布端建议用 C 或 Go 写或者多进程 Python否则它先撑不住测出来的就是发布端的能力。注意发布端和订阅端如果跑在同一台机器上会互相抢 CPU 和网络测出来的数据偏悲观。有条件的话分两台机器用千兆或万兆内网连接。5. 实测数据与性能瓶颈分析5.1 测试环境与参数我的测试环境是一台 4 核 8G 的 Ubuntu 22.04 云主机Mosquitto 2.0.xpaho-mqtt 1.6.1。测试参数如下参数取值订阅连接数1000 / 3000 / 5000QoS0 / 1消息大小64 字节发布速率1000 / 5000 / 10000 msg/s测试时长60 秒进程数45.2 连接数对资源的影响先看连接建立阶段。1000 连接时Mosquitto 进程内存约 80MBCPU 空闲。3000 连接时内存涨到 200MB 左右。5000 连接时内存约 350MB且连接建立时间明显变长从发起到全部连上花了将近 20 秒。这里有个现象值得说连接建立不是瞬间完成的而是有个爬坡过程。如果脚本里用 for 循环同步 connect5000 个连接会串行建立非常慢。改成多进程并发 connect 后建立时间缩短到几秒。但即便如此broker 侧也要处理大量的 SYN 和 CONNECT 包somaxconn和tcp_max_syn_backlog没调的话会看到明显的连接失败。5.3 吞吐量与 QoS 的关系QoS 0 下1000 连接、发布速率 10000 msg/s订阅端实测接收约 9800 msg/s丢失率 2%。这个丢失主要来自网络抖动和客户端处理不及时属于正常范围。QoS 1 下同样的参数接收速率降到约 8500 msg/s但丢失率接近 0。多出来的开销来自 PUBACK 确认包和 broker 侧的重传管理。QoS 1 的吞吐比 QoS 0 低 15% 到 20%这个比例和消息大小、网络延迟有关。连接数增加到 5000 时QoS 0 的吞吐反而下降到 7000 msg/s 左右。原因是客户端侧线程调度开销变大回调排队严重。这时候瓶颈已经从 broker 转移到客户端了。5.4 延迟分布与长尾问题延迟数据比吞吐更有意思。1000 连接、QoS 1、5000 msg/s 下分位延迟P503msP9512msP9945msMax320msP50 很漂亮但 P99 到了 45msMax 超过 300ms。这个长尾主要来自 Python 的 GIL 和线程调度。当某个回调执行时间稍长其他客户端的消息处理就被推迟。如果业务对延迟敏感Python 订阅端不是好选择得换 C 或 Go。5.5 瓶颈定位的排查思路压测数据不好看时怎么定位是 broker 的问题还是客户端的问题我的排查顺序是看 broker 侧 CPU 和内存。如果 broker CPU 没跑满瓶颈大概率在客户端。用mosquitto_sub命令行工具单独订阅看能不能收到全量消息。命令行工具性能比 Python 好如果它也丢说明 broker 有问题。减少订阅连接数看吞吐是否线性上升。如果连接数减半吞吐翻倍说明客户端侧有竞争。检查系统netstat的 Recv-Q 和 Send-Q看是否有大量积压。这套流程能快速区分是服务端还是客户端的问题避免瞎调参数。6. 常见问题与避坑经验实录6.1 连接被拒与 client_id 冲突最常见的报错是连接返回码 2identifier rejected或 5not authorized。返回码 2 基本就是 client_id 重复或超长。Mosquitto 默认 client_id 最大 23 字节超了会被拒。解决办法是在配置里加max_client_id_len 00 表示不限制或者把 client_id 缩短。返回码 5 是认证问题。压测环境记得开allow_anonymous true生产环境则要配用户名密码脚本里用client.username_pw_set()设置。6.2 消息堆积与内存暴涨订阅端处理慢时消息会在 broker 侧排队。max_queued_messages设得太大内存会暴涨甚至 OOM。我踩过一次坑设成 100000结果订阅端卡住broker 内存半小时涨到 4G 被系统杀掉。合理的做法是设一个适中的值比如 10000并且监控 broker 的$SYS/broker/messages/stored主题看堆积情况。订阅端也要做背压处理处理不过来时主动断开或降级。6.3 回调阻塞导致的假死on_message 里如果做了耗时操作比如写数据库、发 HTTP 请求整个客户端的消息处理会卡住。paho 的回调是在网络线程里同步执行的回调不返回后续消息就不处理。解决办法是把耗时操作丢到队列里由单独的消费者线程处理。回调里只做入队import queue msg_queue queue.Queue(maxsize10000) def on_message(client, userdata, msg): try: msg_queue.put_nowait(msg) except queue.Full: stats.incr_dropped()这样回调永远不阻塞但要注意队列满了之后的策略——是丢弃还是阻塞取决于业务。6.4 断线重连的处理压测跑久了偶尔会有连接断开。paho 有自动重连机制但默认行为不一定符合预期。loop_start()模式下断线后会自动重连但重连成功后会重新触发 on_connect如果 on_connect 里重复 subscribe会导致重复订阅。正确做法是在 on_connect 里判断是否是重连或者用client.subscribe的幂等性——MQTT 允许重复订阅同一主题broker 会覆盖不会重复推送。但保险起见还是在 on_connect 里统一处理订阅逻辑。6.5 常见问题速查表现象可能原因排查方向连接返回码 2client_id 重复或超长检查 client_id 生成逻辑连接返回码 5认证失败检查用户名密码或匿名配置消息大量丢失QoS 0 或队列溢出改 QoS 1调大 max_queued_messages延迟长尾严重回调阻塞或 GIL 竞争轻量化回调减少连接数broker 内存暴涨消息堆积监控 stored 主题设合理队列上限连接建立慢系统 backlog 不足调 somaxconn 和 syn_backlog吞吐上不去客户端单核跑满增加进程数分散连接6.6 几条压测的实操心得第一先小规模验证再放大。先用 10 个连接跑通全流程确认统计逻辑正确再逐步加到 100、1000、5000。一上来就压 5000出问题根本不知道是哪一层。第二发布端和订阅端分开机器。同机测试时两者抢资源数据参考价值有限。如果只有一台机器至少用 cgroup 限制各自的 CPU。第三关注 P99 而不是平均值。平均值好看不代表体验好长尾才是用户真正感知到的卡顿。第四压测时长要够。跑 10 秒和跑 10 分钟的结果可能完全不同内存泄漏、连接老化这些问题短时间测不出来。建议至少 5 分钟起步。第五记录完整的环境信息。CPU 型号、内存、内核版本、Mosquitto 版本、paho 版本这些都会影响结果。不记录的话换台机器数据对不上白测。7. 脚本的扩展方向这套脚本跑通之后可以往几个方向扩展。一是加 TLS测加密连接下的性能衰减通常 TLS 会让吞吐下降 30% 到 50%。二是加 MQTT 5.0 特性比如共享订阅、主题别名看新特性对性能的影响。三是接入监控把统计指标打到 Prometheus做长时间的稳定性观测。我个人在实际操作中的体会是订阅端压测最容易被低估的是客户端本身的资源消耗。很多人以为瓶颈在 broker结果一测发现是 Python 的 GIL 先扛不住。所以做容量规划时客户端和服务端的资源要一起算不能只盯着 broker 看。另外QoS 的选择对结果影响极大测试报告里一定要标明 QoS 等级否则数据没有可比性。