import threading, time, queue, random

​

q = queue.Queue(maxsize=8)  # buffer

POISON = object()

​

def producer(n=20):

    for i in range(n):

        msg = {"id": i, "payload": f"order-{i}"}

        q.put(msg)             # blocks if buffer full (backpressure)

        print(f"[PRODUCER] enqueued {msg}")

        time.sleep(0.05)       # simulate bursts

    q.put(POISON)

​

def consumer():

    while True:

        msg = q.get()

        if msg is POISON:

            q.task_done()

            break

        try:

            # simulate flaky processing

            if random.random() < 0.15:

                raise RuntimeError("transient failure")

            print(f"[CONSUMER] processed {msg['id']}")

            # "ack" is our task_done; in real brokers you ack explicitly

            q.task_done()

        except Exception as e:

            print(f"[CONSUMER] error {e}; retrying later")

            time.sleep(0.1)