树莓派Pico W（MicroPython）：umqtt.wait_msg()时板载LED闪烁实现咨询

1. 使用uasyncio（推荐，适配MicroPython异步场景） #

umqtt.wait_msg()是阻塞式方法，用uasyncio可以将MQTT消息处理和LED拆分为两个异步任务，避免互相阻塞。

具体实现：

• 用非阻塞的check_msg()替代wait_msg()，在异步循环中轮询消息
• 单独定义LED闪烁的异步任务，通过await让出执行权

代码示例：

import uasyncio
from umqtt.simple import MQTTClient
from machine import Pin

# 配置板载LED引脚（ESP32一般为2，ESP8266按需调整）
led = Pin(2, Pin.OUT)
# MQTT服务器配置
MQTT_BROKER = "你的Broker地址"
CLIENT_ID = "micropython_device"

async def blink_led():
    while True:
        led.value(not led.value())
        await uasyncio.sleep(0.5)  # 0.5秒间隔闪烁

async def mqtt_task():
    client = MQTTClient(CLIENT_ID, MQTT_BROKER)
    client.connect()
    print("已连接MQTT服务器")
    
    while True:
        client.check_msg()  # 非阻塞检查新消息
        await uasyncio.sleep(0.1)  # 给LED任务让出执行时间

# 启动异步任务
loop = uasyncio.get_event_loop()
loop.create_task(blink_led())
loop.run_until_complete(mqtt_task())

2. 使用_thread（线程并发，适合简单场景） #

利用MicroPython的_thread模块创建独立线程运行LED闪烁逻辑，主线程正常执行wait_msg()阻塞等待MQTT消息。

代码示例：

import _thread
from umqtt.simple import MQTTClient
from machine import Pin
import time

# 配置LED引脚
led = Pin(2, Pin.OUT)
# MQTT服务器配置
MQTT_BROKER = "你的Broker地址"
CLIENT_ID = "micropython_device"

def blink_thread():
    while True:
        led.value(not led.value())
        time.sleep(0.5)

# 启动LED闪烁线程
_thread.start_new_thread(blink_thread, ())

# 主线程处理MQTT
client = MQTTClient(CLIENT_ID, MQTT_BROKER)
client.connect()
print("已连接MQTT服务器")

while True:
    client.wait_msg()  # 阻塞等待消息，线程不受影响

注意：MicroPython线程资源有限，避免在子线程中执行复杂IO操作。

3. 修改主循环（无额外依赖，手动轮询） #

放弃阻塞的wait_msg()，改用check_msg()非阻塞检查消息，同时在主循环中通过时间戳控制LED闪烁时机。

代码示例：

from umqtt.simple import MQTTClient
from machine import Pin
import time

# 配置LED引脚
led = Pin(2, Pin.OUT)
# MQTT服务器配置
MQTT_BROKER = "你的Broker地址"
CLIENT_ID = "micropython_device"

# 记录LED上次切换时间
last_led_time = time.ticks_ms()
blink_interval = 500  # 闪烁间隔500毫秒

client = MQTTClient(CLIENT_ID, MQTT_BROKER)
client.connect()
print("已连接MQTT服务器")

while True:
    # 非阻塞检查MQTT消息
    client.check_msg()
    
    # 处理LED闪烁逻辑
    current_time = time.ticks_ms()
    if time.ticks_diff(current_time, last_led_time) >= blink_interval:
        led.value(not led.value())
        last_led_time = current_time
    
    time.sleep(0.05)  # 降低CPU占用

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