Python异步bot如何实现高效监听列表变更的队列任务以替代while True？

核心思路 #

使用Python标准库asyncio自带的asyncio.Queue替换自定义的全局列表队列，它原生支持异步阻塞等待新元素，完全不需要轮询，且内部实现了原子操作，天然满足协程安全、避免重复消费的需求。

完整改造代码 #

队列操作实现替换 #

import asyncio

# 替换全局列表为asyncio.Queue实例，可传入maxsize参数限制队列最大长度避免内存溢出
global_queue = asyncio.Queue()

async def Queue(action=None, globalmsg=None, original=None, code=None):
    if action is not None:
        if action == "add":
            # 队列满时put操作会自动阻塞等待空位，不会丢消息
            await global_queue.put((globalmsg, original, code))
        elif action == "remove":
            # 常规消费场景不需要手动调用remove，get操作会自动移除已取出的元素
            # 仅特殊场景需要手动标记任务完成时保留
            try:
                global_queue.task_done()
            except ValueError:
                pass
    else:
        # 非必要场景不建议调用，仅兼容原有获取全量队列的逻辑
        return list(global_queue._queue)

消费逻辑改造 #

def __init__(self, client):
    self.client = client
    # 仅启动一次消费协程，保证只有一个消费任务运行，从根源避免重复消费
    client.loop.create_task(self.on_queue())

async def on_queue(self):
    await self.client.wait_until_ready()
    while True:
        # 队列为空时自动挂起等待，直到有新元素加入才唤醒，无任何空转性能损耗
        item = await global_queue.get()
        globalmsg, original, code = item
        try:
            # 原有业务逻辑直接写在这里即可
            pass
        finally:
            # 标记当前任务处理完成，可选，需要等待所有队列任务处理完时才需要调用
            global_queue.task_done()

方案优势 #

• 零轮询开销：无新消息时消费协程完全休眠，不占用CPU资源，新消息到达时立刻唤醒处理，延迟为0
• 天然防重复消费：asyncio.Queue的get操作是原子性的，且仅启动单个消费协程，同一条消息只会被处理一次
• 协程安全：多协程同时往队列添加消息也不会出现数据竞争，不需要额外加锁
• 可扩展能力强：支持设置队列最大长度，满容时自动阻塞写入操作，避免消息堆积导致内存溢出

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