Python中如何等待queue.get()或event.wait()首个完成？解决弃用警告

我需要在Python中实现等待queue.get()或event.wait()任意一个先完成的逻辑，目前用asyncio.wait()实现功能正常，但触发了以下弃用警告，需要修改代码以兼容未来Python版本：

DeprecationWarning: The explicit passing of coroutine objects to asyncio.wait() is deprecated since Python 3.8, and scheduled for removal in Python 3.11.

原代码（触发警告） #

import random
import asyncio

event = asyncio.Event()
queue = asyncio.Queue()

async def producer():
    for i in range(5):
        print(f"Putting {i}")
        await queue.put(i)
        await asyncio.sleep(random.random())
        # 检查是否需要终止
        if event.is_set():
            break
    print("Producer done")

async def terminator():
    await asyncio.sleep(random.random() * 5)
    print("Terminating")
    event.set()
    print("Terminator done")

async def consumer():
    while True:
        print(f"Waiting on the result of either the queue or the event")
        done, _ = await asyncio.wait(
            [queue.get(), event.wait()],
            return_when=asyncio.FIRST_COMPLETED
        )
        # 检查是否需要终止
        if event.is_set():
            break
        # 否则处理队列中的元素
        item = done.pop().result()
        print(f"got {item}")
    print("Consumer done")

async def main():
    await asyncio.gather(producer(), terminator(), consumer())

asyncio.run(main())

修改方案 #

问题根源是直接将协程对象（queue.get()、event.wait()的返回值）传入asyncio.wait()，从Python 3.8开始这种写法被弃用。正确的做法是使用asyncio.create_task()将协程包装为Task对象后再传入asyncio.wait()。

修改后的代码 #

import random
import asyncio

event = asyncio.Event()
queue = asyncio.Queue()

async def producer():
    for i in range(5):
        print(f"Putting {i}")
        await queue.put(i)
        await asyncio.sleep(random.random())
        if event.is_set():
            break
    print("Producer done")

async def terminator():
    await asyncio.sleep(random.random() * 5)
    print("Terminating")
    event.set()
    print("Terminator done")

async def consumer():
    while True:
        print(f"Waiting on the result of either the queue or the event")
        # 用create_task包装协程为Task对象
        get_task = asyncio.create_task(queue.get())
        wait_task = asyncio.create_task(event.wait())
        done, _ = await asyncio.wait(
            [get_task, wait_task],
            return_when=asyncio.FIRST_COMPLETED
        )
        # 检查是否需要终止
        if event.is_set():
            # 取消未完成的任务，避免资源泄漏
            if not get_task.done():
                get_task.cancel()
            break
        # 处理队列元素
        item = get_task.result()
        print(f"got {item}")
    print("Consumer done")

async def main():
    await asyncio.gather(producer(), terminator(), consumer())

asyncio.run(main())

关键修改点 #

• 将queue.get()和event.wait()分别用asyncio.create_task()包装成Task对象，再传入asyncio.wait()
• 终止时取消未完成的get_task，避免协程挂起造成资源泄漏
• 直接通过Task对象的result()方法获取结果，逻辑与原代码一致

示例输出 #

Putting 0
Waiting on the result of either the queue or the event
got 0
Waiting on the result of either the queue or the event
Putting 1
got 1
Waiting on the result of either the queue or the event
Terminating
Terminator done
Consumer done
Producer done

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