Reference library

Streaming & messaging

Kafka-style pub/sub, event consumers, async pipelines, and message-driven workflows.

3 matches

Batch Consume Process Commit Pattern in Python

A mock batch processor that accumulates items in a queue, processes full batches, commits successful or failed results, and flushes remaining items.

streaming batch-processing queues

Python

import random
import threading
import time
from collections import deque


class MockBatchProcessor:
    def __init__(self, process_func, commit_func, batch_size=5):
        self.queue = deque()
        self.batch_size = batch_size
        self.process_func = process_func
        self.commit_func = commit_func

    de…

14 0 Open

Streaming & messaging medium

How to Implement Backpressure Pause Producer with a Bounded Queue in Python

Places a Producer thread that sends items into a bounded queue with backpressure: on Full, it pauses to let the consumer catch up.

queues backpressure threading

Python

import threading
import time
import queue
import random


class Producer:
    def __init__(self, q):
        self.q = q
        self.running = True

    def produce(self):
        while self.running:
            item = random.randint(1, 100)
            try:
                self.q.put(item, timeout=0.5)
              …

14 0 Open

Streaming & messaging medium

How to Simulate RabbitMQ Exchange Routing in Python

Simulate RabbitMQ exchange routing using a nested dict, matching routing keys against patterns like error.* and info.# to return bound queues.

rabbitmq routing messaging

Python

from collections import defaultdict

def route_message(exchanges, exchange_name, routing_key):
    """
    Simulate RabbitMQ exchange routing using a nested dict structure.
    Returns list of queue names that match the routing key.
    """
    queues = exchanges.get(exchange_name, {})
    matched = []
    
    for pa…

14 0 Open

Browse by section

Each section groups closely related Python snippets.