Python中Request阻塞MQTT消息发布的问题解决咨询

核心思路 #

必须把耗时的API请求逻辑完全从MQTT客户端的回调线程/主循环中剥离，避免阻塞MQTT的心跳、发布等核心操作。

具体可行方案 #

1. 独立线程池+非阻塞MQTT配置 #

• 用Paho MQTT的loop_start()启动独立网络线程，让MQTT的消息处理、发布、心跳在单独线程执行，不被主线程阻塞操作影响。
• 把dotherequest放入线程池执行，回调只做任务提交，绝不等待结果：

  from concurrent.futures import ThreadPoolExecutor
  import paho.mqtt.client as mqtt
  
  class TestClassName:
      def __init__(self):
          self.client = mqtt.Client()
          self.client.on_message = self.on_message
          # 启动MQTT独立线程
          self.client.loop_start()
          # 初始化线程池，控制并发数
          self.executor = ThreadPoolExecutor(max_workers=3)
  
      def on_message(self, client, userdata, msg):
          # 仅提交任务，不等待结果，避免阻塞回调
          self.executor.submit(self.dotherequest, msg.payload)
  
      def dotherequest(self, payload):
          # 耗时5秒的API请求逻辑
          result = requests.get("your-api-url")
          # API返回后再发布数据，Paho客户端本身线程安全
          self.uiPublish(result.json())
  
      def uiPublish(self, data):
          self.client.publish("topic/ui", str(data))
  

2. 全异步链路（异步MQTT+异步HTTP） #

如果之前用aiohttp没解决，大概率是MQTT客户端还是同步的，导致异步请求卡在同步回调里。改用asyncio-mqtt配合aiohttp实现全异步：

import asyncio
from asyncio_mqtt import Client
import aiohttp

class TestClassName:
    def __init__(self):
        self.mqtt_sub_topic = "topic/input"
        self.mqtt_pub_topic = "topic/ui"
        self.session = aiohttp.ClientSession()

    async def mqtt_listener(self):
        async with Client("broker-address") as client:
            await client.subscribe(self.mqtt_sub_topic)
            async for message in client.messages:
                # 异步提交API任务，不阻塞监听循环
                asyncio.create_task(self.dotherequest(message.payload))

    async def dotherequest(self, payload):
        async with self.session.get("your-api-url") as resp:
            result = await resp.json()
            # 异步发布API结果
            async with Client("broker-address") as client:
                await client.publish(self.mqtt_pub_topic, str(result))

    async def ui_publish_loop(self):
        # 每秒发布数据的异步循环
        async with Client("broker-address") as client:
            while True:
                data = self.get_cached_data()
                await client.publish(self.mqtt_pub_topic, str(data))
                await asyncio.sleep(1)

    def get_cached_data(self):
        # 获取要发布的旧数据逻辑
        return {"cached_key": "cached_value"}

# 启动所有异步任务
async def main():
    app = TestClassName()
    await asyncio.gather(
        app.mqtt_listener(),
        app.ui_publish_loop()
    )

asyncio.run(main())

3. 排查之前方案失效的常见原因 #

如果试过线程/ThreadPool还是阻塞，可能是这些问题：

• 误用loop_forever()而非loop_start()，导致MQTT主循环和API请求在同一线程。
• 在回调中调用executor.submit()后又用result()等待返回，强行阻塞回调线程。
• 线程池max_workers设置过小，导致任务排队阻塞。

额外注意事项 #

• Paho MQTT客户端的发布操作本身线程安全，但如果有共享数据需要修改后发布，要加threading.Lock()保证线程安全。
• 定时发布的uiPublish()要放在独立线程/异步任务中执行，不和MQTT回调、API请求共用执行流。

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