Raspberry Pi CM4上Python Paho MQTT离线队列监控及断电处理问询

1. 获取离线队列长度 #

Paho MQTT v2客户端内置了直接获取队列长度的方法，你可以通过get_queued_messages_count()实时获取当前离线队列中的消息数量：

# 获取当前队列长度
queue_length = mqtt_client.get_queued_messages_count()
print(f"当前离线队列消息数: {queue_length}")

如果需要持续监控，可结合定时器（如time.sleep()循环或threading.Timer）定期查询并记录。

2. 断电时的队列处理 #

由于断电是突发场景，核心思路是避免队列消息丢失+在可控关机信号下完成收尾处理：

方案一：启用Paho内置文件持久化

Paho v2支持将离线队列消息持久化到本地文件，即使意外断电，重启客户端后会自动读取并发送队列中的消息。使用方式如下：

from paho.mqtt.client import Client, MQTTv5
from paho.mqtt.persistence import FilePersistence

# 指定持久化存储目录（确保该目录有读写权限）
persistence_dir = "/home/pi/mqtt_persistence"
persistence = FilePersistence(persistence_dir)

# 初始化客户端时绑定持久化实例
mqtt_client = Client(client_id="cm4_mqtt_client", protocol=MQTTv5, persistence=persistence)

# 设置最大队列长度
mqtt_client.max_queued_messages_set(100)

方案二：捕获关机信号做自定义处理

对于正常关机（如执行shutdown命令），系统会发送SIGTERM信号，你可以注册信号处理函数，在收到信号时完成队列的备份或收尾操作：

import signal
import sys

def handle_shutdown(signum, frame):
    print("触发关机信号，开始处理离线队列")
    queue_len = mqtt_client.get_queued_messages_count()
    
    # 自定义处理逻辑示例：记录队列状态到本地日志
    with open("/home/pi/mqtt_queue_shutdown.log", "w") as f:
        f.write(f"关机时队列剩余消息数: {queue_len}\n")
    
    # 断开连接，确保持久化数据写入磁盘
    mqtt_client.disconnect()
    sys.exit(0)

# 注册信号监听
signal.signal(signal.SIGINT, handle_shutdown)  # 捕获Ctrl+C
signal.signal(signal.SIGTERM, handle_shutdown) # 捕获系统关机信号

注意：意外断电（如直接拔电源）时系统不会触发信号处理，此时只能依赖文件持久化机制保留队列消息。

3. 额外注意事项 #

• 确保持久化目录的读写权限，树莓派上建议使用当前用户拥有权限的目录（如/home/pi/下的子目录）。
• 当队列达到max_queued_messages_set()设置的上限时，Paho会自动丢弃最早的消息，可通过监控队列长度提前触发告警或扩容处理。

内容的提问来源于stack exchange，提问作者Luigi Juliano Di Leo