Queue¶

classasyncio.Queue(maxsize=0,*,loop=None)¶

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.

Deprecated since version 3.8, will be removed in version 3.10: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.

coroutineget()¶

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

get_nowait()¶

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

coroutinejoin()¶

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.

coroutineput(item)¶

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

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.

task_done()¶

Indicate that a formerly enqueued task is complete.

Used by queue consumers. For eachget() used tofetch a task, a subsequent call totask_done() tells thequeue that the processing on the task 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.

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