Redis Streams with Jedis

Welcome back! In this lesson, we explore Redis Streams, a powerful data structure for processing real-time data efficiently. Streams are ideal for use cases like:

• Chat Applications: Handling messages in real-time.
• Monitoring Systems: Processing logs and events.
• User Activity Tracking: Tracking user actions as they happen.

By the end of this lesson, you’ll know how to add entries to a Redis Stream and read from it using Jedis in Java.

Redis Streams are an append-only log of events where each entry is uniquely identified and stores data as key-value pairs. They are highly versatile for real-time data processing.

Key characteristics of Redis Streams include:

• Append-Only: New entries are always added at the end of the stream.
• Unique Identifiers: Each entry has a unique ID, which can be generated automatically or manually specified.
• Field-Value Pairs: Entries store data as key-value pairs, offering flexibility in structure.
• Efficient Reads and Writes: Streams support high-performance operations, even with large data volumes.

These features make Redis Streams ideal for building scalable, real-time systems like message queues, event logs, and analytics pipelines.

The XADD command is used to append entries to a Redis Stream. If the stream does not exist, it will be created automatically.

// Adding entries to a stream (using XAddParams.xAddParams() to specify the default behavior)
jedis.xadd("mystream", XAddParams.xAddParams(), Map.of("event", "login", "user", "Alice"));
jedis.xadd("mystream", XAddParams.xAddParams(), Map.of("event", "purchase", "user", "Bob", "amount", "100"));
jedis.xadd("mystream", XAddParams.xAddParams(), Map.of("event", "add_to_cart", "user", "Alice", "product", "laptop"));

In this snippet, events like "login," "purchase," and "add_to_cart" are added to the mystream stream along with associated user details. Each entry is stored as a set of key-value pairs, using XAddParams.xAddParams() to ensure the default behavior, making it easy to track various types of data in real time.

The XREAD command retrieves entries from a Redis Stream, starting from a specified ID. This is how you consume and process the stored events.

// Create XReadParams (1 message, block indefinitely = 0 ms)
XReadParams xReadParams = XReadParams.xReadParams().count(1).block(0);

// Reading entries from the stream: must pass Map<String, StreamEntryID>
List<Map.Entry<String, List<StreamEntry>>> messages =
    jedis.xread(xReadParams, Map.of("mystream", new StreamEntryID("0-0")));

System.out.println("Stream messages:");
for (Map.Entry<String, List<StreamEntry>> stream : messages) {
    for (StreamEntry entry : stream.getValue()) {
        System.out.println("ID: " + entry.getID() + " Fields: " + entry.getFields());
    }
}

Here, XREAD fetches messages with the specified XReadParams, starting from the beginning of the mystream stream (0-0 is the starting ID). Note that although we start reading from ID 0-0, the actual ID of the first entry in the stream will differ, as IDs are stored as timestamps by default.

The .block(0) parameter makes the XREAD command wait indefinitely for new entries if none are available, which is useful in real-time processing scenarios. Each entry is identified by a unique ID and contains its fields and values. This allows you to process events sequentially while maintaining order and reliability.