如何在Python中为MCU UDP监听线程实现超时删除订单功能

需求说明 #

• 向微控制器（MCU）发送信号生成指令时，将包含频率、时长的订单加入跟踪集合tracked_orders
• MCU按指定频率在设定时长内发送数据，由UDP读取线程__read_thread_UDP__负责监听
• 核心目标：当tracked_orders为空时停止UDP线程，需实现异步超时函数，超时后自动从集合中删除对应订单

现有代码片段 #

def read_pin(self,pin:Pin,duration,freq):
    self.tracked_orders.insert(Order(pin,duration,freq))

def __read_thread_UDP__(self):
    while len(self.tracked_orders)!= 0:
          # 解析接收的UDP消息逻辑

# 待实现异步超时删除逻辑
def __remove_order_by_timeout__(self,order):
    # 当前为阻塞轮询的伪代码
    start = time.time()
    while start + order.duration < time.time():
        continue
    finally:
       self.tracked_orders.remove(order)

改进后的异步实现 #

原伪代码用阻塞轮询的方式等待超时，会浪费CPU资源，以下是基于asyncio的异步非阻塞实现，同时加入线程安全保障：

import asyncio
from threading import Lock

class YourDeviceHandler:
    def __init__(self):
        # 改用set存储订单，删除操作更高效，需确保Order类可哈希
        self.tracked_orders = set()
        # 线程锁：避免UDP线程和异步超时任务同时操作集合引发冲突
        self.order_lock = Lock()

    def read_pin(self, pin: Pin, duration, freq):
        order = Order(pin, duration, freq)
        # 加锁操作集合，保证线程安全
        with self.order_lock:
            self.tracked_orders.add(order)
        # 启动异步超时删除任务
        asyncio.create_task(self.__remove_order_by_timeout__(order))

    def __read_thread_UDP__(self):
        while True:
            with self.order_lock:
                # 集合为空时退出线程
                if not self.tracked_orders:
                    break
            # 解析UDP接收消息的业务逻辑
            # ...

    async def __remove_order_by_timeout__(self, order):
        # 异步等待订单时长，不占用CPU资源
        await asyncio.sleep(order.duration)
        # 再次检查订单是否存在，避免已被其他逻辑删除
        with self.order_lock:
            if order in self.tracked_orders:
                self.tracked_orders.remove(order)

关键改进点 #

• 替换阻塞轮询为asyncio.sleep，实现真正的异步等待，降低CPU占用
• 新增线程锁order_lock，保障多线程环境下集合操作的安全性
• 将tracked_orders改为set，删除操作时间复杂度从O(n)降至O(1)
• 每个订单独立启动异步超时任务，精准控制超时删除时机

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