Queue¶

classasyncio.Queue(maxsize=0)¶

A first in, first out (FIFO) queue.

Ifmaxsize is less than or equal to zero, the queue size isinfinite. If it is an integer greater than0, thenawaitput() blocks when the queue reachesmaxsizeuntil an item is removed byget().

Unlike the standard library threadingqueue, the size ofthe queue is always known and can be returned by calling theqsize() method.

Changed in version 3.10:Removed theloop parameter.

This class isnot thread safe.

maxsize¶

Number of items allowed in the queue.

empty()¶

ReturnTrue if the queue is empty,False otherwise.

full()¶

ReturnTrue if there aremaxsize items in the queue.

If the queue was initialized withmaxsize=0 (the default),thenfull() never returnsTrue.

asyncget()¶

Remove and return an item from the queue. If queue is empty,wait until an item is available.

RaisesQueueShutDown if the queue has been shut down andis empty, or if the queue has been shut down immediately.

get_nowait()¶

Return an item if one is immediately available, else raiseQueueEmpty.

asyncjoin()¶

Block until all items in the queue have been received and processed.

The count of unfinished tasks goes up whenever an item is addedto the queue. The count goes down whenever a consumer coroutine callstask_done() to indicate that the item was retrieved and allwork on it is complete. When the count of unfinished tasks dropsto zero,join() unblocks.

asyncput(item)¶

Put an item into the queue. If the queue is full, wait until afree slot is available before adding the item.

RaisesQueueShutDown if the queue has been shut down.

put_nowait(item)¶

Put an item into the queue without blocking.

If no free slot is immediately available, raiseQueueFull.

qsize()¶

Return the number of items in the queue.

shutdown(immediate=False)¶

Shut down the queue, makingget() andput()raiseQueueShutDown.

By default,get() on a shut down queue will onlyraise once the queue is empty. Setimmediate to true to makeget() raise immediately instead.

All blocked callers ofput() andget()will be unblocked. Ifimmediate is true, a task will be markedas done for each remaining item in the queue, which may unblockcallers ofjoin().

Added in version 3.13.

task_done()¶

Indicate that a formerly enqueued work item is complete.

Used by queue consumers. For eachget() used tofetch a work item, a subsequent call totask_done() tells thequeue that the processing on the work item is complete.

If ajoin() is currently blocking, it will resume when allitems have been processed (meaning that atask_done()call was received for every item that had beenput()into the queue).

shutdown(immediate=True) callstask_done() for eachremaining item in the queue.

RaisesValueError if called more times than there wereitems placed in the queue.

Priority Queue¶

classasyncio.PriorityQueue¶

A variant ofQueue; retrieves entries in priority order(lowest first).

Entries are typically tuples of the form(priority_number,data).

LIFO Queue¶

classasyncio.LifoQueue¶

A variant ofQueue that retrieves most recently addedentries first (last in, first out).

Exceptions¶

exceptionasyncio.QueueEmpty¶

This exception is raised when theget_nowait() methodis called on an empty queue.

exceptionasyncio.QueueFull¶

Exception raised when theput_nowait() method is calledon a queue that has reached itsmaxsize.

exceptionasyncio.QueueShutDown¶

Exception raised whenput() orget() iscalled on a queue which has been shut down.

Added in version 3.13.

Examples¶

Queues can be used to distribute workload between severalconcurrent tasks:

importasyncioimportrandomimporttimeasyncdefworker(name,queue):whileTrue:# Get a "work item" out of the queue.sleep_for=awaitqueue.get()# Sleep for the "sleep_for" seconds.awaitasyncio.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')asyncdefmain():# 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=0for_inrange(20):sleep_for=random.uniform(0.05,1.0)total_sleep_time+=sleep_forqueue.put_nowait(sleep_for)# Create three worker tasks to process the queue concurrently.tasks=[]foriinrange(3):task=asyncio.create_task(worker(f'worker-{i}',queue))tasks.append(task)# Wait until the queue is fully processed.started_at=time.monotonic()awaitqueue.join()total_slept_for=time.monotonic()-started_at# Cancel our worker tasks.fortaskintasks:task.cancel()# Wait until all worker tasks are cancelled.awaitasyncio.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())