Skip to content

Processing Queue Items in Order per Key

Account events arrive in order — opened, deposited, withdrew, closed — and eight workers process them concurrently from one queue. Almost always the result is right. Occasionally a withdrew for an account is applied before its deposited, because the worker that picked up the deposit happened to await a slower database round trip, and the balance check fails on money that should have been there. The events for one account need strict order; events for different accounts are independent and should run in parallel. That per-key ordering requirement shows up everywhere: order updates, document edits, chat messages per conversation, inventory adjustments per SKU. This guide measures how a plain shared queue breaks it, implements hash partitioning and shows how a single hot key stalls unrelated keys under it, then builds on-demand key lanes that keep per-key order without that head-of-line blocking, and handles retries without reordering.

Prerequisites

Three ways to parallelise keyed events 3 columns contrasting shared queue, hash partitions, key lanes. Three ways to parallelise keyed events shared queue any worker, any event fastest overall per-key order breaks no hot-key stall hash partitions key -> fixed worker per-key order kept hot key stalls neighbours stable hash required key lanes one lane per busy key per-key order kept hot key stalls itself only bounded by a semaphore Lanes keep partitioning's guarantee without tying unrelated keys together.

1. Show that a shared queue reorders a key's events

Model each event as (key, sequence, processing_time) and record the order in which work completes. With several workers on one queue, an event with a shorter processing time can finish before an earlier event for the same key.

import asyncio
import collections
import random

Event = tuple[str, int, float]


def make_events(hot: bool = False, n: int = 400, keys: int = 40) -> list[Event]:
    rng = random.Random(3)
    names = [f"acct-{i}" for i in range(keys)]
    seq: collections.Counter[str] = collections.Counter()
    events = []
    for _ in range(n):
        key = "acct-0" if hot and rng.random() < 0.5 else rng.choice(names)
        events.append((key, seq[key], rng.uniform(0.001, 0.004)))
        seq[key] += 1
    return events


async def process(event: Event, log: list[Event]) -> None:
    await asyncio.sleep(event[2])                        # variable latency per event
    log.append(event)


def in_order(log: list[Event]) -> bool:
    last: dict[str, int] = {}
    for key, seq, _ in log:
        if last.get(key, -1) > seq:
            return False
        last[key] = seq
    return True


async def shared_queue(events: list[Event], workers: int) -> list[Event]:
    queue: asyncio.Queue[Event] = asyncio.Queue()
    for event in events:
        queue.put_nowait(event)
    log: list[Event] = []

    async def worker() -> None:
        while not queue.empty():
            await process(queue.get_nowait(), log)

    await asyncio.gather(*(worker() for _ in range(workers)))
    return log


print("shared queue keeps per-key order:", in_order(asyncio.run(shared_queue(make_events(), 8))))

The check reports False: with 400 events over 40 accounts and eight workers, some account's events completed out of sequence. The failure is probabilistic in production, which is why it survives testing — the run above makes it deterministic with a fixed seed.

Verify: the shared-queue run prints False; reducing workers to one prints True, at eight times the processing time.

2. Partition by key hash

The classic fix routes every event for a key to the same worker: hash the key, take it modulo the worker count, and give each worker its own queue. One worker per partition processes that partition's events sequentially, so each key's order is preserved.

import asyncio
import zlib


async def partitioned(events: list[Event], workers: int) -> list[Event]:
    queues: list[asyncio.Queue[Event]] = [asyncio.Queue() for _ in range(workers)]
    for event in events:
        partition = zlib.crc32(event[0].encode()) % workers   # stable across processes
        queues[partition].put_nowait(event)
    log: list[Event] = []

    async def worker(queue: asyncio.Queue[Event]) -> None:
        while not queue.empty():
            await process(queue.get_nowait(), log)

    await asyncio.gather(*(worker(q) for q in queues))
    return log


print("partitioned keeps per-key order:", in_order(asyncio.run(partitioned(make_events(), 8))))

Use a stable hash such as zlib.crc32 rather than Python's built-in hash(), which is randomised per process for strings and would route a key differently after a restart or in another worker process. This is the model Kafka uses with partitions and consumer groups, covered in consuming Kafka topics with aiokafka.

Verify: the partitioned run prints True.

3. Measure the hot-key problem

Partitioning ties unrelated keys together. When one key is hot — a big customer, a busy conversation — its partition backs up, and every other key hashed into that partition waits behind it even though its own traffic is light.

import asyncio
import statistics
import time


async def cold_key_latency(strategy, events: list[Event], workers: int) -> tuple[float, float]:
    finished: dict[tuple[str, int], float] = {}
    started = time.perf_counter()
    global process

    async def timed_process(event: Event, log: list[Event]) -> None:
        await asyncio.sleep(event[2])
        log.append(event)
        finished[(event[0], event[1])] = time.perf_counter() - started

    original, process = process, timed_process
    try:
        await strategy(events, workers)
    finally:
        process = original
    cold = sorted(t for (key, _), t in finished.items() if key != "acct-0")
    return statistics.median(cold), cold[-1]


async def main() -> None:
    hot_events = make_events(hot=True)                   # half of all events belong to acct-0
    median, worst = await cold_key_latency(partitioned, hot_events, 8)
    print(f"partitioned: cold keys median {median:.3f}s, last finished at {worst:.2f}s")


asyncio.run(main())

With half of the traffic on acct-0, the median cold key still finished quickly, but the last cold events finished at about 0.65 seconds — the time it took to work through the hot key's backlog in the shared partition. Those unlucky accounts waited for another customer's events.

Verify: the worst cold-key completion time is close to the total time needed to process the hot key's events.

When the last light key finished, with one hot key 3 bars comparing hash partitions, last cold key with the others. When the last light key finished, with one hot key hash partitions, last cold key 0.65 s key lanes, last cold key 0.10 s either, median cold key 0.045 s Measured with the step 3 and 4 code: 400 events, 40 keys, half on acct-0, 8 workers. Partitioning makes some unlucky keys wait for someone else's backlog.

4. Give each key its own lane, on demand

Instead of a fixed number of partitions, create a lane — a small deque with one drainer task — for each key that currently has pending events, and remove it when it empties. A semaphore bounds how many lanes run at once. Each key is still processed sequentially, but a hot key occupies one slot, not a whole partition of unrelated keys.

import asyncio
import collections


class KeyLanes:
    def __init__(self, max_active: int) -> None:
        self._lanes: dict[str, collections.deque[Event]] = {}
        self._slots = asyncio.Semaphore(max_active)
        self._tasks: set[asyncio.Task] = set()
        self.log: list[Event] = []

    def submit(self, event: Event) -> None:
        key = event[0]
        lane = self._lanes.get(key)
        if lane is None:                                   # first pending event for this key
            lane = self._lanes[key] = collections.deque()
            task = asyncio.create_task(self._drain(key), name=f"lane:{key}")
            self._tasks.add(task)
            task.add_done_callback(self._tasks.discard)
        lane.append(event)                                 # later events join the existing lane

    async def _drain(self, key: str) -> None:
        async with self._slots:                            # at most max_active keys in progress
            lane = self._lanes[key]
            while lane:
                await process(lane.popleft(), self.log)
            del self._lanes[key]                           # no await between check and delete

    async def join(self) -> None:
        while self._tasks:
            await asyncio.gather(*list(self._tasks))


async def lanes(events: list[Event], workers: int) -> list[Event]:
    key_lanes = KeyLanes(max_active=workers)
    for event in events:
        key_lanes.submit(event)
    await key_lanes.join()
    return key_lanes.log


async def main() -> None:
    print("lanes keep per-key order:", in_order(await lanes(make_events(hot=True), 8)))
    median, worst = await cold_key_latency(lanes, make_events(hot=True), 8)
    print(f"lanes: cold keys median {median:.3f}s, last finished at {worst:.2f}s")


asyncio.run(main())

Order is preserved, and cold keys finished by about 0.1 seconds instead of 0.65: the hot key still takes as long as its own backlog requires, but nobody else waits behind it. The while lane check and the del happen without an await in between, so an event submitted for the key can never slip into a lane that is being deleted.

Verify: the ordering check prints True, and the worst cold-key completion time is several times lower than with partitioning.

5. Retry without breaking the order

A failing event must not be skipped while later events for the same key proceed — that reorders them just as surely as a shared queue. Retry inside the lane, blocking only that key, and park the lane's remaining events if the event finally fails.

import asyncio
import collections


class RetryingLanes(KeyLanes):
    def __init__(self, max_active: int, attempts: int = 3) -> None:
        super().__init__(max_active)
        self.attempts = attempts
        self.parked: dict[str, list[Event]] = {}

    async def _drain(self, key: str) -> None:
        async with self._slots:
            lane = self._lanes[key]
            while lane:
                event = lane[0]                            # peek: stays first until done
                for attempt in range(1, self.attempts + 1):
                    try:
                        await flaky_process(event, self.log, attempt)
                        lane.popleft()
                        break
                    except ConnectionError:
                        await asyncio.sleep(0.001 * 2 ** attempt)   # only this key waits
                else:
                    self.parked[key] = list(lane)          # keep the rest in order for replay
                    lane.clear()
            del self._lanes[key]


async def flaky_process(event: Event, log: list[Event], attempt: int) -> None:
    key, seq, _ = event
    if key == "acct-5" and seq == 1 and attempt < 3:       # transient failure, twice
        raise ConnectionError("database failover")
    if key == "acct-9" and seq == 0:                       # permanent failure
        raise ConnectionError("account locked")
    await process(event, log)


async def main() -> None:
    retrying = RetryingLanes(max_active=8)
    for event in make_events():
        retrying.submit(event)
    await retrying.join()
    print("order kept:", in_order(retrying.log))
    print("parked keys:", {k: len(v) for k, v in retrying.parked.items()})


asyncio.run(main())

The transient failure on acct-5 delayed only that account, and its later events ran after the retry succeeded. The permanent failure parked all of acct-9's events, in order, for inspection and replay — the per-key equivalent of a dead-letter queue. Other accounts were unaffected throughout.

Verify: the log is in order, and the parked dictionary contains only acct-9 with all of its events.

Failure handling that keeps order A grid of 3 rows by 2 columns. Failure handling that keeps order on failure order for the key other keys skip and continue broken unaffected retry at the head of the lane kept unaffected park the rest of the lane kept for replay unaffected Never let a later event for a key overtake an earlier one that has not succeeded.

Verification

Per-key ordered processing is correct when:

  • An ordering check passes under concurrency: completion order per key matches sequence numbers with many workers and variable latency.
  • Routing is stable: partitioned designs use a stable hash, so a key maps to the same worker across restarts and processes.
  • Hot keys do not stall cold keys: worst-case latency for light keys stays low while a heavy key has a backlog.
  • Concurrency is bounded: the number of keys processed at once is capped, and idle lanes are removed.
  • Retries preserve order: a failing event blocks only its key, and permanently failing keys park their remaining events in order.

Pitfalls & edge cases

  • Using hash() for partitioning. String hashes are randomised per process, so the same key routes differently after a restart. Use zlib.crc32 or another stable hash.
  • Order across keys. Neither design orders events between keys; if a transfer must see both accounts in a consistent state, it needs a transaction or a combined key.
  • Unbounded lane memory. A hot key's lane can grow without limit. Bound per-key backlog and apply backpressure or reject when it is exceeded.
  • Resharding partitions. Changing the partition count moves keys between workers; in-flight events for a moved key can race with new ones. Drain before resharding, or use lanes, which have no partition count.
  • Multiple processes. In-memory lanes guarantee order within one process only. Across processes, route by key at the source — a partitioned broker topic — and run lanes inside each consumer.

Frequently Asked Questions

How do I process asyncio queue items in parallel but in order per key?

Serialise work per key while allowing different keys to run concurrently. Either route each key to a fixed worker by a stable hash of the key, or keep a small queue per key with one draining task and bound how many keys run at once with a semaphore.

Why do my events get processed out of order with multiple asyncio workers?

Several workers take events from one queue in order, but each event takes a different amount of time, so a later event for the same key can finish before an earlier one that awaited a slower call. Sequential processing per key is needed wherever order matters.

What is the hot-key problem with hash partitioning?

With a fixed number of partitions, every key that hashes to the same partition shares one worker. When one key receives a large share of traffic, its backlog delays all the unrelated keys in that partition. Per-key lanes avoid it because a busy key occupies only its own lane.

How do I retry a failed event without breaking per-key order?

Retry the failing event inside its key's processing loop, keeping it at the front so later events for that key wait, while other keys continue. If it still fails after the retry budget, move it and every remaining event for that key to a parked or dead-letter store in their original order.