Python MQTT：on_subscribe重复触发及loop struct格式错误问题

核心问题根源 #

你同时使用了loop_start()（开启后台线程自动处理MQTT网络循环）和手动调用loop()，这两种方式不能同时存在——它们会同时操作客户端的网络缓冲区和报文处理逻辑，导致重复触发回调、报文结构损坏，最终出现bad char in struct format错误。

另外，原代码中先调用connect再设置回调的顺序也有问题，可能导致首次连接时回调未生效，加重异常情况。

解决方案二选一 #

方案1：保留loop_start()，移除手动loop()调用

先设置所有回调，再连接并启动后台循环，后续无需手动调用loop()：

# 先设置回调函数，再执行连接
self.client.on_connect = self.onConnect
self.client.on_publish = self.onPublish
self.client.on_subscribe = self.on_subscribe

self.client.connect(self.broker, self.port, 60)
self.client.loop_start()

def run(self):
    while not self.client.is_connected():
        time.sleep(1)

    self.running = True
    while(self.running):
        row = self.db.getMessage()
        if(row != None):
            message = row[2]
            topic = 'data'
            timestamp = strftime("%Y-%m-%d %H:%M:%S", localtime())
            try:
                ret = self.client.publish(topic, message, 0)
            except Exception as ex:
                print(str(ex))
        # 移除手动loop调用，添加短延迟避免空循环占用CPU
        time.sleep(0.1)

方案2：弃用loop_start()，全程手动调用loop()

不开启后台线程，在循环中手动调用loop()处理网络事件：

# 先设置回调函数
self.client.on_connect = self.onConnect
self.client.on_publish = self.onPublish
self.client.on_subscribe = self.on_subscribe

self.client.connect(self.broker, self.port, 60)

def run(self):
    while not self.client.is_connected():
        # 连接过程也要调用loop处理网络事件
        self.client.loop(0.1)

    self.running = True
    while(self.running):
        row = self.db.getMessage()
        if(row != None):
            message = row[2]
            topic = 'data'
            timestamp = strftime("%Y-%m-%d %H:%M:%S", localtime())
            try:
                ret = self.client.publish(topic, message, 0)
            except Exception as ex:
                print(str(ex))
        # 持续调用loop处理MQTT报文
        self.client.loop(0.1)

效果说明 #

• 调整后，on_subscribe只会触发一次，因为客户端只有一个网络循环处理订阅确认报文。
• 不再出现bad char in struct format错误，因为不会有两个线程/循环同时破坏内部报文缓冲区。

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