Skip to content

Comparing asyncio.Queue with Alternatives

asyncio.Queue is the default channel between coroutines, and usually the right one. The alternatives exist for specific reasons — a faster hand-rolled buffer, AnyIO's streams with close semantics that work on trio, janus for passing items between threads and coroutines — and each changes throughput, backpressure and shutdown behaviour. Measured on Python 3.14 moving 300,000 integers from one producer coroutine to one consumer: asyncio.Queue unbounded moved 3.07 million items per second and bounded at 100 2.60 million; a collections.deque with an asyncio.Event moved 10.65 million — with no backpressure at all; an AnyIO memory object stream with a buffer of 100 moved 288,000; janus with both sides async 805,000. Between a producer thread and a consumer coroutine, janus moved 148,000 items per second with backpressure on the thread, and loop.call_soon_threadsafe(queue.put_nowait, item) moved 175,000 with none. This guide compares them on what each actually provides.

Prerequisites

1. Measure the channel with your item rate in mind

Run the same producer-consumer pair over each channel, with a sentinel or close to end it:

async def bench_asyncio_queue(n, maxsize):
    q = asyncio.Queue(maxsize)

    async def produce():
        for i in range(n):
            await q.put(i)
        await q.put(None)

    async def consume():
        while (item := await q.get()) is not None:
            pass

    start = time.perf_counter()
    await asyncio.gather(produce(), consume())
    return n / (time.perf_counter() - start)

Measured with 300,000 items: 3.07 million per second unbounded, 2.60 million bounded at 100 — about 0.3–0.4 µs per item. At those rates the queue is never the bottleneck of a service whose items do any I/O; it matters for pipelines moving millions of tiny items, where per-item overhead is the cost. A consumer that, after each get, drains whatever else is already queued with get_nowait() reached 2.72 million — a small gain here, a larger one when per-item handling can be batched.

Verify: you know your items-per-second requirement and whether it is within an order of magnitude of the channel's limit.

Items per second, one producer, one consumer, 300,000 items 7 horizontal bars comparing deque + Event (no backpressure) with the others. Items per second, one producer, one consumer, 300,000 items deque + Event (no backpressure) 10.65 M/s asyncio.Queue() 3.07 M/s asyncio.Queue(100) 2.60 M/s janus.Queue(100), async-async 805 k/s anyio stream, buffer 100 288 k/s thread: call_soon_threadsafe 175 k/s thread: janus.Queue(100) 148 k/s Python 3.14; items are integers.

2. Know what a faster hand-rolled channel gives up

A deque and an Event is the minimum channel: append, set the event; the consumer waits, clears, and drains:

buffer = collections.deque()
ready = asyncio.Event()

def push(item):
    buffer.append(item)
    ready.set()

async def drain():
    while True:
        await ready.wait()
        ready.clear()
        while buffer:
            handle(buffer.popleft())

Measured: 10.65 million items per second, three and a half times asyncio.Queue. It achieves that by leaving out what makes a queue safe to use: no bound, so a fast producer grows memory without limit; no task_done/join, so there is no way to wait until work is finished; one consumer only, since two drains would race for items between wait and popleft; and no shutdown signal. It is reasonable inside a component that owns both ends and has its own bound — a batching writer, a coalescing buffer — and a liability as a general queue. The accounting asyncio.Queue adds is described in tracking unfinished work with task_done and join.

Verify: any hand-rolled channel documents its single consumer and has an explicit bound.

3. Use AnyIO streams for close semantics and trio support

AnyIO's memory object streams split the channel into a send end and a receive end. Closing every send end ends the receivers' async for; cloning gives several producers or consumers their own handles; the same code runs on asyncio and trio:

send, receive = anyio.create_memory_object_stream[int](max_buffer_size=100)

async def produce(send):
    async with send:                         # closing ends the consumer's loop
        for i in range(n):
            await send.send(i)

async def consume(receive):
    async with receive:
        async for item in receive:
            handle(item)

Measured: 288,000 items per second, about a ninth of asyncio.Queue on this benchmark, because each send and receive goes through AnyIO's checkpoint and cancellation machinery. That is still far above what most coroutine pipelines move, and the structural guarantees — no sentinels, EndOfStream when producers are done, BrokenResourceError when consumers are gone — remove a class of shutdown bugs. asyncio.Queue.shutdown() in Python 3.13+ provides similar end-of-stream semantics, as in shutting down queues with Queue.shutdown in Python 3.13; the AnyIO mapping is covered in bridging AnyIO memory object streams and asyncio queues.

Verify: if you need trio support or clone-based fan-in and fan-out, the stream's throughput fits your item rate.

What each channel provides A grid of 5 rows by 6 columns. What each channel provides channel bound / backpressure end-of-stream many consumers across threads items/s here asyncio.Queue yes shutdown() in 3.13+ yes no 2.6-3.1 M deque + Event no no no no 10.7 M anyio memory stream yes close / EndOfStream clone() no (same loop) 288 k janus.Queue yes, both sides close() yes yes 805 k / 148 k call_soon_threadsafe + Queue no (thread side) - yes yes 175 k The right channel is the one whose guarantees you need, at a rate you can afford.

4. Cross threads with janus or call_soon_threadsafe

asyncio.Queue is not thread-safe: calling its methods from another thread corrupts its state or never wakes the consumer. Two correct ways to get items from a thread into a coroutine:

# janus: a queue with a sync face and an async face; the thread blocks when it is full
q = janus.Queue(maxsize=100)

def producer_thread():
    for item in read_device():
        q.sync_q.put(item)                    # blocks the thread, never the loop

async def consumer():
    while True:
        handle(await q.async_q.get())

# call_soon_threadsafe: schedule the put on the loop; no bound on the thread side
def producer_thread_unbounded(loop, q):
    for item in read_device():
        loop.call_soon_threadsafe(q.put_nowait, item)

Measured: janus moved 148,000 items per second from a thread with backpressure on the thread; call_soon_threadsafe moved 175,000 with none — a thread producing faster than the loop consumes fills the loop's callback queue without limit. For bursty producers that must be throttled, janus's bounded sync side is the reason to use it; for low-rate notifications from a thread, call_soon_threadsafe needs no dependency. Both are covered with their pitfalls in bridging queues between threads and asyncio tasks.

Verify: no thread calls asyncio.Queue methods directly, and thread producers that can outpace the loop are bounded.

5. Choose by guarantees, then check the rate

The decision is rarely about speed. Start from what the channel must guarantee and pick the simplest that does:

# defaults that cover most cases
work = asyncio.Queue(maxsize=1000)                      # coroutines to coroutines

# same loop, structured shutdown, trio-compatible code
send, receive = anyio.create_memory_object_stream(1000)

# threads to coroutines, with backpressure on the thread
bridge = janus.Queue(maxsize=1000)

Then confirm that the channel's rate is well above what you need — here, every option exceeded 140,000 items per second, which is more than nearly any I/O-bound service will push through one channel. Where it is not — millions of tiny items per second — batch items into lists before putting them, which divides per-item channel cost by the batch size, before reaching for a hand-rolled buffer. Durability is a separate axis entirely; none of these survives a crash, which persisting queue items to disk addresses.

Verify: each channel in the codebase was chosen for a guarantee it provides, and its measured rate exceeds the required rate with room to spare.

Which channel fits? A decision on Who produces and consumes with 4 outcomes. Which channel fits? Who produces and consumes? coroutines on one loop asyncio.Queue(maxsize) 2.6 M/s AnyIO / trio code memory object stream close semantics, 288 k/s a thread produces janus (bounded) or call_soon_threadsafe 148-175 k/s one owner, own bound, extreme rate deque + Event 10.7 M/s, no guarantees Batch items before swapping channels for speed.

Verification

A channel choice is sound when:

  • Its guarantees match the need: bound, end-of-stream, consumers, thread safety.
  • Its measured rate is well above the required item rate.
  • No asyncio.Queue is touched from another thread.
  • Hand-rolled buffers are single-consumer and bounded, and documented as such.

Diagnostic Hook: if a pipeline's CPU profile shows significant time in queue put/get or AnyIO checkpoint frames, the items are too small for the channel: batch them. A channel that shows up in a profile of an I/O-bound service is almost always carrying one item where it could carry a list.

Pitfalls & edge cases

  • Choosing a deque for speed. It gave up the bound, join and multiple consumers.
  • Expecting AnyIO streams to be free. Measured: about a ninth of asyncio.Queue's rate.
  • Calling asyncio.Queue from threads. Use janus or call_soon_threadsafe.
  • Unbounded thread-to-loop delivery. call_soon_threadsafe cannot push back on the thread.

Frequently Asked Questions

How fast is asyncio.Queue?

About 2.6 to 3.1 million items per second between two coroutines in testing, roughly 0.3 to 0.4 µs per item; it is rarely a bottleneck.

Is a deque faster than asyncio.Queue?

Yes, 10.65 million items per second with an asyncio.Event, but without a bound, join, or safe multiple consumers.

Should I use anyio memory object streams instead of asyncio.Queue?

When you want close-based end-of-stream, cloning, or trio compatibility. They moved 288,000 items per second here, slower but ample for most pipelines.

How do I pass items from a thread to an asyncio coroutine?

janus.Queue gives the thread a blocking, bounded put (148,000 items/s here); loop.call_soon_threadsafe(queue.put_nowait, item) works without a dependency but without backpressure (175,000/s).