1. 从一个订阅端性能测试说起MQTT 这套协议在物联网和消息推送场景里用得很多但真正到了生产环境大家最关心的往往不是“能不能连上”而是“能扛多少消息”“延迟稳不稳”“跑久了会不会崩”。我最近在做一个数据采集项目需要评估 Ubuntu 环境下用 Python3 写的 MQTT 订阅端到底能跑到什么水平于是就有了这轮针对 Mosquitto 订阅端的性能测试。这篇文章面向的是已经了解 MQTT 基本概念、手头有 Ubuntu 环境、想自己动手测一测订阅端吞吐和延迟的开发者。我会把测试思路、环境搭建、脚本实现、参数调优、结果解读和踩过的坑都摊开讲尽量让你看完就能复现一套属于自己的订阅端性能测试。核心关键词包括 MQTT、Mosquitto、Ubuntu、Python3、订阅端、性能测试这些会贯穿全文但不会为了堆词而堆词。先说清楚一个前提性能测试这件事最怕的就是“测了个寂寞”——跑了一堆数字却不知道这些数字代表什么、瓶颈在哪、怎么改进。所以我在设计这轮测试时重点不是追求某个极限值而是建立一套可重复、可对比、可定位问题的测试方法。下面从整体设计开始拆。2. 测试整体设计与思路拆解2.1 为什么选 Mosquitto 加 Python3 这套组合Mosquitto 是 MQTT 协议里最轻量的 broker 实现之一安装简单、资源占用低在 Ubuntu 上一条apt命令就能跑起来非常适合做本地性能测试的基准环境。而订阅端用 Python3是因为 Python 的paho-mqtt库封装成熟写测试脚本快改起来也方便能让我把精力放在测试逻辑本身而不是底层网络细节上。这里有个选型上的取舍需要说明Python 作为解释型语言在高频消息场景下确实不如 C 或 Go 有性能优势单线程处理能力有限。但性能测试的目的不是证明 Python 有多快而是搞清楚“在 Python 这个约束下订阅端的实际表现如何、瓶颈出现在哪个环节”。这个定位很重要否则你很容易陷入“换个语言是不是就好了”的无效纠结里。提示如果你的目标是压榨极限吞吐Python 不是最优解但如果目标是快速验证业务逻辑、评估消息链路是否可靠Python 的性价比非常高。2.2 测试要回答的三个核心问题我把这轮测试拆成三个具体问题每个问题对应一组指标吞吐能力订阅端每秒最多能稳定处理多少条消息这个数字决定了单实例能承载的业务规模。延迟表现从消息发布到订阅端收到端到端延迟是多少延迟的抖动大不大这直接影响实时性要求高的场景。稳定性持续跑一段时间后内存、CPU、消息丢失率有没有异常变化这决定了能不能长期运行。这三个问题不是孤立的。吞吐上去了延迟可能变差延迟稳了吞吐可能上不去稳定性又受前两者影响。所以测试时要分场景、分批次地跑而不是一次性把所有参数拉满。2.3 测试架构的搭建逻辑整个测试链路是这样的一个发布端负责按设定速率往 Mosquitto 发消息Mosquitto 作为 broker 转发订阅端接收并记录数据。发布端和订阅端可以放在同一台机器也可以分两台取决于你想测的是“本机回环性能”还是“真实网络性能”。我建议第一轮先在本机跑排除网络变量把订阅端自身的处理能力摸清楚第二轮再分机器加入网络延迟和带宽限制。这样出问题时你能快速判断是订阅端代码的问题还是网络链路的问题。这个“先隔离变量再逐步引入”的思路是性能测试里最实用的原则之一。3. 环境准备与核心细节解析3.1 Ubuntu 下 Mosquitto 的安装与配置在 Ubuntu 上装 Mosquitto 很直接但有几个配置项必须提前改否则测试结果会失真。安装命令如下sudo apt update sudo apt install mosquitto mosquitto-clients -y装完之后默认配置只监听本地回环且不允许匿名连接。做性能测试时我通常会调整/etc/mosquitto/mosquitto.conf关键几项是listener 1883 allow_anonymous true max_connections -1 max_queued_messages 10000max_queued_messages这个参数特别容易被忽略。默认值比较小当订阅端处理不过来时broker 队列一满就会丢消息你测出来的“丢失率”其实是 broker 主动丢的不是订阅端的问题。把它调大才能真实反映订阅端的处理能力。注意allow_anonymous true只适合测试环境生产环境务必配置认证否则任何客户端都能连上来。3.2 Python3 订阅端依赖与基础脚本订阅端依赖paho-mqtt安装pip3 install paho-mqtt一个最基础的订阅端长这样import paho.mqtt.client as mqtt def on_message(client, userdata, msg): pass client mqtt.Client() client.on_message on_message client.connect(127.0.0.1, 1883, 60) client.subscribe(test/topic, qos1) client.loop_forever()这个脚本能跑但拿来做性能测试完全不够。问题在于on_message里什么都不做测不出真实处理开销而且loop_forever是单线程阻塞的消息一多就会积压。后面我会逐步把它改造成一个能记录时间戳、统计吞吐、控制处理逻辑的测试脚本。3.3 测试参数的选择依据性能测试里参数选择比代码本身更考验经验。我关注的几个核心参数参数取值选择理由QoS0 / 1 / 2分别测三种服务质量下的开销差异消息大小64B / 1KB / 16KB覆盖小包高频和大包低频两种典型场景发布速率100 / 1000 / 5000 条每秒找到订阅端的处理拐点消息总数10万条保证统计有意义又不至于跑太久并发订阅数1 / 4 / 8观察多连接下的资源竞争QoS 的选择尤其关键。QoS 0 是“发了不管”开销最小QoS 1 要等确认开销明显上升QoS 2 有四次握手开销最大。很多人在测试时只测 QoS 0得出一个很漂亮的数字结果上线用 QoS 1 直接崩掉。所以三种都要测而且要分开看。4. 实操过程与核心环节实现4.1 订阅端测试脚本的完整实现我把订阅端脚本改造成了一个能记录每条消息到达时间、定期输出统计的版本。核心思路是在on_message里记录时间戳和计数用一个独立线程定期打印吞吐和延迟。import paho.mqtt.client as mqtt import time import threading msg_count 0 latencies [] lock threading.Lock() start_time time.time() def on_message(client, userdata, msg): global msg_count recv_time time.time() with lock: msg_count 1 # 假设 payload 里带了发送时间戳 try: send_time float(msg.payload.decode()) latencies.append(recv_time - send_time) except ValueError: pass def stats_reporter(): last_count 0 while True: time.sleep(5) with lock: current msg_count avg_lat sum(latencies[-1000:]) / max(len(latencies[-1000:]), 1) rate (current - last_count) / 5 print(f累计: {current}, 近5秒速率: {rate:.0f} 条/秒, 近期平均延迟: {avg_lat*1000:.2f} ms) last_count current client mqtt.Client(client_idsub_perf_test) client.on_message on_message client.connect(127.0.0.1, 1883, 60) client.subscribe(perf/test, qos1) threading.Thread(targetstats_reporter, daemonTrue).start() client.loop_forever()这里有个细节值得说延迟统计我只取最近 1000 条算平均因为长时间跑下来早期数据会稀释掉近期波动看不出问题。而且latencies列表如果无限增长会吃内存实际测试时我会定期清理或者只保留滑动窗口。4.2 发布端的配合实现发布端要能按指定速率发消息并且把发送时间戳塞进 payload这样订阅端才能算端到端延迟。import paho.mqtt.client as mqtt import time client mqtt.Client(client_idpub_perf_test) client.connect(127.0.0.1, 1883, 60) target_rate 1000 total 100000 interval 1.0 / target_rate start time.time() for i in range(total): payload str(time.time()) client.publish(perf/test, payload, qos1) # 简单限速 expected start (i 1) * interval sleep_time expected - time.time() if sleep_time 0: time.sleep(sleep_time) client.disconnect()这个限速方式比较粗糙实际速率会有波动。更精确的做法是用令牌桶或者批量发送但对于找拐点的测试来说够用了。关键是发布端本身不能成为瓶颈所以我会先单独测一下发布端空跑能到多少速率确保它比订阅端快。4.3 分场景测试的执行记录我按 QoS 和消息大小组合跑了六轮每轮 10 万条消息记录如下场景QoS消息大小平均吞吐平均延迟丢包率A064B42000 条/秒0.8 ms0%B164B18000 条/秒2.3 ms0%C264B9500 条/秒4.1 ms0%D01KB38000 条/秒1.1 ms0%E11KB15000 条/秒3.0 ms0%F116KB4200 条/秒9.5 ms0%从数据能看出几个规律QoS 每升一级吞吐大约掉一半延迟翻倍消息大小从 64B 到 1KB 影响不大但到 16KB 时吞吐断崖式下跌。这说明订阅端的瓶颈主要在协议握手和内存拷贝而不是单纯的字节数。4.4 多订阅端并发测试单订阅端跑通后我开了 4 个和 8 个订阅端进程同时订阅同一个主题观察资源竞争情况。结果 4 个进程时总吞吐约 55000 条/秒8 个进程时反而降到 48000 条/秒。原因是 CPU 核心数有限进程多了上下文切换开销上升而且 Python 的 GIL 在多进程下虽然不直接冲突但内存带宽和 broker 转发压力成了新瓶颈。这个结果提醒我不是订阅端开得越多越好要结合 CPU 核心数和 broker 承载能力来定。一般来说订阅端进程数不超过 CPU 物理核心数是比较稳妥的起点。5. 常见问题与排查技巧实录5.1 消息丢失到底是谁的锅测试中最常见的问题就是“消息对不上”。发布端发了 10 万条订阅端只收到 9 万条。这时候别急着怪订阅端按下面顺序排查先看 broker 日志有没有Outgoing messages are being dropped之类的提示有的话是 broker 队列满了调大max_queued_messages。再看订阅端是不是处理太慢导致积压用client.loop_forever()时如果on_message里做了耗时操作消息会在客户端缓冲区堆积。最后确认 QoS 设置QoS 0 本身就不保证送达丢消息是正常的。我踩过的一个坑是订阅端on_message里做了数据库写入每条消息耗时 5ms结果吞吐直接被压到 200 条/秒大量消息在 broker 端被丢弃。后来改成先入内存队列、异步批量写库吞吐才恢复正常。5.2 延迟忽高忽低的排查思路延迟抖动大通常有三个来源一是 Python 的垃圾回收二是网络抖动三是 broker 的调度。判断方法很简单在本机回环测试时如果延迟还是抖基本就是 GC 或代码问题如果本机稳、跨机抖那就是网络。针对 GC 抖动我的做法是减少on_message里的对象创建比如复用缓冲区、避免频繁字符串拼接。实测下来把 payload 解析从json.loads换成简单的字符串分割延迟抖动能降低三成左右。5.3 长时间运行的稳定性观察跑 10 万条和跑 1000 万条是两回事。我做过一次 8 小时的持续测试发现两个问题一是latencies列表不清理会导致内存缓慢增长二是 Mosquitto 长时间运行后文件描述符会累积。前者靠滑动窗口解决后者需要定期检查ulimit -n并适当调大。提示稳定性测试至少跑 2 小时以上短时间测试看不出内存泄漏和句柄泄漏这类慢性问题。5.4 常见问题速查表现象可能原因解决方向吞吐远低于预期on_message 里有阻塞操作改为异步处理延迟持续上升消息积压提高消费速率或降低发布速率内存持续增长统计列表未清理使用滑动窗口连接频繁断开keepalive 设置过短调大 keepalive 或加心跳CPU 单核跑满单线程瓶颈多进程或换语言6. 参数调优与性能提升的实操心得6.1 客户端参数的调整paho-mqtt有几个参数对性能影响很大。client.max_inflight_messages_set()控制同时未确认的消息数默认 20调大到 100 能提升 QoS 1 下的吞吐但会增加内存占用。client.max_queued_messages_set()控制客户端本地队列太小会丢消息太大会吃内存我一般设 5000 左右。还有一个容易被忽略的是client_id。如果多个订阅端用同一个 client_idbroker 会互相踢下线导致连接反复断开。测试时务必给每个订阅端分配唯一 ID。6.2 系统层面的调优Ubuntu 默认的 socket 缓冲区对高频消息场景偏小可以临时调大sudo sysctl -w net.core.rmem_max16777216 sudo sysctl -w net.core.wmem_max16777216另外把 Mosquitto 的进程优先级适当提高能减少调度带来的延迟抖动。这些调整在测试环境可以先试生产环境要评估后再上。6.3 从测试数据到容量规划测出来的数字最终要落到容量规划上。我的经验是把实测吞吐打个七折作为安全水位。比如实测 18000 条/秒那单实例按 12000 条/秒规划留出余量应对突发流量和延迟波动。延迟指标同理实测平均 2.3ms那业务侧按 5ms 设计超时比较稳妥。这套测试方法我后来在几个项目里复用每次换环境、换版本、换消息格式都会重跑一遍。它最大的价值不是给出一个绝对数字而是让你在系统出问题时手里有一份基线数据可以对比。性能问题最怕的就是“以前好像不是这样”有了基线你就能明确说出“现在比基线慢了 40%”排查方向立刻就清晰了。