企业官网建设流程全解析

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.conf：

listener 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_messages：QoS 1/2 下同时处于“已发送未确认”状态的消息数。这个值影响吞吐和内存的平衡。

改完重启：

sudo systemctl restart mosquitto

3.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 文件里单独配LimitNOFILE=65535，或者用systemctl edit mosquitto加 override。

TCP 相关参数。大量短连接或频繁重连时，TIME_WAIT 会堆积。编辑/etc/sysctl.conf：

net.core.somaxconn = 32768 net.ipv4.tcp_max_syn_backlog = 32768 net.ipv4.tcp_tw_reuse = 1 net.ipv4.ip_local_port_range = 1024 65535

tcp_tw_reuse允许复用 TIME_WAIT 状态的连接，对压测场景很有用。改完sudo sysctl -p生效。

提示：这些参数在生产环境要谨慎，尤其是tcp_tw_reuse，某些场景下可能引发连接异常。压测环境可以放开，生产环境建议只调 somaxconn 和 backlog。

3.3 Python 环境与依赖安装

Python3 一般系统自带。装 paho-mqtt：

pip3 install paho-mqtt

建议用虚拟环境隔离，避免污染系统 Python：

python3 -m venv mqtt-test source mqtt-test/bin/activate pip install paho-mqtt

paho-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(f"connected with result code {rc}") client.subscribe("test/topic", qos=1) def on_message(client, userdata, msg): print(f"received: {msg.payload.decode()} on {msg.topic}") client = mqtt.Client(client_id="sub_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 = f"sub_{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 = f"sub_{proc_id}_{i}" client = mqtt.Client(client_id=cid) 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(target=worker, 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 += 1

get_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, qos=1) seq += 1 time.sleep(interval)

rate是目标发送速率。如果要测极限吞吐，就去掉 sleep，用循环狂发，但要注意发布端自己别成为瓶颈。发布端建议用 C 或 Go 写，或者多进程 Python，否则它先撑不住，测出来的就是发布端的能力。

注意：发布端和订阅端如果跑在同一台机器上，会互相抢 CPU 和网络，测出来的数据偏悲观。有条件的话分两台机器，用千兆或万兆内网连接。

5. 实测数据与性能瓶颈分析

5.1 测试环境与参数

我的测试环境是一台 4 核 8G 的 Ubuntu 22.04 云主机，Mosquitto 2.0.x，paho-mqtt 1.6.1。测试参数如下：

5.2 连接数对资源的影响

先看连接建立阶段。1000 连接时，Mosquitto 进程内存约 80MB，CPU 空闲。3000 连接时内存涨到 200MB 左右。5000 连接时内存约 350MB，且连接建立时间明显变长，从发起到全部连上花了将近 20 秒。

这里有个现象值得说：连接建立不是瞬间完成的，而是有个爬坡过程。如果脚本里用 for 循环同步 connect，5000 个连接会串行建立，非常慢。改成多进程并发 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 下：

P50 很漂亮，但 P99 到了 45ms，Max 超过 300ms。这个长尾主要来自 Python 的 GIL 和线程调度。当某个回调执行时间稍长，其他客户端的消息处理就被推迟。如果业务对延迟敏感，Python 订阅端不是好选择，得换 C 或 Go。

5.5 瓶颈定位的排查思路

压测数据不好看时，怎么定位是 broker 的问题还是客户端的问题？我的排查顺序是：

1. 看 broker 侧 CPU 和内存。如果 broker CPU 没跑满，瓶颈大概率在客户端。
2. 用mosquitto_sub命令行工具单独订阅，看能不能收到全量消息。命令行工具性能比 Python 好，如果它也丢，说明 broker 有问题。
3. 减少订阅连接数，看吞吐是否线性上升。如果连接数减半吞吐翻倍，说明客户端侧有竞争。
4. 检查系统netstat的 Recv-Q 和 Send-Q，看是否有大量积压。

这套流程能快速区分是服务端还是客户端的问题，避免瞎调参数。

6. 常见问题与避坑经验实录

6.1 连接被拒与 client_id 冲突

最常见的报错是连接返回码 2（identifier rejected）或 5（not authorized）。返回码 2 基本就是 client_id 重复或超长。Mosquitto 默认 client_id 最大 23 字节，超了会被拒。解决办法是在配置里加max_client_id_len 0（0 表示不限制），或者把 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(maxsize=10000) 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 常见问题速查表

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 等级，否则数据没有可比性。