Docker部署Mosquitto MQTT Broker性能测试及优化咨询

测试场景 #

场景1 #

3台Docker容器化部署的机器，以每秒100条记录、QoS=1的速率，向Docker容器部署的MQTT Broker并行发送30秒消息。每台机器总计发送3000条消息，单条消息大小为260字节。

接收端统计数据（每台机器的接收记录数）：

JSON_DATE:MACHINE_NAME                COUNT(*)
"09166423-892a-4c84-9717-83b91ecbf2d5"  1,956
"b05eb890-ed0a-4272-aa95-7dce4f066cdb"  2,037
"e1ede903-f8f5-4bc8-95bb-ee286e6c00c3"  2,419

推论：每台机器均存在数据丢失

场景2 #

5台Docker容器化部署的机器，每5秒发送100条记录、QoS=1，向Docker容器部署的MQTT Broker并行发送30秒消息。每台机器总计发送3000条消息，单条消息大小为260字节。

接收端统计数据：

JSON_DATE:MACHINE_NAME                 COUNT(*)
"043fc770-104c-4d0f-99e8-aae4a22bea60"  3,721
"06d39ec0-d799-43a9-9c58-16bbb67c0217"  3,085
"4374073b-6c75-4788-b1f7-bcd65aa755aa"  3,587
"675467d5-4361-4a61-becf-0a20fc5ad84f"  3,500
"825d3880-99db-425b-b867-3254fef394f3"  3,495

推论：所有机器均无数据丢失，多数机器存在消息重复交付

场景3 #

10台Docker容器化部署的机器，每5秒发送100条记录、QoS=1，向Docker容器部署的MQTT Broker并行发送30秒消息。每台机器总计发送3000条消息，单条消息大小为260字节。

接收端统计数据：

JSON_DATE:MACHINE_NAME                 COUNT(*)
"06e08aeb-f258-40eb-adcd-efe1b2e3725e"  3,162
"0be5975c-b364-450e-b550-62834d32a18f"  3,181
"30f6cea9-efde-42b5-bcb2-b81e360d27b4"  2,959
"500465c1-7ff0-4097-a0d7-d22b55b73079"  3,142
"ac61be0c-ac40-44fd-930c-10b01d98aa4a"  3,235
"b259f4cd-cd69-4582-b428-53192ac5d13e"  3,027
"b8567bf2-cdc8-4907-8c03-7bbe78d36cb8"  3,212
"ca3bda96-4b60-4d97-a88a-22b10d500650"  2,932
"d6bb5650-1fdb-45cf-9bfc-a979594aba0f"  3,287
"dff45255-b48e-45a9-9e71-a7cfd9559139"  3,069

推论：部分机器存在轻微数据丢失

场景2的Docker资源占用情况 #

相关代码 #

发布者代码（运行于Docker容器内） #

## Publisher Code. This is run within a docker container

import paho.mqtt.client as paho 
from paho import mqtt
from random import randrange, uniform
import time
from time import sleep
from multiprocessing import Process
from os import getpid
import os
import random
import datetime
import os
import uuid

def task():
    pid = getpid()
    machine_name =  str(uuid.uuid4())
    topic_name =  machine_name
    print("{0} : {1}".format(machine_name, topic_name))

    for i in range(30) :

        mqttBroker ="localhost" 
        client = paho.Client("Temperature_Inside")
        client.connect(mqttBroker,1883)
        client.loop_start()

        for _ in range(100):
            sensor_reading = str(random.uniform(10, 12))
            ingest_time = str(datetime.datetime.now())
            sensor_payload = "{ 'Topic_Name': \'" + topic_name + "\', 'Msg_Id': \'" + str(i)+ '-'+ str(_) + "\', 'Ingestion_Time': \'" + ingest_time + "\', 'Machine_Name': \'" + machine_name + "\','Reading': \'"+ sensor_reading +"\' }"
            print(sensor_payload)
            client.publish(topic_name,sensor_payload,qos=1)
            
        sleep(5)

# entry point
if __name__ == '__main__':
    sleep(1)
    task()

Docker容器触发代码 #

## Code to trigger the docker container

from multiprocessing import Process
import os
import subprocess
import docker
import sys

def task():
    client = docker.from_env()
    container = client.containers.run('datagen', detach=True,network="host",auto_remove=True)
    print(container.logs())

# entry point
if __name__ == '__main__':
    print("Number of threads to start : {0}".format(sys.argv[1]))
    n=int(sys.argv[1])
    for i in range(n):
        process = Process(target=task)
        process.start()

订阅者代码 #

## Finally here the subscriber code

import paho.mqtt.client as paho 
from paho import mqtt
import time

def on_message(client, userdata, message):
    print("received message: " ,str(message.payload.decode("utf-8")))
    with open('files_sub/data_recv.txt','a+') as f:
         f.write("Message received: "  + str(message.payload) + "\n")

mqttBroker ="localhost"
client = paho.Client("Smartphone")
client.connect(mqttBroker,1883) 

client.loop_start()
client.subscribe("#",qos=1)
client.on_message=on_message 

time.sleep(300)
client.loop_stop()

待解答问题 #

1. 是否存在可作为Docker部署MQTT Broker性能基准的阈值？
2. 有哪些方法可以提升该部署方式下的性能？

问题解答 #

1. Docker部署MQTT Broker的性能基准阈值 #

不存在通用的性能基准阈值，性能表现受多维度因素影响：

• Broker选型：不同MQTT Broker的性能基线差异极大。例如Mosquitto单节点在Docker环境下，QoS=1时可达每秒数万条消息吞吐量；EMQX集群模式则能支撑每秒数十万级别的消息处理。
• 硬件资源：宿主机的CPU核心数、内存大小、磁盘IO、网络带宽直接决定Broker性能上限。CPU核心不足会导致消息处理阻塞，内存不够会引发Broker频繁GC或崩溃。
• 配置参数：Broker的连接数限制、消息队列长度、QoS级别、持久化设置等都会改变性能表现。开启消息持久化会降低吞吐量，但提升可靠性。
• 场景特性：消息大小、并发客户端数、消息分发模式（广播/点对点）也会影响基准值。

实际场景中，建议基于自身业务需求，参考所选Broker官方性能测试报告，结合自身硬件环境压测，得到符合自身场景的基准阈值。

2. Docker部署MQTT Broker的性能优化方法 #

（1）Broker层面优化

• 选择高性能Broker：替换轻量型的Mosquitto为EMQX、VerneMQ这类专为高并发设计的Broker，它们支持集群扩展，性能上限更高。
• 调整Broker配置：
  • 关闭不必要的持久化（非关键业务无需存储消息），减少磁盘IO开销；
  • 增大消息队列缓冲区大小，避免消息因队列满被丢弃；
  • 调整连接超时、心跳间隔参数，减少无效连接占用资源；
  • 开启批量消息处理，降低单条消息的处理开销。
• 集群化部署：单节点性能不足时，搭建Broker集群，通过负载均衡分散客户端连接和消息流量。

（2）Docker部署层面优化

• 资源限制与预留：为Broker容器分配足够的CPU和内存资源（如--cpus=4 --memory=8g），避免容器被Docker调度限制资源使用；同时开启CPU绑定（--cpuset-cpus），减少CPU上下文切换开销。
• 网络优化：
  • 使用主机网络模式（--network=host）替代默认桥接网络，减少网络转发的性能损耗；
  • 优化宿主机TCP参数（如增大TCP缓冲区），提升网络传输效率。
• 存储优化：若需持久化消息，使用SSD挂载到容器的持久化目录，避免Docker默认overlay2存储驱动的性能损耗；或采用外部数据库（如Redis、PostgreSQL）存储消息，提升读写效率。
• 容器镜像优化：使用精简的Broker镜像（如alpine基础镜像），减少镜像体积和容器启动时间，降低资源占用。

（3）客户端代码优化

• 复用MQTT连接：当前发布者代码每次循环都创建新连接，频繁连接建立会消耗大量资源。应在进程启动时创建一次连接，复用该连接发送所有消息。
• 异步发送与流量控制：使用异步发布接口，避免同步发送导致的阻塞；实现客户端侧流量控制，根据Broker反馈调整发送速率，避免Broker被消息压垮。
• 减少冗余操作：发布者代码中每次发送都打印消息内容，订阅者每次接收都写入文件并打印，生产环境应关闭调试输出，或优化为批量写入文件，减少IO开销。

内容的提问来源于stack exchange，提问作者Ritab