Event-Driven Architecture Without Kafka

An order is placed. The API writes it to the database, returns 200, and closes the connection.

Three things still need to happen: inventory needs to update, a confirmation email needs to go out, and the analytics pipeline needs to record the sale. You can do all three in the request handler. Most teams start there.

The problem shows up when the email service is slow. Or when the inventory call times out. Or when a new team needs to hook into the same event and you have to touch the endpoint to do it.

The answer is to get that work out of the request path. How you do it depends on what the work actually requires.

The simplest queue that works

If you already run Redis, you already have everything you need for a background task system. Celery uses Redis as a broker. The API publishes tasks to a queue. A separate worker process picks them up and runs them.

Nothing else changes: the order endpoint writes to the database and returns, and the rest happens elsewhere.

# celery_app.py
from celery import Celery
from kombu import Queue

app = Celery(
    "shop_worker",
    broker="redis://localhost:6379/0",
    backend=None,  # fire-and-forget — no result storage needed
)

app.conf.update(
    task_serializer="json",
    accept_content=["json"],
    task_acks_late=True,        # ack after the handler finishes, not on receipt
    task_reject_on_worker_lost=True,  # re-queue if the worker dies mid-task
    worker_prefetch_multiplier=1,     # one task at a time per worker slot
    task_queues=(
        Queue("shop_events_high"),  # fast, critical (email confirmations, stock checks)
        Queue("shop_events_low"),   # slow, heavy (nightly reports, bulk exports)
    ),
    task_default_queue="shop_events_low",
)

Two queues matter more than they look. A nightly report job shouldn’t sit in the same queue as an order confirmation. If the report runs long, email confirmations pile up behind it. Separate queues let you run more workers on the high-priority side when load is high, without touching the slow work. The difference between a queue and a retained event log matters too—queues and streams solve different problems once consumers need replay or independent ownership.

Events decouple. Tasks depend.

The distinction between a queue and an event system is really a question of coupling.

A task says “do this.” A direct call to send_confirmation_email.delay(order_id) is explicit and easy to follow. It’s also a dependency. The order service now knows about the email service, so when a third thing needs to react to an order, the order service has to change.

An event says “this happened.” The order service publishes OrderPlaced. Anything that cares about orders subscribes to that event and handles it independently. The order service doesn’t know or care what happens next.

In practice, the cleanest version splits this across two buses:

# events/types.py
import uuid
from datetime import UTC, datetime
from typing import ClassVar
from pydantic import BaseModel, Field

class DomainEvent(BaseModel):
    _queue: ClassVar[str] = "shop_events_low"

    event_id: uuid.UUID = Field(default_factory=uuid.uuid4)
    occurred_at: datetime = Field(default_factory=lambda: datetime.now(UTC))

class OrderPlaced(DomainEvent):
    _queue: ClassVar[str] = "shop_events_high"
    order_id: uuid.UUID
    customer_id: uuid.UUID
    total: float

class ProductViewed(DomainEvent):
    _queue: ClassVar[str] = "shop_events_low"
    product_id: uuid.UUID
    customer_id: uuid.UUID | None

The API container uses an async bus that serializes events and sends them to Celery. It never imports or runs handler code.

# events/bus.py
import json

class AsyncEventBus:
    """Runs in the API container. Serialises events and sends to the worker queue."""

    TASK_NAME = "worker.tasks.handle_event"

    def __init__(self, celery_app) -> None:
        self._celery = celery_app

    def dispatch(self, *events) -> None:
        for event in events:
            payload = json.dumps({
                "event_type": type(event).__name__,
                "data": event.model_dump(mode="json"),
            })
            self._celery.send_task(
                self.TASK_NAME,
                args=[payload],
                queue=type(event)._queue,
            )

The worker container uses an in-process bus. Handlers register with a decorator, and the bus calls them when a matching event arrives.

# events/bus.py (continued)
from collections import defaultdict

class EventBus:
    """Runs inside the worker. Dispatches synchronously to registered handlers."""

    def __init__(self) -> None:
        self._handlers = defaultdict(list)

    def on(self, event_type):
        def decorator(handler):
            self._handlers[event_type].append(handler)
            return handler
        return decorator

    def dispatch(self, *events) -> None:
        for event in events:
            for handler in self._handlers[type(event)]:
                handler(event)

The handle_event Celery task sits between them. It deserializes the JSON payload, deduplicates by event ID, and dispatches to the in-process bus.

# worker/tasks.py
import json
from celery import shared_task
from redis import Redis

_EVENT_REGISTRY = {}  # populated by @domain_event decorator at import time

@shared_task(
    name="worker.tasks.handle_event",
    bind=True,
    max_retries=3,
    default_retry_delay=5,
    autoretry_for=(Exception,),
    acks_late=True,
)
def handle_event(self, payload_json: str) -> None:
    redis = get_redis()
    bus = get_bus()

    payload = json.loads(payload_json)
    event_cls = _EVENT_REGISTRY.get(payload["event_type"])
    if event_cls is None:
        return  # unknown event type — drop it

    event = event_cls.model_validate(payload["data"])

    # Deduplication: Redis NX with 24-hour TTL
    # If two workers race on the same event_id, only one proceeds.
    dedup_key = f"shop:event_seen:{event.event_id}"
    if not redis.set(dedup_key, 1, nx=True, ex=86400):
        return  # already processed

    bus.dispatch(event)

The deduplication step is easy to skip and worth not skipping. Celery can redeliver a task if the worker dies between processing and acknowledgment. Without it, order_confirmed_email goes out twice. That is why idempotency is not a nice-to-have when delivery is at least once.

Handlers stay in their own domain

Each domain registers its own handlers. Adding a new reaction to OrderPlaced means adding a handler in the relevant module, not changing the endpoint.

# shop/handlers.py
from events.bus import EventBus
from events.types import OrderPlaced, ProductViewed

def register_shop_handlers(bus: EventBus) -> None:

    @bus.on(OrderPlaced)
    def send_confirmation(event: OrderPlaced) -> None:
        send_email(event.customer_id, template="order_confirmation", order_id=event.order_id)

    @bus.on(OrderPlaced)
    def update_inventory(event: OrderPlaced) -> None:
        decrement_stock(event.order_id)

    @bus.on(ProductViewed)
    def record_signal(event: ProductViewed) -> None:
        # Increment a sorted set for product ranking signals.
        redis.zincrby("shop:product_signals", 1, str(event.product_id))

The order endpoint dispatches OrderPlaced and closes. It doesn’t know that inventory and email both run. That knowledge lives in the worker.

Scheduled work is the same system

Some things need to run on a clock, not in response to an event. Product ranking needs to be recomputed periodically. Stale signal data needs to be purged.

Celery Beat handles this. It fires tasks on a schedule and sends them to the same queues.

# worker/beat.py
from datetime import timedelta
from celery.schedules import crontab

BEAT_SCHEDULE = {
    "refresh-product-rankings": {
        "task": "worker.tasks.refresh_product_rankings",
        "schedule": timedelta(minutes=15),
        "options": {"queue": "shop_events_low"},
    },
    "purge-stale-signals": {
        "task": "worker.tasks.purge_stale_signals",
        "schedule": crontab(hour=3, minute=0),
        "options": {"queue": "shop_events_low"},
    },
}

The problem with Beat in a container environment is that it runs as a single process. If you deploy two Beat replicas for redundancy, both fire the same task at the same time. Two ranking refreshes run in parallel. The second one overwrites the first halfway through.

RedBeat solves this by storing the schedule in Redis and using a distributed lock. Only the replica holding the lock fires tasks. The others wait as hot standbys.

# celery_app.py (additions)
app.conf.update(
    beat_scheduler="redbeat.RedBeatScheduler",
    redbeat_redis_url="redis://localhost:6379/0",
    redbeat_key_prefix="shop:beat:",
    redbeat_lock_timeout=600,
    beat_max_loop_interval=5,
)

Inside the tasks themselves, a mutex prevents two workers from running the same scheduled task in parallel even if Beat fires it twice.

# worker/tasks.py
from contextlib import contextmanager

@contextmanager
def task_mutex(redis, name: str, timeout: int = 300):
    """Atomic lock. At most one instance runs across all workers."""
    lock_key = f"shop:task_lock:{name}"
    acquired = bool(redis.set(lock_key, 1, nx=True, ex=timeout))
    try:
        yield acquired
    finally:
        if acquired:
            redis.delete(lock_key)

@shared_task(name="worker.tasks.refresh_product_rankings", bind=True, max_retries=0)
def refresh_product_rankings(self) -> None:
    with task_mutex(get_redis(), "refresh_product_rankings", timeout=300) as acquired:
        if not acquired:
            return  # another worker is already running this
        # aggregate ZINCRBY signal data, compute scores, write to cache

Some work has no triggering event at all. Nobody checks out an abandoned cart. Nobody fires an event when a session goes cold. The state just accumulates until something explicitly cleans it up.

That is what Beat is actually for. A cart older than 72 hours with no activity is not something a user did. It is something time did. The cleanup runs on a clock, not in response to anything.

# worker/beat.py (addition)
"purge-abandoned-carts": {
    "task": "worker.tasks.purge_abandoned_carts",
    "schedule": crontab(hour=2, minute=0),  # 02:00 UTC, away from peak
    "options": {"queue": "shop_events_low"},
},

@shared_task(name="worker.tasks.purge_abandoned_carts", bind=True, max_retries=0)
def purge_abandoned_carts(self) -> None:
    with task_mutex(get_redis(), "purge_abandoned_carts", timeout=1800) as acquired:
        if not acquired:
            return  # already running — skip this firing
        cutoff = datetime.now(UTC) - timedelta(hours=72)
        deleted = db.execute(
            "DELETE FROM carts WHERE updated_at < %s AND status = 'abandoned' RETURNING id",
            [cutoff],
        ).rowcount
        logger.info("purged abandoned carts", count=deleted, cutoff=cutoff.isoformat())

The mutex matters here for the same reason it does on the ranking refresh. If the table is large and the delete runs slow, the next Beat firing shouldn’t start a second pass on top of it.

This is not the same as a stream

Everything above uses Redis as a queue. A queue delivers a message and removes it. The worker processes the event and it’s gone.

That works for the ecom shop. Email confirmations don’t need to be replayed. Inventory updates happen once. Most background work in most products fits this model.

A stream is different. It records events and keeps them. Multiple independent consumers can read the same event at different offsets. The history exists after the event has been processed.

This is what Kafka does. The same OrderPlaced event can be read by a fulfillment service, a fraud detection system, and an analytics pipeline, all independently, without the producer knowing or caring. A new service can join six months later and backfill from the beginning of the log.

This connects to what makes data pipelines useful in the first place: the ability to reprocess the same data with different logic. A queue can’t do that.

With the Redis queue above, adding a second consumer of OrderPlaced means adding another handler in the in-process bus. That works when all consumers run inside the same worker. It breaks when you want a genuinely separate service with its own codebase, its own deployment, and its own replay capability.

That’s the line.

Kafka earns its place when Redis runs out of answers

The Redis plus Celery setup scales further than most teams expect. It handles thousands of events per second on commodity hardware. It’s operationally simple if Redis is already running. The failure modes are predictable.

Kafka earns its complexity when multiple independent services need to consume the same events. Not the same worker with different handlers, but genuinely separate deployments, each with its own codebase and its own reason to care about OrderPlaced. The in-process bus doesn’t cross that boundary.

It earns its complexity when a new consumer needs to read historical events. A fraud model added six months after launch might need to backfill against every order ever placed. Replaying from a Redis queue isn’t possible once the messages are consumed.

It earns its complexity when ordering guarantees matter across services. Kafka partitions route events by key so all events for a given order arrive at the same consumer in sequence. A refund that lands before the charge it’s refunding isn’t just out of order. It’s wrong.

And it earns its complexity when Redis memory becomes the constraint. Kafka stores events on disk with configurable retention. Redis doesn’t.

If none of those conditions apply, you’re adding Kafka’s operational overhead without using any of the capabilities that justify it. Brokers to manage, partitions to configure, consumer group offsets to reason about, retention policies to set.

The same tradeoff appears in observability: more infrastructure adds capability and cost. The question is whether the capability is needed now.

The right tool for the scale you have

The API publishes events and returns. The worker processes them independently. Handlers are isolated by domain. Scheduled work runs through the same queues. That separation is what matters, and it works with Redis and Celery today.

If Redis stops covering the load, or a second independent service needs to consume the same events, the event contracts are already there. The migration to Kafka is smaller than it looks because the business logic doesn’t move.

If Redis covers the load, Kafka adds complexity without giving you anything back. The bus abstraction is already there when you need it.

Build for the scale you have. Know where the line is.