队列¶
asyncio 队列的设计与 queue 模块的类相似。尽管 asyncio 队列不是线程安全的，但它们是专门设计用于 async/await 代码的。
请注意，asyncio 队列的方法没有 timeout 参数；使用 asyncio.wait_for() 函数来执行带有超时的队列操作。
另请参见下面的 示例 部分。
队列¶
- class asyncio.Queue(maxsize=0)¶
一个先进先出 (FIFO) 队列。
如果 maxsize 小于或等于零，则队列大小是无限的。如果它是一个大于
0的整数，则当队列达到 maxsize 时，await put()会阻塞，直到通过get()删除一个项目。与标准库线程
queue不同，队列的大小始终是已知的，并且可以通过调用qsize()方法返回。在 3.10 版本中更改: 删除了 loop 参数。
此类是非线程安全的。
- maxsize¶
队列中允许的项目数。
- empty()¶
如果队列为空，则返回
True，否则返回False。
- 协程 get()¶
从队列中删除并返回一个项目。如果队列为空，则等待直到有项目可用。
如果队列已关闭且为空，或者如果队列已立即关闭，则引发
QueueShutDown。
- get_nowait()¶
如果项目立即可用，则返回一个项目，否则引发
QueueEmpty。
- 协程 join()¶
阻塞，直到队列中的所有项目都被接收和处理。
每当向队列中添加一个项目时，未完成任务的计数就会增加。每当消费者协程调用
task_done()来指示该项目已被检索并且对其的所有工作都已完成时，计数就会减少。当未完成任务的计数降至零时，join()将取消阻塞。
- 协程 put(item)¶
将一个项目放入队列。如果队列已满，则等待，直到有空闲槽位可用，然后再添加项目。
如果队列已关闭，则引发
QueueShutDown。
- qsize()¶
返回队列中的项目数。
- shutdown(immediate=False)¶
关闭队列，使
get()和put()引发QueueShutDown。默认情况下，关闭队列上的
get()仅在队列为空时引发。将 immediate 设置为 true 以使get()立即引发。所有被阻塞的
put()和get()调用者将被取消阻塞。如果 immediate 为 true，则队列中每个剩余的项目都将被标记为已完成，这可能会取消阻塞join()的调用者。在 3.13 版本中添加。
- task_done()¶
指示先前排队的任务已完成。
由队列消费者使用。对于用于获取任务的每个
get()，后续调用task_done()会告诉队列该任务的处理已完成。如果
join()当前正在阻塞，则当所有项目都已处理完毕时（意味着已收到task_done()调用，该调用对应于放入队列中的每个项目），它将恢复。shutdown(immediate=True)为队列中的每个剩余项目调用task_done()。如果调用的次数多于放入队列中的项目数，则引发
ValueError。
优先级队列¶
后进先出队列¶
异常¶
- exception asyncio.QueueEmpty¶
当在空队列上调用
get_nowait()方法时，会引发此异常。
- exception asyncio.QueueFull¶
当在已达到其 maxsize 的队列上调用
put_nowait()方法时，会引发此异常。
示例¶
队列可以用于在多个并发任务之间分配工作负载。
import asyncio
import random
import time
async def worker(name, queue):
while True:
# Get a "work item" out of the queue.
sleep_for = await queue.get()
# Sleep for the "sleep_for" seconds.
await asyncio.sleep(sleep_for)
# Notify the queue that the "work item" has been processed.
queue.task_done()
print(f'{name} has slept for {sleep_for:.2f} seconds')
async def main():
# Create a queue that we will use to store our "workload".
queue = asyncio.Queue()
# Generate random timings and put them into the queue.
total_sleep_time = 0
for _ in range(20):
sleep_for = random.uniform(0.05, 1.0)
total_sleep_time += sleep_for
queue.put_nowait(sleep_for)
# Create three worker tasks to process the queue concurrently.
tasks = []
for i in range(3):
task = asyncio.create_task(worker(f'worker-{i}', queue))
tasks.append(task)
# Wait until the queue is fully processed.
started_at = time.monotonic()
await queue.join()
total_slept_for = time.monotonic() - started_at
# Cancel our worker tasks.
for task in tasks:
task.cancel()
# Wait until all worker tasks are cancelled.
await asyncio.gather(*tasks, return_exceptions=True)
print('====')
print(f'3 workers slept in parallel for {total_slept_for:.2f} seconds')
print(f'total expected sleep time: {total_sleep_time:.2f} seconds')
asyncio.run(main())