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)¶

Put aQueue instance into a shutdown mode.

The queue can no longer grow.Future calls toput() raiseQueueShutDown.Currently blocked callers ofput() will be unblockedand will raiseQueueShutDown in the formerly blocked thread.

Ifimmediate is false (the default), the queue can be wounddown normally withget() calls to extract tasksthat have already been loaded.

And iftask_done() is called for each remaining task, apendingjoin() will be unblocked normally.

Once the queue is empty, future calls toget() willraiseQueueShutDown.

Ifimmediate is true, the queue is terminated immediately.The queue is drained to be completely empty and the countof unfinished tasks is reduced by the number of tasks drained.If unfinished tasks is zero, callers ofjoin()are unblocked. Also, blocked callers ofget()are unblocked and will raiseQueueShutDown because thequeue is empty.

Use caution when usingjoin() withimmediate setto true. This unblocks the join even when no work has been doneon the tasks, violating the usual invariant for joining a queue.

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).

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())