asyncio Queue, Events and Locks

The asyncio package ships its own synchronization primitives - async versions of the threading ones. They look similar but behave differently: an await queue.get() suspends the current coroutine without blocking the thread, so the loop keeps running other tasks while you wait. These primitives build the standard async patterns: worker pools, start signals, and guarded shared state.

asyncio.Queue

import asyncio

async def producer(q):
    for i in range(6):
        await q.put(i)
        await asyncio.sleep(0.1)
    await q.put(None)

async def consumer(q, name):
    while True:
        item = await q.get()        # suspends, does not block the loop
        if item is None:
            q.task_done()
            break
        print(name, "processed", item)
        q.task_done()
        q.join()  # note: usually the main coroutine joins, not each consumer

async def main():
    q = asyncio.Queue(maxsize=3)
    workers = [asyncio.create_task(consumer(q, f"W{i}")) for i in range(2)]
    await producer(q)
    await q.join()                  # wait until all items processed
    for w in workers:
        w.cancel()

asyncio.run(main())

Same contract as queue.Queue - put, get, task_done, join, maxsize - but every blocking call becomes a suspend. Put the join() in the coordinator, and signal shutdown with a sentinel or by cancelling the workers.

Event

An asyncio.Event is a start/stop flag: tasks await event.wait() until another task calls event.set(). The classic use is a shutdown flag checked by long-running workers:

stop = asyncio.Event()

async def worker():
    while not stop.is_set():
        do_unit_of_work()
        await asyncio.sleep(0.05)

async def main():
    workers = [asyncio.create_task(worker()) for _ in range(4)]
    await asyncio.sleep(1)
    stop.set()                      # all workers exit their loops
    await asyncio.gather(*workers)

Lock, Semaphore and Condition

asyncio.Lock, asyncio.Semaphore, and asyncio.Condition mirror their threading cousins but are not reentrant-safe across tasks in the same way and are not shareable across event loops (each loop gets its own objects). The typical uses:

• Lock: guard in-memory state mutated by many tasks (a cache dict, a counter) with async with lock:.
• Semaphore: cap concurrent API calls, exactly like the threading chapter's rate-limit example.
• Condition: wait for a predicate on shared state, with the same while not pred: await cond.wait() loop discipline.

sem = asyncio.Semaphore(5)

async def call_api(url):
    async with sem:                  # at most 5 in flight
        return await fetch(url)

await asyncio.gather(*(call_api(u) for u in urls))

Summary

In asyncio, every synchronization object is awaitable: use asyncio.Queue for worker pools, Event for start/stop signaling, and Lock/Semaphore/Condition for guarded state - remembering they belong to one event loop and every "block" is really a suspend.