Part V — Inside the Consumer Pipeline

In the previous article we followed trade events as they left the matching engine and were appended to a Redis Stream. Publishing an event, however, is only half of the story. Events become useful when external services consume them and perform work in response.

In this article we will examine the consumer side of the architecture.

From Event Stream to Consumer

#

Visually:

flowchart TD

RS[Redis Stream]
EC["External Consumers"]
CA[Consumer A]
CB[Consumer B]
CC[Consumer C]

RS --> EC

EC --> CA

EC --> CB

EC --> CC

Notice that producer part of the architecture has been omitted. That is intentional and the goal is to highlight how consumers are decoupled from producers. Both layers in the architecture don’t know about each other. The link between the two parts is Redis, which acts as the intermediary.

Consumers communicate only with Redis and read events from it, not directly from the producer.

At this level of the system, Redis is the only shared dependency between all components.

Independent Consumption with Consumer Groups

#

In this system, a consumer is a long-running process that reads trade events from Redis and performs some form of downstream work. Multiple consumers can exist at the same time, but they remain independent processes and communicate only through Redis.

When scaling this pattern, Redis Consumer Groups allow multiple independent consumption streams over the same event log. Each group maintains its own read position, meaning different downstream systems can process the same events without interfering with each other.

Visually:

flowchart TD

RS[Redis Stream]
CA[Consumer A]
GA[Group A]
CB[Consumer B]
GB[Group B]

RS --> GA --> CA

RS --> GB --> CB

Consumer Responsibilities

#

Once trade events are available in the stream, consumers are free to interpret them according to their own domain responsibilities. A consumer might:

• persist trades (or other events)
• compute analytics
• publish notifications
• update dashboards
• feed Machine Learning systems

Consumer Implementation (Python)

#

The definition of the consumer is:

# consumer.py

import redis.asyncio as redis
import json

from process import process_event

async def redis_consumer():
    """
    Reads a stream from Redis via XREAD from a consumer group, aka XREADGROUP.
    :return:
    """
    STREAM = "events_stream"
    GROUP = "analytics_group"
    CONSUMER = "analytics_worker"

    r = redis.Redis(
        host="localhost",
        port=6379,
        decode_responses=True,
        socket_timeout=None
    )

    last_id = ">"

    # create group, if it doesn't exist
    try:
        await r.xgroup_create(STREAM, GROUP, id="$", mkstream=True)
    except redis.ResponseError as err:
        if "BUSYGROUP" not in str(err):
            raise

    while True:
        messages = await r.xreadgroup(
            groupname=GROUP,
            consumername=CONSUMER,
            streams={STREAM: last_id},
            count=50,
            block=0
        )

        if not messages:
            continue

        for _, entries, in messages:
            for message_id, fields in entries:
                event = json.loads(fields["event"])

                await process_event(event)

                await r.xack(STREAM, GROUP, message_id)

The key part lives inside the while True block. The consumer runs indefinitely as part of the analytics_group consumer group. As new trade events arrive in the stream, Redis delivers them to the consumer, which processes the event and acknowledges successful handling via XACK command.

The > character is a stream identifier that instructs Redis to deliver only new messages that have not been seen by the consumer group. This allows consumers to continuously process newly generated trade events without manually tracking stream offsets.

For the sake of completion, the definition of process_event is:

async def process_event(event):
    print("\n*** Analytics happenings ***\n")
    print(f"=== Trade information === \n")
    print(f"Trade id: {event["trade_id"]}")
    print(f"Market id: {event["market_id"]}")
    print(f"Odds: {event["odds"]["ticks"]}")
    print(f"Stake: {event["stake"]["value"]}")
    print(f"Timestamp: {event["timestamp"]}")

Why a separate process?

#

Separation of concerns

#

• Matching engine matches orders (producer).

• Consumers react to events (consumers).

Failure isolation

#

Consumer crashes do not affect the producer, so the engine can keep matching.

Consumer groups also isolate downstream services from one another by maintaining independent read positions.

Language independence

#

Consumers can be written in any language.

Scalability

#

Consumers can evolve independently.

Design decisions

#

Event-driven architecture

#

The system is organised around events rather than direct service-to-service calls. Producers publish trade events once, and consumers decide independently how to react to them.

Loose coupling

#

The matching engine has no knowledge of downstream consumers, while consumers have no knowledge of how events were produced. Redis acts as the intermediary between the two layers.

Independent deployment

#

Consumers can be developed, deployed, and modified independently of the matching engine. New consumers can be introduced without changing producer code.

Replayability

#

Because events are stored in Redis Streams, newly introduced consumers can process historical events rather than only future ones, subject to stream retention policy. This is a useful property when introducing new consumer services.

Limitations and future work

#

The consumer implemented here is intentionally simplified and merely logs received events. Real systems would likely persist data, compute analytics, expose APIs, or publish websocket updates back to clients.

At this point the end-to-end architecture is complete. Orders enter the matching engine, trades are generated deterministically, events are streamed through Redis, and independent consumers react to those events without affecting the matching path.

Together, these components form a simple event-driven exchange architecture that cleanly separates matching, event distribution, and downstream processing.