Working with Redis Streams

Conceptually, a Stream in Redis is a list where you can append entries. Each entry has a unique ID and a value. The ID is auto-generated by default, and it includes a timestamp. The value is a hash. You can query ranges or use blocking commands to read entries as they come. Typical of Redis, you can combine different ingredients to get the result you need. As Niklaus Wirth once said, programs are algorithms plus data structures, and Redis already gives you a bit of both.

Redis streams are ideal for building history preserving message brokers, message queues, unified logs, and chat systems. Unlike Pub/Sub messages which are fire and forget, Redis streams preserve messages in perpetuity. Redis streams implement consumer groups, a feature that allows a group of clients to cooperate when consuming elements from a stream. For example, consumers in a group can lookup items by ID, have to acknowledge the processing of an item, or claim ownership of a pending message. We will explore consumer groups in a future blog post.

Here's one way of sending messages to a channel on IRC:

And this is how you publish a message in Redis:

It looks extremely similar and the behavior is almost the same. Now, if you want to read messages sent to an IRC channel, you need to join it first:

Likewise if you want to read messages published in Redis:

Still very similar. In both cases, clients can read incoming messages as long as they are connected. If they want to stop receiving messages, they have commands for that too. The IRC version:

The Redis counterpart:

Note that the name #foo is significant in IRC, but not in Redis. You could have used any other string.

Now, let's implement the same behavior with Streams. Publishing a message would look like this:

What's going on here? There's a lot to explain, so let's get started:

XADD is the command we have to use for creating a stream and pushing entries.

#foo is the name of a Redis key that will represent the stream. The name is arbitrary, and we are using #foo just to mimic the channel names on IRC.

* is a special value that takes the place of an ID. It tells Redis to generate a new ID greater than any previous one.

msg is a field. As the value of an entry is a hash, you have to send at least one field and one value. In this example, we chose msg as the field.

"hey!" is the value for msg.

The command looks overly complex for this simple use case. That's because it's very flexible, as you will see below. For now, let's move on to reading messages:

The command itself doesn’t make much sense until you break it down to its components:

XREAD is the command for fetching entries.

BLOCK 1000 means the client will block if there are no entries and it will time out after 1000 milliseconds if nothing comes in.

STREAMS is a directive that accepts a list of keys followed by a list of IDs. In our example, we are sending one key and one pseudo ID. Read below.

#foo is the name of a Redis key that will represent the stream.

$ is a pseudo ID that represents any ID created after our command blocked. It means we want to ignore any previous entries in the stream, and we will focus only on entries that arrive from now on.

Whether you get an entry or the command times out, you need to call XREAD again in order to fetch more entries. That means XREAD will usually be called inside a loop. If you want to stop reading entries, you just have to break out of the loop.

You have now seen how to get the same result using IRC, Pub/Sub and Streams. There are other uses for Streams, and incidentally you have also seen how to use a stream as a log: push entries, then use the XREAD command to read the entries as they come in. The version we provided is the equivalent of using tail -f on a file. (If you are not familiar with tail -f, run man tail for more info).

Now we can model a modern chat application that lets you catch up with previous conversations. In this version, each client will be able to read all the messages, regardless if the client was connected when the messages arrived on the stream.

If you want to implement a chat where people can see previous conversations, as well as new messages when they arrive, you need the clients to keep track of the last message seen. In this case, clients need to remember the ID of the last entry they read. What a great opportunity to explore Stream IDs. You need to know how they are created and how they can be used.

Let's go back to the command for creating an entry:

If you feed it to Redis 5 you will get this reply:

1536495531827-0

That's the new entry's ID, and it follows this format:

<timestamp>-<sequence>

The first part is a timestamp in milliseconds, the second part is a sequence number. If Redis receives more than one XADD command during the same millisecond, the sequence number is incremented. That guarantees that no two entries will have the same ID, and also that any new ID will be greater than all previous IDs.

Now you can revisit the XREAD command from the point of view of a history-aware client. Let's say your client is connecting for the first time and it wants to read all previous messages. The command will look as follows:

This time the client doesn't block; it just asks for all entries in #foo with IDs great than 0-0. Just to clarify, 0-0 is not a valid ID: even if you create your own IDs instead of letting Redis generate them, Redis will reject 0-0 because it's a special value. And because its use is frequent, Redis accepts 0 as a shorthand, so you can as well write the command like this:

And the result you get from Redis:

1) 1) "#foo"
    2) 1) 1) 1536495531827-0
        2) 1) "msg"
            2) "hey!"

That's what you will see if you use redis-cli. If you call the same command from, say, Python, you get this structure:

Let's expand the result to see its actual shape:

[
    [
    "#foo",
        [
            [
            "1536495531827-0",
                [
                "msg",
                "hey!"
                ]
            ]
        ]
    ]
]

It's an array of streams (in this case, with only one element "#foo"), where each stream has an associated array of entries composed as an ID and a list of field/value pairs.

Continuing with the chat metaphor, the important piece of information you need to save is the last ID you've seen, so you have to store "1536495531827-0" in your program. When you are ready to check for newer entries, simply call XREAD again and pass the last ID you saw:

The call returned nil because there were no new entries. Had you received any new entries, you would have updated the ID of the last-seen entry and your user-facing application would have presented the newly arrived messages as unread.