MQTT相关Docker容器运行时冻结延迟问题求助

我搭建了三个Docker容器：

• 基于Python paho-mqtt的消息发布容器：每秒通过MQTT协议发送消息
• 消息订阅容器：接收MQTT消息
• Mosquitto MQTT服务器容器

容器构建命令 #

docker image build -t publisher /path/to/dockerfile folder
docker image build -t subscriber /path/to/dockerfile folder

发布者容器Dockerfile #

FROM python:3.9.7

WORKDIR /app

COPY requirements.txt requirements.txt
RUN pip3 install -r requirements.txt

EXPOSE 1883

COPY . .

CMD [ "python", "app.py" ]

Mosquitto服务器日志（运行正常） #

1658502881: mosquitto version 2.0.14 running
1658502881: New connection from 172.17.0.5:53757 on port 1883.
1658502881: New client connected from 172.17.0.5:53757 as python-mqtt-29 (p2, c1, k60).
1658502881: New connection from 172.17.0.4:45437 on port 1883.
1658502881: New client connected from 172.17.0.4:45437 as python-mqtt-12 (p2, c1, k60).

异常现象 #

运行docker run publisher或订阅者容器时，终端会冻结2-3分钟，之后所有消息突然批量显示。

附代码 #

发布者应用代码

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

broker = '172.17.0.3'
port = 1883
topic = "python/mqtt"

client_id = f'python-mqtt-{random.randint(0, 1000)}'

def connect_mqtt():
    def on_connect(client, userdata, flags, rc):
        if rc == 0:
            print("Connected to MQTT Broker!")
        else:
            print("Failed to connect, return code %d\n", rc)

    client = mqtt_client.Client(client_id)
    client.on_connect = on_connect
    client.connect(broker, port)
    return client

def publish(client):
    msg_count = 0
    while True:
        time.sleep(1)
        msg = f"messages: {msg_count}"
        result = client.publish(topic, msg)
        # result: [0, 1]
        status = result[0]
        if status == 0:
            print(f"Send `{msg}` to topic `{topic}`")
        else:
            print(f"Failed to send message to topic {topic}")
        msg_count += 1

def run():
    client = connect_mqtt()
    client.loop_start()
    publish(client)

if __name__ == '__main__':
    run()

订阅者应用代码

import random
import pandas
from paho.mqtt import client as mqtt_client

broker = '172.17.0.3'
port = 1883
topic = "python/mqtt"
client_id = f'python-mqtt-{random.randint(0, 100)}'


def connect_mqtt() -> mqtt_client:
    def on_connect(client, userdata, flags, rc):
        if rc == 0:
            print("Connected to MQTT Broker!")
        else:
            print("Failed to connect, return code %d\n", rc)

    client = mqtt_client.Client(client_id)
    client.on_connect = on_connect
    client.connect(broker, port)
    return client


def subscribe(client: mqtt_client):
    def on_message(client, userdata, msg):
        print(f"Received `{msg.payload.decode()}` from `{msg.topic}` topic")

    client.subscribe(topic)
    client.on_message = on_message


def run():
    client = connect_mqtt()
    subscribe(client)
    client.loop_forever()


if __name__ == '__main__':
    run()

核心原因：Python输出缓冲延迟 #

Python的print函数默认会缓冲输出，当容器运行时stdout不是交互式终端（TTY），缓冲会累积到一定阈值才一次性输出，这就是你看到2-3分钟后批量显示的根本原因。

具体解决方法 #

方法1：启动容器时分配TTY

运行容器时添加-it参数，强制Docker分配交互式终端，Python会自动禁用输出缓冲：

docker run -it publisher
docker run -it subscriber

方法2：修改代码强制刷新输出

在所有print语句后添加flush=True参数，实时刷新输出缓冲：

# 发布者代码修改
print(f"Send `{msg}` to topic `{topic}`", flush=True)
print(f"Failed to send message to topic {topic}", flush=True)

# 订阅者代码修改
print(f"Received `{msg.payload.decode()}` from `{msg.topic}` topic", flush=True)

方法3：修改Dockerfile禁用Python缓冲

在启动命令中添加-u参数，强制Python使用无缓冲输出：

CMD [ "python", "-u", "app.py" ]

额外排查点：容器网络延迟 #

如果以上方法无效，可检查容器间网络连通性：

1. 进入发布者容器ping Mosquitto服务器IP：

docker exec -it <publisher-container-id> ping 172.17.0.3

若延迟过高，建议改用Docker自定义网桥替代默认bridge网络，避免IP冲突或网络拥堵。

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