Redis Streams with Java Using Lettuce API

Welcome back! In this lesson we will dive into Redis Streams — a powerful feature used for processing streams of data. This lesson will guide you through the basics and show you how Redis Streams can be essential for high-performance applications.

In this lesson, we'll learn about streams in Redis and how they can be used to handle continuous data flows. We'll learn how to create streams, add events to them, and read events from them.

Streams are a powerful data structure that allows you to process real-time data efficiently. Redis Streams differ from Lists as they are designed for time-ordered event storage and retrieval, allowing consumers to read and process messages at their own pace. Unlike Pub/Sub, Redis Streams retain messages for replay and provide more control over message acknowledgment and processing.

Here are a few real-world scenarios where Redis Streams can be useful:

• Chat Applications: Redis Streams can be used to handle messages in real-time.
• Monitoring Systems: Streams can be used to process logs and events.
• User Activity Tracking: Streams can be used to track user actions in real-time.

Let's dive into the details!

Here’s a quick preview:

• To add an event to a stream, you can use the xadd command.
• To read events from a stream, you can use the xread command. Use the xread command when you need to continuously read new entries from a Redis stream in order to handle real-time data as it arrives.

Let's see how these commands work in practice.

import io.lettuce.core.RedisClient;
import io.lettuce.core.StreamMessage;
import io.lettuce.core.XReadArgs.StreamOffset;
import io.lettuce.core.api.StatefulRedisConnection;
import io.lettuce.core.api.sync.RedisCommands;

public class Main {

    public static void main(String[] args) {
        // Connect to Redis server
        RedisClient redisClient = RedisClient.create("redis://localhost:6379");
        StatefulRedisConnection<String, String> connection = redisClient.connect();
        RedisCommands<String, String> syncCommands = connection.sync();

        // Add events to the stream
        syncCommands.xadd("mystream", "event", "login", "user", "Alice");
        syncCommands.xadd("mystream", "event", "purchase", "user", "Bob", "amount", "100");
        syncCommands.xadd("mystream", "event", "add_to_cart", "user", "Alice", "product", "laptop");

        // Read messages from the stream
        var messages = syncCommands.xread(
            StreamOffset.from("mystream", "0") // Start from the beginning
        );

        // Print the retrieved messages
        for (StreamMessage<String, String> message : messages) {
            System.out.println("ID: " + message.getId());
            message.getBody().forEach((key, value) ->
                System.out.println("  " + key + ": " + value));
        }

        // Close connection
        connection.close();
        redisClient.shutdown();
    }
}

The above code snippet demonstrates how to add events to a Redis stream called mystream. Each event contains key-value pairs representing different actions by users.

The code subsequently reads messages from mystream starting from the beginning, using the xread command. It then iterates over the retrieved messages, printing each message's ID and the key-value pairs contained in the message body.