基于Python Paho-MQTT与Mosquitto的MQTTS发布订阅连接问题排查求助

环境说明 #

• 本地Ubuntu机器，通过snap安装Mosquitto
• 已配置TLS/SSL，MQTTS运行在默认端口8883
• 无TLS的MQTT发布订阅功能正常，切换到MQTTS后出现异常

问题现象 #

1. 发布端异常 #

运行脚本后前1-5条消息可成功发布，之后连接断开，剩余消息发布失败。

发布端脚本：

import time
import paho.mqtt.client as mqtt

messaging_protocol = 'MQTTS'
ca_file = '/home/tp/mosquitto_poc/steves-internet/ca.crt'
broker_address = 'localhost'
topic = 'topic1'

client = mqtt.Client('publisher-client')

if messaging_protocol == 'MQTTS':
    port = 8883
    client.tls_set(ca_file)
else:
    # When messaging_protocol will be 'MQTT'
    port = 1883

client.connect(broker_address, port)

for i in range(10):
    payload = f'Message no. {i+1}'
    print(f'Attempting to publish message: `{payload}`')
    mqtt_msg_info = client.publish(topic, payload)
    try:
        is_published = mqtt_msg_info.is_published()
    except RuntimeError as e:
        print(f'Error occured: `{str(e)}`. Rc value: {mqtt_msg_info.rc}.', end='\n\n')
    except ValueError as e:
        print(f'Error occured: `{str(e)}`. Rc value: {mqtt_msg_info.rc}. Error string: `{mqtt.error_string(mqtt_msg_info.rc)}`', end='\n\n')
    else:
        print(f'Message published: {is_published}. Rc value: {mqtt_msg_info.rc}. Error string: `{mqtt.error_string(mqtt_msg_info.rc)}`', end='\n\n')
    finally:
        time.sleep(0.005)

发布端运行输出：

Attempting to publish message: `Message no. 1`
Message published: True. Rc value: 0. Error string: `No error.`

Attempting to publish message: `Message no. 2`
Message published: True. Rc value: 0. Error string: `No error.`

Attempting to publish message: `Message no. 3`
Message published: True. Rc value: 0. Error string: `No error.`

Attempting to publish message: `Message no. 4`
Error occured: `Message publish failed: The connection was lost.`. Rc value: 7.

Attempting to publish message: `Message no. 5`
Error occured: `Message publish failed: The client is not currently connected.`. Rc value: 4.

Attempting to publish message: `Message no. 6`
Error occured: `Message publish failed: The client is not currently connected.`. Rc value: 4.

Attempting to publish message: `Message no. 7`
Error occured: `Message publish failed: The client is not currently connected.`. Rc value: 4.

Attempting to publish message: `Message no. 8`
Error occured: `Message publish failed: The client is not currently connected.`. Rc value: 4.

Attempting to publish message: `Message no. 9`
Error occured: `Message publish failed: The client is not currently connected.`. Rc value: 4.

Attempting to publish message: `Message no. 10`
Error occured: `Message publish failed: The client is not currently connected.`. Rc value: 4.

2. 订阅端异常 #

运行脚本时on_connect返回rc=5，提示“Connection refused - not authorised”，反复断开重连。

订阅端脚本：

import paho.mqtt.client as mqtt

messaging_protocol = 'MQTTS'
ca_file = '/home/tp/mosquitto_poc/steves-internet/ca.crt'
broker_address = 'localhost'
topic = 'topic1'


def on_connect(client, userdata, flags, rc):
    print(f'Connected with result code {str(rc)}')
    result, _ = client.subscribe(topic)
    print(f'Attempting to subscribe topic: `{topic}`. Result value: {result}. Error string: `{mqtt.error_string(result)}`')


def on_message(client, userdata, msg):
    print(f'Received message: `{msg.payload}` on topic: {msg.topic}')


def on_disconnect(client, userdata, rc):
    print(f'Client disconnected. Rc value: {rc}. Error string: `{mqtt.error_string(rc)}`', end='\n\n')


client = mqtt.Client('subscriber-client')

if messaging_protocol == 'MQTTS':
    port = 8883
    client.tls_set(ca_file)
else:
    # When messaging_protocol will be 'MQTT'
    port = 1883

client.on_connect = on_connect
client.on_message = on_message
client.on_disconnect = on_disconnect

client.connect(broker_address, port)

client.loop_forever()

订阅端运行输出：

Connected with result code 5
Attempting to subscribe topic: `topic1`. Result value: 0. Error string: `No error.`
Client disconnected. Rc value: 5. Error string: `The connection was refused.`

Connected with result code 5
Attempting to subscribe topic: `topic1`. Result value: 0. Error string: `No error.`
Client disconnected. Rc value: 5. Error string: `The connection was refused.`

Connected with result code 5
Attempting to subscribe topic: `topic1`. Result value: 0. Error string: `No error.`
Client disconnected. Rc value: 5. Error string: `The connection was refused.`

Connected with result code 5
Attempting to subscribe topic: `topic1`. Result value: 0. Error string: `No error.`
Client disconnected. Rc value: 5. Error string: `The connection was refused.`

Connected with result code 5
Attempting to subscribe topic: `topic1`. Result value: 0. Error string: `No error.`
Client disconnected. Rc value: 5. Error string: `The connection was refused.`

排查与修复方案 #

1. 订阅端rc=5（未授权）问题 #

rc=5表示连接被Broker拒绝，核心原因是Mosquitto配置要求客户端提供认证凭证，但脚本未配置：

• 检查Mosquitto配置：
  打开snap安装的配置文件（路径通常为/var/snap/mosquitto/common/mosquitto.conf），查看以下项：
  • 若存在require_certificate true：需在客户端脚本中添加客户端证书和私钥：

    client.tls_set(ca_file, certfile="/path/to/client.crt", keyfile="/path/to/client.key")
    

  • 若存在allow_anonymous false或password_file：需在客户端连接前设置用户名密码：

    client.username_pw_set("your_username", "your_password")
    

  • 确保listener 8883配置完整，包含证书路径：

    listener 8883
    cafile /path/to/ca.crt
    certfile /path/to/server.crt
    keyfile /path/to/server.key
    

2. 发布端连接断开问题 #

发布端未启动MQTT客户端循环，无法处理TLS连接的心跳与后台通信，导致Broker主动断开连接：

• 修复发布端脚本：
  在client.connect后启动后台循环，发布完成后停止：

  client.connect(broker_address, port)
  client.loop_start()  # 启动后台线程处理网络事件
  
  for i in range(10):
      # 原发布逻辑保持不变
      ...
  
  client.loop_stop()  # 发布完成后停止循环
  

  或在每次发布后调用client.loop()处理事件：

  for i in range(10):
      # 原发布逻辑
      ...
      client.loop()  # 处理网络事件
      time.sleep(0.005)
  

3. 通用验证步骤 #

• 用Mosquitto官方工具测试MQTTS连接，排除脚本问题：

  # 测试发布
  mosquitto_pub -h localhost -p 8883 -t topic1 -m "test" --cafile /home/tp/mosquitto_poc/steves-internet/ca.crt
  # 测试订阅
  mosquitto_sub -h localhost -p 8883 -t topic1 --cafile /home/tp/mosquitto_poc/steves-internet/ca.crt
  

• 查看Mosquitto日志（路径通常为/var/snap/mosquitto/common/log/mosquitto.log），获取详细错误信息（如证书验证失败、客户端认证失败等）。

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