VALKEY-STREAMS-INTRO(7)

📦

version 9.0.6 · 2 versions

9.0.6 · fedora 44 (latest)
8.1.10 · fedora 43

valkey-streams-intro(7) — the Valkey Manual, but make it a group chat 🫶

✨ NAME ✨

valkey-streams-intro — Streams 💧

🌈 DESCRIPTION 🌈

okay so bestie, picture a data structure that's basically an append-only log but with ✨main character energy✨. That's a Stream. It acts like an append-only log, yes, but it also implements several operations to overcome some of the limits of a typical append-only log — like random access in O(1) time and complex consumption strategies, such as consumer groups. You can use streams to record and simultaneously syndicate events in real time. 💅

Examples of Stream use cases include:

Event sourcing (e.g., tracking user actions, clicks, etc.) 🖱️
Sensor monitoring (e.g., readings from devices in the field) 📡
Notifications (e.g., storing a record of each user's notifications in a separate stream) 🔔

Valkey generates a unique ID for each stream entry. You can use these IDs to retrieve their associated entries later, or to read and process all subsequent entries in the stream. Note that because these IDs are related to time, the ones shown here may vary and will be different from the IDs you see in your own Valkey instance. no cap. 🕐

Streams support several trimming strategies (to prevent streams from growing unbounded) and more than one consumption strategy (see XREAD, XREADGROUP, and XRANGE). 🎀

🎯 Basic commands 🎯

XADD adds a new entry to a stream. ✍️
XREAD reads one or more entries, starting at a given position and moving forward in time. ⏩
XRANGE returns a range of entries between two supplied entry IDs. 📏
XLEN returns the length of a stream. 🔢

See the valkey-commands(7) complete list of stream commands. 📚

🌸 EXAMPLES 🌸

When our racers pass a checkpoint, we add a stream entry for each racer that includes the racer's name, speed, position, and location ID: 🏁

127.0.0.1:6379> XADD race:france * rider Castilla speed 30.2 position 1 location_id 1
"1692632086370-0"
127.0.0.1:6379> XADD race:france * rider Norem speed 28.8 position 3 location_id 1
"1692632094485-0"
127.0.0.1:6379> XADD race:france * rider Prickett speed 29.7 position 2 location_id 1
"1692632102976-0"

Read two stream entries starting at ID 1692632086370-0: 👀

127.0.0.1:6379> XRANGE race:france 1692632086370-0 + COUNT 2
1) 1) "1692632086370-0"
   2) 1) "rider"
      2) "Castilla"
      3) "speed"
      4) "30.2"
      5) "position"
      6) "1"
      7) "location_id"
      8) "1"
2) 1) "1692632094485-0"
   2) 1) "rider"
      2) "Norem"
      3) "speed"
      4) "28.8"
      5) "position"
      6) "3"
      7) "location_id"
      8) "1"

Read up to 100 new stream entries, starting at the end of the stream, and block for up to 300 ms if no entries are being written: ⏳

127.0.0.1:6379> XREAD COUNT 100 BLOCK 300 STREAMS race:france $
(nil)

⚡ Performance ⚡

Adding an entry to a stream is O(1). Accessing any single entry is O(n), where n is the length of the ID. Since stream IDs are typically short and of a fixed length, this effectively reduces to a constant time lookup. For details on why, note that streams are implemented as radix trees. 🌳

Simply put, Streams provide highly efficient inserts and reads. See each command's time complexity for the details. 🔥

🧱 Streams basics 🧱

Streams are an append-only data structure. The fundamental write command, called XADD, appends a new entry to the specified stream. ✍️

Each stream entry consists of one or more field-value pairs, somewhat like a dictionary or a Hash:

127.0.0.1:6379> XADD race:france * rider Castilla speed 29.9 position 1 location_id 2
"1692632147973-0"

The above call to the XADD command adds an entry rider: Castilla, speed: 29.9, position: 1, location_id: 2 to the stream at key race:france, using an auto-generated entry ID, which is the one returned by the command, specifically 1692632147973-0. It gets as its first argument the key name race:france, the second argument is the entry ID that identifies every entry inside a stream. However, in this case, we passed * because we want the server to generate a new ID for us. Every new ID will be monotonically increasing, so in more simple terms, every new entry added will have a higher ID compared to all the past entries. Auto-generation of IDs by the server is almost always what you want, and the reasons for specifying an ID explicitly are very rare. We'll talk more about this later. The fact that each Stream entry has an ID is another similarity with log files, where line numbers, or the byte offset inside the file, can be used in order to identify a given entry. Returning back at our XADD example, after the key name and ID, the next arguments are the field-value pairs composing our stream entry. 💖

It is possible to get the number of items inside a Stream just using the XLEN command:

127.0.0.1:6379> XLEN race:france
(integer) 4

🆔 Entry IDs 🆔

The entry ID returned by the XADD command, and identifying univocally each entry inside a given stream, is composed of two parts:

<millisecondsTime>-<sequenceNumber>

The milliseconds time part is actually the local time in the local Valkey node generating the stream ID, however if the current milliseconds time happens to be smaller than the previous entry time, then the previous entry time is used instead, so if a clock jumps backward the monotonically incrementing ID property still holds. The sequence number is used for entries created in the same millisecond. Since the sequence number is 64 bit wide, in practical terms there is no limit to the number of entries that can be generated within the same millisecond. ⏱️

The format of such IDs may look strange at first, and the gentle reader may wonder why the time is part of the ID. The reason is that Streams support range queries by ID. Because the ID is related to the time the entry is generated, this gives the ability to query for time ranges basically for free. We will see this soon while covering the XRANGE command. 🕰️

If for some reason the user needs incremental IDs that are not related to time but are actually associated to another external system ID, as previously mentioned, the XADD command can take an explicit ID instead of the * wildcard ID that triggers auto-generation, like in the following examples:

127.0.0.1:6379> XADD race:usa 0-1 racer Castilla
0-1
127.0.0.1:6379> XADD race:usa 0-2 racer Norem
0-2

Note that in this case, the minimum ID is 0-1 and that the command will not accept an ID equal or smaller than a previous one:

127.0.0.1:6379> XADD race:usa 0-1 racer Prickett
(error) ERR The ID specified in XADD is equal or smaller than the target stream top item

If you're running Redis OSS 7 or later, you can also provide an explicit ID consisting of the milliseconds part only. In this case, the sequence portion of the ID will be automatically generated. To do this, use the syntax below:

127.0.0.1:6379> XADD race:usa 0-* racer Prickett
0-3

📖 Getting data from Streams 📖

Now we are finally able to append entries in our stream via XADD. However, while appending data to a stream is quite obvious, the way streams can be queried in order to extract data is not so obvious. If we continue with the analogy of the log file, one obvious way is to mimic what we normally do with the Unix command tail -f, that is, we may start to listen in order to get the new messages that are appended to the stream. Note that unlike the blocking list operations of Valkey, where a given element will reach a single client which is blocking in a pop style operation like BLPOP, with streams we want multiple consumers to see the new messages appended to the stream (the same way many tail -f processes can see what is added to a log). Using the traditional terminology we want the streams to be able to fan out messages to multiple clients. 📢

However, this is just one potential access mode. We could also see a stream in quite a different way: not as a messaging system, but as a time series store. In this case, maybe it's also useful to get the new messages appended, but another natural query mode is to get messages by ranges of time, or alternatively to iterate the messages using a cursor to incrementally check all the history. This is definitely another useful access mode. 📊

Finally, if we see a stream from the point of view of consumers, we may want to access the stream in yet another way, that is, as a stream of messages that can be partitioned to multiple consumers that are processing such messages, so that groups of consumers can only see a subset of the messages arriving in a single stream. In this way, it is possible to scale the message processing across different consumers, without single consumers having to process all the messages: each consumer will just get different messages to process. This is basically what Kafka (TM) does with consumer groups. Reading messages via consumer groups is yet another interesting mode of reading from a Stream. 🧑‍🤝‍🧑

Streams support all three of the query modes described above via different commands. The next sections will show them all, starting from the simplest and most direct to use: range queries. ✨

🔍 Querying by range: XRANGE and XREVRANGE 🔍

To query the stream by range we are only required to specify two IDs, start and end. The range returned will include the elements having start or end as ID, so the range is inclusive. The two special IDs - and + respectively mean the smallest and the greatest ID possible. 🎯

127.0.0.1:6379> XRANGE race:france - +
1) 1) "1692632086370-0"
   2) 1) "rider"
      2) "Castilla"
      3) "speed"
      4) "30.2"
      5) "position"
      6) "1"
      7) "location_id"
      8) "1"
2) 1) "1692632094485-0"
   2) 1) "rider"
      2) "Norem"
      3) "speed"
      4) "28.8"
      5) "position"
      6) "3"
      7) "location_id"
      8) "1"
3) 1) "1692632102976-0"
   2) 1) "rider"
      2) "Prickett"
      3) "speed"
      4) "29.7"
      5) "position"
      6) "2"
      7) "location_id"
      8) "1"
4) 1) "1692632147973-0"
   2) 1) "rider"
      2) "Castilla"
      3) "speed"
      4) "29.9"
      5) "position"
      6) "1"
      7) "location_id"
      8) "2"

Each entry that comes back is an array of two things: the ID, and the list of field-value pairs. 💖

We already said the entry IDs have a whole relationship with time, because the part on the left of the - character is the Unix time in milliseconds of the local node that created the stream entry, at the exact moment the entry got created (though note: streams are replicated with fully specified XADD commands, so the replicas end up with IDs identical to the primary's). Which means I could totally query a range of time using XRANGE. To do that though, I might want to drop the sequence part of the ID: if you omit it, at the start of the range it's assumed to be 0, and at the end of the range it's assumed to be the maximum sequence number available. So querying with just two millisecond Unix times gets you every entry that was generated in that range of time, inclusive bestie. 🌸 For instance, if I want to query a two milliseconds period I could use:

127.0.0.1:6379> XRANGE race:france 1692632086369 1692632086371
1) 1) "1692632086370-0"
   2) 1) "rider"
      2) "Castilla"
      3) "speed"
      4) "30.2"
      5) "position"
      6) "1"
      7) "location_id"
      8) "1"

I've only got a single entry in this range. But in real data sets I could be querying ranges of hours, or there could be a ton of items in just two milliseconds, and the result returned could be huge. That's literally why XRANGE supports an optional COUNT option at the end. By specifying a count, I can just get the first N items. Want more? Grab the last ID returned, increment the sequence part by one, and query again. 👑

Let's see this in the following example. Let's assume the stream race:france was populated with 4 items. To start my iteration, getting 2 items per command, I start with the full range, but with a count of 2.

127.0.0.1:6379> XRANGE race:france - + COUNT 2
1) 1) "1692632086370-0"
   2) 1) "rider"
      2) "Castilla"
      3) "speed"
      4) "30.2"
      5) "position"
      6) "1"
      7) "location_id"
      8) "1"
2) 1) "1692632094485-0"
   2) 1) "rider"
      2) "Norem"
      3) "speed"
      4) "28.8"
      5) "position"
      6) "3"
      7) "location_id"
      8) "1"

To continue the iteration with the next two items, I have to pick the last ID returned, which is 1692632094485-0, and add the prefix ( to it. The resulting exclusive range interval — that's (1692632094485-0 in this case — can now be used as the new start argument for the next XRANGE call:

127.0.0.1:6379> XRANGE race:france (1692632094485-0 + COUNT 2
1) 1) "1692632102976-0"
   2) 1) "rider"
      2) "Prickett"
      3) "speed"
      4) "29.7"
      5) "position"
      6) "2"
      7) "location_id"
      8) "1"
2) 1) "1692632147973-0"
   2) 1) "rider"
      2) "Castilla"
      3) "speed"
      4) "29.9"
      5) "position"
      6) "1"
      7) "location_id"
      8) "2"

Now that we've retrieved 4 items out of a stream that only had 4 entries in it, if we try to retrieve more items, we'll get an empty array:

127.0.0.1:6379> XRANGE race:france (1692632147973-0 + COUNT 2
(empty array)

Since XRANGE complexity is O(log(N)) to seek, and then O(M) to return M elements, with a small count the command has a logarithmic time complexity, which means every step of the iteration is fast. So XRANGE is also the de facto streams iterator and it does not require an XSCAN command. slay. 💅

The command XREVRANGE is the equivalent of XRANGE but returning the elements in inverted order, so a practical use for XREVRANGE is checking what the last item in a Stream is:

127.0.0.1:6379> XREVRANGE race:france + - COUNT 1
1) 1) "1692632147973-0"
   2) 1) "rider"
      2) "Castilla"
      3) "speed"
      4) "29.9"
      5) "position"
      6) "1"
      7) "location_id"
      8) "2"

Note that the XREVRANGE command takes the start and stop arguments in reverse order. 🙈

🎧 Listening for new items with XREAD

When we don't want to access items by a range in a stream, usually what we want instead is to subscribe to new items arriving to the stream. ✨ This concept may look related to Valkey Pub/Sub, where you subscribe to a channel, or to Valkey blocking lists, where you wait for a key to get new elements to fetch — but there are fundamental differences in the way you consume a stream:

A stream can have multiple clients (consumers) waiting for data. Every new item, by default, will be delivered to every consumer that is waiting for data in a given stream. This behavior is different from blocking lists, where each consumer will get a different element. However, the ability to fan out to multiple consumers is similar to Pub/Sub. 🫶
While in Pub/Sub messages are fire and forget and are never stored anyway, and while when using blocking lists a message is popped (effectively removed) from the list the moment the client receives it — streams work in a fundamentally different way. All the messages are appended in the stream indefinitely (unless the user explicitly asks to delete entries): different consumers know what a new message is from their own point of view by remembering the ID of the last message received. 🧠
Streams Consumer Groups provide a level of control that Pub/Sub or blocking lists cannot achieve, with different groups for the same stream, explicit acknowledgment of processed items, ability to inspect the pending items, claiming of unprocessed messages, and coherent history visibility for each single client, which is only able to see its private past history of messages. 👑

The command that gives you the ability to listen for new messages arriving into a stream is called XREAD. It's a bit more complex than XRANGE, so we'll start by showing the simple forms, and later the whole command layout will be provided.

127.0.0.1:6379> XREAD COUNT 2 STREAMS race:france 0
1) 1) "race:france"
   2) 1) 1) "1692632086370-0"
         2) 1) "rider"
            2) "Castilla"
            3) "speed"
            4) "30.2"
            5) "position"
            6) "1"
            7) "location_id"
            8) "1"
      2) 1) "1692632094485-0"
         2) 1) "rider"
            2) "Norem"
            3) "speed"
            4) "28.8"
            5) "position"
            6) "3"
            7) "location_id"
            8) "1"

That right there is the non-blocking form of XREAD. Note that the COUNT option is not mandatory — in fact the only mandatory option of the command is the STREAMS option, which specifies a list of keys together with the corresponding maximum ID already seen for each stream by the calling consumer, so that the command will provide the client only with messages having an ID greater than the one we specified. 🎀

In the above command we wrote STREAMS race:france 0, so we want all the messages in the Stream race:france having an ID greater than 0-0. And as you can see in the example above, the command returns the key name, because it's actually possible to call this command with more than one key to read from different streams at the same time. I could write, for instance: STREAMS race:france race:italy 0 0. Notice how after the STREAMS option we need to provide the key names, and later the IDs. Because of that, the STREAMS option must always be the last option. Any other options must come before the STREAMS option. ⚠️

Apart from the fact that XREAD can access multiple streams at once, and that we're able to specify the last ID we own to just get newer messages, in this simple form the command isn't doing something so different compared to XRANGE. But the interesting part? We can turn XREAD into a blocking command easily, by specifying the BLOCK argument:

> XREAD BLOCK 0 STREAMS race:france $

Note that in the example above, besides removing COUNT, I specified the new BLOCK option with a timeout of 0 milliseconds (which means to never timeout). Also, instead of passing a normal ID for the stream race:france, I passed the special ID $. This special ID means that XREAD should use as last ID the maximum ID already stored in the stream race:france, so that we'll receive only new messages, starting from the time we started listening. It's similar to the tail -f Unix command in some way. 🐟

Note that when the BLOCK option is used, we do not have to use the special ID $. We can use any valid ID. If the command is able to serve our request immediately without blocking, it will do so, otherwise it will block. Normally, if we want to consume the stream starting from new entries, we start with the ID $, and after that we keep using the ID of the last message received to make the next call, and so forth. 🔁

The blocking form of XREAD is also able to listen to multiple Streams, just by specifying multiple key names. If the request can be served synchronously because there is at least one stream with elements greater than the corresponding ID we specified, it returns with the results. Otherwise, the command will block and will return the items of the first stream which gets new data (according to the specified ID).

Similarly to blocking list operations, blocking stream reads are fair from the point of view of clients waiting for data, since the semantics is FIFO style. The first client that blocked for a given stream will be the first to be unblocked when new items are available. 🥇

XREAD has no other options than COUNT and BLOCK, so it's a pretty basic command with a specific purpose: attaching consumers to one or multiple streams. More powerful features to consume streams are available using the consumer groups API — however, reading via consumer groups is implemented by a different command called XREADGROUP, covered in the next section of this guide. 🦋

💖 Consumer groups 💖

Okay so here's the tea, bestie: when the task at hand is consuming the same stream from a bunch of different clients, XREAD already gives you a way to fan-out to N clients — and you can even throw replicas at it to get more read scalability. Slay. 🔥

But sometimes that's not what you want. Sometimes you don't want to hand the identical stream of messages to many clients — you want to hand a different subset of messages from the same stream to many clients. The most obvious case where this eats: messages that are slow to process. If you can have N different workers each receiving a different part of the stream, you can scale up your message processing by routing different messages to different workers that are ready to do more work. Iconic, honestly. 💅

Practically speaking, imagine three consumers C1, C2, C3 and a stream holding the messages 1, 2, 3, 4, 5, 6, 7. What we want is to serve the messages like this:

1 -> C1
2 -> C2
3 -> C3
4 -> C1
5 -> C2
6 -> C3
7 -> C1

To make this happen, Valkey uses a concept called consumer groups. 🎀 And this part is super important: from an implementation standpoint, Valkey consumer groups have nothing to do with Kafka (TM) consumer groups. Like, zero. Nada. But they are similar in functionality, so the terminology from Kafka (TM) got kept around, because Kafka (TM) is what originally popularized this whole idea.

So a consumer group is basically like a pseudo consumer that gets data from a stream and then actually serves multiple consumers, and it comes with some guarantees. Bestie, read these, they matter:

Every message gets served to a different consumer, so it is not possible for the same message to be delivered to multiple consumers. No double-dipping. 🙈
Consumers are identified within a consumer group by a name — a case-sensitive string that the clients implementing consumers must choose. This means that even after a disconnect, the stream consumer group retains all the state, because the client will just claim again to be the same consumer. But heads up: this also means it's up to the client to provide a unique identifier. 👑
Each consumer group has this concept of the first ID never consumed, so that when a consumer asks for new messages, it can be handed just the messages that weren't previously delivered.
Consuming a message, though, requires an explicit acknowledgment using a specific command. Valkey reads that acknowledgment as: "this message was correctly processed so it can be evicted from the consumer group." Periodt.
A consumer group tracks all the messages that are currently pending — meaning messages that were delivered to some consumer of the consumer group but haven't been acknowledged as processed yet. Thanks to this feature, when you access the message history of a stream, each consumer will only see the messages that were delivered to it. 🫶

In a way, you can imagine a consumer group as a little pile of state about a stream:

+----------------------------------------+
| consumer_group_name: mygroup           |
| consumer_group_stream: somekey         |
| last_delivered_id: 1292309234234-92    |
|                                        |
| consumers:                             |
|    "consumer-1" with pending messages  |
|       1292309234234-4                  |
|       1292309234232-8                  |
|    "consumer-42" with pending messages |
|       ... (and so forth)               |
+----------------------------------------+

Look at it from that angle and it's honestly so simple to understand what a consumer group can do — how it's able to just hand consumers their history of pending messages, and how consumers asking for new messages will just get served message IDs greater than last_delivered_id. ✨

And at the same time, if you look at the consumer group as an auxiliary data structure for Streams, it's pretty obvious that a single stream can have multiple consumer groups, each with a different set of consumers. Actually — and this is the fun part — it's even possible for the same stream to have clients reading without consumer groups via XREAD, and clients reading via XREADGROUP in different consumer groups. All at once. Main character stream energy. 💅

Now let's zoom in on the fundamental consumer group commands, because these are the ones you'll actually be typing:

XGROUP — used in order to create, destroy and manage consumer groups.
XREADGROUP — used to read from a stream via a consumer group.
XACK — the command that lets a consumer mark a pending message as correctly processed.

🎀 Creating a consumer group 🎀

Okay so let's say I've already got a key race:france of type stream just chilling there. To create a consumer group I just do this:

127.0.0.1:6379> XGROUP CREATE race:france france_riders $
OK

See how, when creating the consumer group, we have to specify an ID? In the example it's just $. This is needed because the consumer group, among all its other state, needs an idea of what message to serve next when the first consumer connects — i.e. what the last message ID was back when the group was just created.

If we give it $, like we just did, then only new messages arriving in the stream from now on will be provided to the consumers in the group.
If we specify 0 instead, the consumer group will consume all the messages in the stream history to start with.
And of course you can specify any other valid ID. What you know for sure is that the consumer group will start delivering messages that are greater than the ID you specify. Since $ means the current greatest ID in the stream, specifying $ has the effect of consuming only new messages. 💖

XGROUP CREATE also supports creating the stream automatically if it doesn't exist, using the optional MKSTREAM subcommand as the last argument:

127.0.0.1:6379> XGROUP CREATE race:italy italy_riders $ MKSTREAM
OK

Now that the consumer group exists, we can immediately try reading messages through it with the XREADGROUP command. Let's read from two consumers, who we'll call Alice and Bob, and watch the system hand different messages to Alice or Bob. 🦋

XREADGROUP is very similar to XREAD and provides the same BLOCK option — otherwise it's a synchronous command. But there's a mandatory option that always has to be specified, and that's GROUP, which takes two arguments: the name of the consumer group, and the name of the consumer that is attempting to read. The COUNT option is also supported and is identical to the one in XREAD.

So let's add some riders to the race:italy stream and try reading something using the consumer group. Quick note: here rider is the field name, and the name is the associated value. Remember, stream items are small dictionaries. 🧚

127.0.0.1:6379> XADD race:italy * rider Castilla
"1692632639151-0"
127.0.0.1:6379> XADD race:italy * rider Royce
"1692632647899-0"
127.0.0.1:6379> XADD race:italy * rider Sam-Bodden
"1692632662819-0"
127.0.0.1:6379> XADD race:italy * rider Prickett
"1692632670501-0"
127.0.0.1:6379> XADD race:italy * rider Norem
"1692632678249-0"
127.0.0.1:6379> XREADGROUP GROUP italy_riders Alice COUNT 1 STREAMS race:italy >
1) 1) "race:italy"
   2) 1) 1) "1692632639151-0"
         2) 1) "rider"
            2) "Castilla"

XREADGROUP replies are just like XREAD replies. But notice the GROUP <group-name> <consumer-name> we passed above. It's saying: I want to read from the stream using the consumer group mygroup and I'm the consumer Alice. Every single time a consumer performs an operation with a consumer group, it must specify its name, uniquely identifying this consumer inside the group. No anonymity allowed. 💅

There's another very important detail in that command line up there: after the mandatory STREAMS option, the ID requested for the key race:italy is the special ID >. That special ID is only valid in the context of consumer groups, and it means: messages never delivered to other consumers so far. 🌸

This is almost always what you want. That said, it's also possible to specify a real ID, like 0 or any other valid ID — but in that case, what happens is that we're asking XREADGROUP to just give us the history of pending messages, and in that case we will never see new messages in the group. So basically XREADGROUP behaves like this depending on the ID we give it:

If the ID is the special ID > then the command will return only new messages never delivered to other consumers so far, and as a side effect, it will update the consumer group's last ID.
If the ID is any other valid numerical ID, then the command lets us access our history of pending messages. That is, the set of messages that were delivered to this specified consumer (identified by the provided name), and never acknowledged so far with XACK.

Let's test that behavior immediately by specifying an ID of 0, without any COUNT option. We'll just see the only pending message — the one about Castilla:

127.0.0.1:6379> XREADGROUP GROUP italy_riders Alice STREAMS race:italy 0
1) 1) "race:italy"
   2) 1) 1) "1692632639151-0"
         2) 1) "rider"
            2) "Castilla"

But if we acknowledge that message as processed, it's no longer part of the pending messages history, so the system won't report anything at all:

127.0.0.1:6379> XACK race:italy italy_riders 1692632639151-0
(integer) 1
127.0.0.1:6379> XREADGROUP GROUP italy_riders Alice STREAMS race:italy 0
1) 1) "race:italy"
   2) (empty array)

Don't worry if you don't know how XACK works yet — the whole idea here is just that processed messages are no longer part of the history we can access. Simple as that. 🫶

Okay now it's Bob's turn to read something:

127.0.0.1:6379> XREADGROUP GROUP italy_riders Bob COUNT 2 STREAMS race:italy >
1) 1) "race:italy"
   2) 1) 1) "1692632647899-0"
         2) 1) "rider"
            2) "Royce"
      2) 1) "1692632662819-0"
         2) 1) "rider"
            2) "Sam-Bodden"

Bob asked for a maximum of two messages and is reading via the same group mygroup. So what happens is that Valkey reports just new messages. As you can see, the "Castilla" message is not delivered, since it was already delivered to Alice — so Bob gets Royce and Sam-Bodden, and so forth. 🌈

So Alice, Bob, and literally every other consumer in the group can each read different messages off the same stream, peek at their own lil backlog of unprocessed stuff, or mark things as done. That's the whole point. It means you can build all sorts of different topologies and semantics for how you consume from a stream. Iconic, honestly. 💅

But bestie, there's a few things you gotta keep in your head:

Consumers are auto-created the first time you mention them. No explicit creation step, no ceremony, they just exist now. ✨
Even with XREADGROUP you can read from multiple keys at the same time — but for that to actually work, you need to create a consumer group with the same name in every stream. Not a common need, but lowkey worth mentioning that the feature is technically available. 🫶
XREADGROUP is a write command. Yes, even though it reads from the stream. The consumer group gets modified as a side effect of reading, so it can only be called on primary instances. No cap. ⚠️

Here's a whole consumer implementation using consumer groups, written in Ruby. It's written to be readable by basically any experienced programmer, even if they've never touched Ruby in their life:

require 'redis'

if ARGV.length == 0
    puts "Please specify a consumer name"
    exit 1
end

ConsumerName = ARGV[0]
GroupName = "mygroup"
r = Redis.new

def process_message(id,msg)
    puts "[#{ConsumerName}] #{id} = #{msg.inspect}"
end

$lastid = '0-0'

puts "Consumer #{ConsumerName} starting..."
check_backlog = true
while true
    # Pick the ID based on the iteration: the first time we want to
    # read our pending messages, in case we crashed and are recovering.
    # Once we consumed our history, we can start getting new messages.
    if check_backlog
        myid = $lastid
    else
        myid = '>'
    end

    items = r.xreadgroup('GROUP',GroupName,ConsumerName,'BLOCK','2000','COUNT','10','STREAMS',:my_stream_key,myid)

    if items == nil
        puts "Timeout!"
        next
    end

    # If we receive an empty reply, it means we were consuming our history
    # and that the history is now empty. Let's start to consume new messages.
    check_backlog = false if items[0][1].length == 0

    items[0][1].each{|i|
        id,fields = i

        # Process the message
        process_message(id,fields)

        # Acknowledge the message as processed
        r.xack(:my_stream_key,GroupName,id)

        $lastid = id
    }
end

Okay so the vibe here? You start by consuming your history — aka your list of pending messages. Why? Because your consumer might've crashed before, and on restart you wanna re-read the messages that got delivered to you but never got acknowledged. 💖

Note: you might process a message multiple times, or exactly one time. At least in the case of consumer failures — but there's also the limits of Valkey persistence and replication in play, so go check the specific section about that topic. 🙈

Once the history is fully consumed and you get back an empty list of messages, you can switch over to using the > special ID and start slurping up brand-new messages. ✨

🌸 Recovering from permanent failures 🌸

The example up there gives us consumers that all live in the same consumer group, each grabbing a subset of messages to process, and on recovery re-reading the pending messages that were delivered just to them. Cute. 🎀

But real world moment: consumers can permanently fail and never come back. So what happens to the pending messages of a consumer that never recovers after stopping for any reason? 😭

Valkey consumer groups got a feature for exactly this. It lets you claim the pending messages of a given consumer so those messages change ownership and get re-assigned to a different consumer. And it's very explicit — a consumer has to inspect the list of pending messages and claim specific messages using a special command. Otherwise the server will leave those messages pending forever, still assigned to the old consumer. This way different applications get to choose whether they use this feature at all, and exactly how. 💅

First step is just a command that gives you observability into the pending entries in the consumer group. It's called XPENDING. It's read-only, always safe to call, and it will not change ownership of any message. In its simplest form you call it with two arguments: the name of the stream and the name of the consumer group.

127.0.0.1:6379> XPENDING race:italy italy_riders
1) (integer) 2
2) "1692632647899-0"
3) "1692632662819-0"
4) 1) 1) "Bob"
      2) "2"

Called like this, it tells you the total number of pending messages in the consumer group (2 here), the lower and higher message ID among the pending ones, and then a list of consumers plus how many pending messages each of them has. Only Bob shows up with two pending messages, because the single message Alice requested got acknowledged with XACK. Slay. ✅

You can ask for more info by passing more arguments to XPENDING, because the full command signature is:

XPENDING <key> <groupname> [[IDLE <min-idle-time>] <start-id> <end-id> <count> [<consumer-name>]]

Give it a start and end ID (and yeah, those can just be - and + like in XRANGE) plus a count to control how much info comes back, and now you know way more about your pending messages. There's also an optional final argument — the consumer name — which you'd use if you only want messages pending for one specific consumer. We won't use that feature in the next example tho.

127.0.0.1:6379> XPENDING race:italy italy_riders - + 10
1) 1) "1692632647899-0"
   2) "Bob"
   3) (integer) 74642
   4) (integer) 1
2) 1) "1692632662819-0"
   2) "Bob"
   3) (integer) 74642
   4) (integer) 1

Now we've got the deets for each message: the ID, the consumer name, the idle time in milliseconds (how many ms have passed since the message was last delivered to some consumer), and finally the number of times a given message has been delivered. Two messages from Bob, idle for 60000+ milliseconds — about a minute. ⏳

And bestie, nobody's stopping you from checking what the first message actually said — just use XRANGE. 🦋

127.0.0.1:6379> XRANGE race:italy 1692632647899-0 1692632647899-0
1) 1) "1692632647899-0"
   2) 1) "rider"
      2) "Royce"

You just repeat the same ID twice in the arguments. That's it. 💖

Okay so now that we have some ideas — Alice might decide that after 1 minute of not processing messages, Bob is probably not recovering quickly, and it's time to claim those messages and resume processing in his place. To do that, we use the XCLAIM command. 👑

This command is very complex and absolutely full of options in its full form, since it's used for replication of consumer group changes. But we'll just use the arguments we normally need. Here it is in its simple form:

XCLAIM <key> <group> <consumer> <min-idle-time> <ID-1> <ID-2> ... <ID-N>

Basically: for this specific key and group, I want the message IDs listed to change ownership and get assigned to the specified consumer name <consumer>. But we also hand it a minimum idle time, so the operation only goes through if the idle time of those messages is greater than the specified idle time. Why? Because maybe two clients are retrying to claim the same message at the same time:

Client 1: XCLAIM race:italy italy_riders Alice 60000 1692632647899-0
Client 2: XCLAIM race:italy italy_riders Lora 60000 1692632647899-0

And here's the sneaky part — as a side effect, claiming a message resets its idle time and increments its deliveries counter, so the second client will fail to claim it. That's how we dodge trivial re-processing of messages. (Even though in the general case you can't get exactly once processing. Sorry bestie. 💀)

Here's what running it looks like:

127.0.0.1:6379> XCLAIM race:italy italy_riders Alice 60000 1692632647899-0
1) 1) "1692632647899-0"
   2) 1) "rider"
      2) "Royce"

The message got successfully claimed by Alice, who can now process it and acknowledge it, and keep things moving even though the original consumer is never coming back. Ate. 🫶

Now, notice how as a side effect of successfully claiming a message, XCLAIM also returns it. That's not mandatory tho. There's a JUSTID option that makes it return only the IDs of the successfully claimed messages. Useful when you wanna cut down the bandwidth between client and server (and also for the performance of the command) and you genuinely don't care about the message body — like when your consumer is built to rescan the history of pending messages every so often anyway. 🎀

Claiming can also be done by a separate process — one that just checks the list of pending messages and assigns idle messages to consumers that look active. You can find active consumers using one of the observability features of Streams. Which brings us to the next section… ✨

🌈 Automatic claiming 🌈

The XAUTOCLAIM command, added in Redis OSS 6.2, implements that whole claiming process we just described. XPENDING and XCLAIM give you the basic building blocks for different types of recovery mechanisms — this one optimizes the generic process by having Valkey manage it for you, and it's a simple solution for most recovery needs. 💅

XAUTOCLAIM identifies idle pending messages and transfers ownership of them to a consumer. The signature looks like this:

XAUTOCLAIM <key> <group> <consumer> <min-idle-time> <start> [COUNT count] [JUSTID]

So in the example above, I could've used automatic claiming to claim a single message like this:

127.0.0.1:6379> XAUTOCLAIM race:italy italy_riders Alice 60000 0-0 COUNT 1
1) "0-0"
2) 1) 1) "1692632662819-0"
      2) 1) "rider"
         2) "Sam-Bodden"

Same energy as XCLAIM — the reply is an array of the messages you just claimed — but XAUTOCLAIM also hands you back a stream ID that lets you keep iterating through the pending entries. That ID is a cursor bb, and you just plug it into your next call to keep scooping up idle pending messages:

127.0.0.1:6379> XAUTOCLAIM race:italy italy_riders Lora 60000 (1692632662819-0 COUNT 1
1) "1692632662819-0"
2) 1) 1) "1692632647899-0"
      2) 1) "rider"
         2) "Royce"

When XAUTOCLAIM gives you back the "0-0" stream ID as the cursor, that means it hit the end of the consumer group pending entries list. Important though: that does not mean there are no new idle pending messages, so the process keeps going by calling XAUTOCLAIM again from the beginning of the stream.

Claiming and the delivery counter 💀

The counter you see in the XPENDING output is the number of deliveries of each message. It gets incremented in two ways: when a message is successfully claimed via XCLAIM, or when an XREADGROUP call is used to access the history of pending messages.

When things fail, it's totally normal that messages get delivered multiple times — but eventually they usually get processed and acknowledged. Usually. Sometimes there's a message that just will not process, because it's corrupted or crafted in a way that triggers a bug in your processing code. And when that happens, what you get is consumers continuously failing on that one specific message. It's giving cursed.

But because we have the counter of delivery attempts, we can use that counter to detect messages that for some reason are not processable. So once the deliveries counter reaches some big number that you chose, it's probably wiser to put those messages in another stream and send a notification to the system administrator. That's basically how Streams implements the dead letter concept. 🫶

Streams observability 🔍✨

Messaging systems that lack observability are so hard to work with. Not knowing who's consuming messages, what messages are pending, which consumer groups are active in a given stream — it makes everything opaque, and we hate that. So Streams and consumer groups have a bunch of different ways to observe what's happening. We already covered XPENDING, which lets you inspect the list of messages that are under processing at a given moment, along with their idle time and number of deliveries.

But we may want to do more than that, and the XINFO command is the observability interface — it's used with sub-commands to get information about streams or consumer groups.

This command uses subcommands to show different info about the status of the stream and its consumer groups. For instance, XINFO STREAM reports information about the stream itself.

127.0.0.1:6379> XINFO STREAM race:italy
 1) "length"
 2) (integer) 5
 3) "radix-tree-keys"
 4) (integer) 1
 5) "radix-tree-nodes"
 6) (integer) 2
 7) "last-generated-id"
 8) "1692632678249-0"
 9) "groups"
10) (integer) 1
11) "first-entry"
12) 1) "1692632639151-0"
    2) 1) "rider"
       2) "Castilla"
13) "last-entry"
14) 1) "1692632678249-0"
    2) 1) "rider"
       2) "Norem"

The output shows info about how the stream is encoded internally, and also shows the first and last message in the stream. Another piece of info available is the number of consumer groups associated with this stream. And we can dig even further, asking for more info about the consumer groups.

127.0.0.1:6379> XINFO GROUPS race:italy
1) 1) "name"
   2) "italy_riders"
   3) "consumers"
   4) (integer) 3
   5) "pending"
   6) (integer) 2
   7) "last-delivered-id"
   8) "1692632662819-0"

As you can see in this and the previous output, XINFO outputs a sequence of field-value items. Because it's an observability command, that lets the human user immediately understand what information is being reported — and it lets the command report more information in the future by adding more fields without breaking compatibility with older clients. Obsessed with the future-proofing. Other commands that have to be more bandwidth efficient, like XPENDING, just report the information without the field names.

The output of the example above, where the GROUPS subcommand is used, should be clear just from looking at the field names. And we can check the state of a specific consumer group in more detail by checking the consumers that are registered in the group.

127.0.0.1:6379> XINFO CONSUMERS race:italy italy_riders
1) 1) "name"
   2) "Alice"
   3) "pending"
   4) (integer) 1
   5) "idle"
   6) (integer) 177546
2) 1) "name"
   2) "Bob"
   3) "pending"
   4) (integer) 0
   5) "idle"
   6) (integer) 424686
3) 1) "name"
   2) "Lora"
   3) "pending"
   4) (integer) 1
   5) "idle"
   6) (integer) 72241

And if you can't remember the syntax of the command, bestie, just ask the command itself for help:

> XINFO HELP
1) XINFO <subcommand> [<arg> [value] [opt] ...]. Subcommands are:
2) CONSUMERS <key> <groupname>
3)     Show consumers of <groupname>.
4) GROUPS <key>
5)     Show the stream consumer groups.
6) STREAM <key> [FULL [COUNT <count>]
7)     Show information about the stream.
8) HELP
9)     Prints this help.

Differences with Kafka (TM) partitions 🌈

Consumer groups in Streams may resemble Kafka (TM) partitioning-based consumer groups in some way, but lowkey note that Streams are, in practical terms, very different. The partitions are only logical and the messages are just put into a single Valkey key, so the way the different clients get served is based on who is ready to process new messages — not on which partition clients are reading from. For instance, if the consumer C3 at some point fails permanently, Valkey will continue to serve C1 and C2 all the new messages arriving, as if there are now only two logical partitions.

Similarly, if a given consumer is much faster at processing messages than the other consumers, that consumer will receive proportionally more messages in the same unit of time. This is possible since Valkey tracks all the unacknowledged messages explicitly, and remembers who received which message and the ID of the first message never delivered to any consumer.

But this also means that in Valkey, if you really want to partition messages in the same stream into multiple Valkey instances, you have to use multiple keys and some sharding system such as Valkey Cluster or some other application-specific sharding system. A single Stream is not automatically partitioned to multiple instances. No cap.

We could say that schematically the following is true:

If you use 1 stream -> 1 consumer, you are processing messages in order.
If you use N streams with N consumers, so that only a given consumer hits a subset of the N streams, you can scale the above model of 1 stream -> 1 consumer.
If you use 1 stream -> N consumers, you are load balancing to N consumers — however in that case, messages about the same logical item may be consumed out of order, because a given consumer may process message 3 faster than another consumer is processing message 4.

So basically Kafka partitions are more similar to using N different Valkey keys, while Valkey consumer groups are a server-side load balancing system of messages from a given stream to N different consumers. 💅

Capped Streams 🎀

Many applications do not want to collect data into a stream forever. Sometimes it's useful to have at maximum a given number of items inside a stream; other times, once a given size is reached, it's useful to move data from Valkey to a storage which is not in memory and not as fast, but suited to store the history for, potentially, decades to come. Streams have some support for this. One of them is the MAXLEN option of the XADD command. This option is very simple to use:

127.0.0.1:6379> XADD race:italy MAXLEN 2 * rider Jones
"1692633189161-0"
127.0.0.1:6379> XADD race:italy MAXLEN 2 * rider Wood
"1692633198206-0"
127.0.0.1:6379> XADD race:italy MAXLEN 2 * rider Henshaw
"1692633208557-0"
127.0.0.1:6379> XLEN race:italy
(integer) 2
127.0.0.1:6379> XRANGE race:italy - +
1) 1) "1692633198206-0"
   2) 1) "rider"
      2) "Wood"
2) 1) "1692633208557-0"
   2) 1) "rider"
      2) "Henshaw"

Using MAXLEN, the old entries are automatically evicted when the specified length is reached, so the stream is left at a constant size. There is currently no option to tell the stream to just retain items that are not older than a given period — because such a command, in order to run consistently, would potentially block for a long time in order to evict items. Imagine for example what happens if there is an insertion spike, then a long pause, and then another insertion, all with the same maximum time. The stream would block to evict the data that became too old during the pause. So it's up to you, the user, to do some planning and understand what the maximum stream length you want is. Moreover, while the length of the stream is proportional to the memory used, trimming by time is less simple to control and anticipate: it depends on the insertion rate, which often changes over time (and when it does not change, then to just trim by size is trivial).

However, trimming with MAXLEN can be expensive: streams are represented by macro nodes into a radix tree, in order to be very memory efficient. Altering the single macro node, consisting of a few tens of elements, is not optimal. So it's possible to use the command in the following special form:

XADD race:italy MAXLEN ~ 1000 * ... entry fields here ...

The ~ argument between the MAXLEN option and the actual count means I don't really need this to be exactly 1000 items. It can be 1000 or 1010 or 1030, just make sure to save at least 1000 items. With this argument, the trimming is performed only when we can remove a whole node. This makes it much more efficient, and it is usually what you want. You'll note here that the client libraries have various implementations of this. For example, the Python client defaults to approximate and has to be explicitly set to a true length. ✨ efficiency queen ✨

okay so there's also the XTRIM command, and it does pretty much exactly what the MAXLEN option does up above 👆 — except it can run all by itself, no XADD required:

127.0.0.1:6379> XTRIM race:italy MAXLEN 10
(integer) 0

Or, if you wanna do it the XADD-option way:

127.0.0.1:6379> XTRIM race:italy MAXLEN ~ 10
(integer) 0

But here's the thing bestie: XTRIM was designed to accept different trimming strategies, not just the one. Another one is MINID, which evicts entries whose IDs are lower than the one you specify. 🎀

And since XTRIM is an explicit command, the user is expected to know about the possible shortcomings of the different trimming strategies. No hand-holding, iykyk 🫶

Another useful eviction strategy that may be added to XTRIM in the future is removing by a range of IDs — which would make it easier to use XRANGE and XTRIM together to move data from Valkey into other storage systems if needed. ✨

🌈 Special IDs in the streams API 🌈

You may have noticed there are a bunch of special IDs floating around the Valkey API. Here's a short recap so they actually make sense later, fr fr.

The first two special IDs are - and +, and they get used in range queries with the XRANGE command. Those two mean, respectively, the smallest ID possible (basically 0-1) and the greatest ID possible (that's 18446744073709551615-18446744073709551615). Like, look at those numbers. Writing - and + instead is so much cleaner, no cap. 💅

Then there are APIs where we wanna say "the ID of the item with the greatest ID inside the stream". That's what $ means. So for instance, if I want only new entries with XREADGROUP, I use this ID to signify I already have all the existing entries, but not the new ones that will be inserted in the future. Same energy when I create or set the ID of a consumer group — I can set the last delivered item to $ so that only new entries get delivered to the consumers in the group. 🦋

And as you can see, $ does not mean +. Two different things!! + is the greatest ID possible in every possible stream, while $ is the greatest ID in a given stream containing given entries. Also, APIs will usually only understand + or $ — but it was useful to avoid loading one single symbol with multiple meanings. Vibes preserved. ✨

Another special ID is >, and it has a special meaning only related to consumer groups, and only when the XREADGROUP command is used. This one means we want only the entries that were never delivered to other consumers so far. So basically the > ID is the last delivered ID of a consumer group. 👑

Finally there's *, which can only be used with the XADD command, and it means: auto-select an ID for us for the new entry. ⭐

So yeah: we have -, +, $, >, and * — all with a different meaning, and most of the time they can be used in different contexts. 🌸

💾 Persistence, replication and message safety 💾

A Stream, like any other Valkey data structure, is asynchronously replicated to replicas and persisted into AOF and RDB files. But what may not be so obvious is that the consumer groups' full state also gets propagated to AOF, RDB, and replicas — so if a message is pending in the primary, the replica will have the same information too. Same deal after a restart: the AOF will restore the consumer groups' state. 🫶

However, note that Streams and consumer groups are persisted and replicated using the Valkey default replication, so:

AOF must be used with a strong fsync policy if persistence of messages is important in your application.
By default the asynchronous replication will not guarantee that XADD commands or consumer groups state changes are replicated: after a failover something can be missing, depending on the ability of replicas to receive the data from the primary.
The WAIT command may be used in order to force the propagation of the changes to a set of replicas. However note that while this makes it very unlikely that data is lost, the Valkey failover process as operated by Sentinel or Valkey Cluster performs only a best effort check to failover to the replica which is the most updated, and under certain specific failure conditions may promote a replica that lacks some data. 💀

So when you're designing an application using Streams and consumer groups, make sure to understand the semantical properties your application should have during failures, and configure things accordingly — evaluating whether it's safe enough for your use case. That part is on you bestie 😭

🗑️ Removing single items from a stream 🗑️

Streams also have a special command for removing items from the middle of a stream, just by ID. Normally, for an append-only data structure, this might look like an odd feature — but it's actually useful for applications involving, for instance, privacy regulations. 👀 The command is called XDEL and it receives the name of the stream followed by the IDs to delete:

127.0.0.1:6379> XRANGE race:italy - + COUNT 2
1) 1) "1692633198206-0"
   2) 1) "rider"
      2) "Wood"
2) 1) "1692633208557-0"
   2) 1) "rider"
      2) "Henshaw"
127.0.0.1:6379> XDEL race:italy 1692633208557-0
(integer) 1
127.0.0.1:6379> XRANGE race:italy - + COUNT 2
1) 1) "1692633198206-0"
   2) 1) "rider"
      2) "Wood"

However, in the current implementation, memory is not really reclaimed until a macro node is completely empty — so don't abuse this feature, okay? 🙈

🫧 Zero length streams 🫧

One difference between streams and other Valkey data structures: when the other data structures no longer have any elements, as a side effect of calling commands that remove elements, the key itself gets removed. So for instance, a sorted set will be completely removed when a call to ZREM removes the last element in it. Streams, on the other hand, are allowed to stay at zero elements — whether that's because you used a MAXLEN option with a count of zero (that's XADD and XTRIM commands), or because XDEL was called.

Why does this asymmetry exist? Because Streams may have associated consumer groups, and we do not want to lose the state those consumer groups defined just because there are no longer any items in the stream. Currently the stream is not deleted even when it has no associated consumer groups. Periodt. 💅

⏱️ Total latency of consuming a message ⏱️

Non-blocking stream commands like XRANGE and XREAD, or XREADGROUP without the BLOCK option, are served synchronously just like any other Valkey command — so discussing the latency of such commands is meaningless. It's way more interesting to check the time complexity of the commands in the Valkey documentation. It should be enough to say that stream commands are at least as fast as sorted set commands when extracting ranges, and that XADD is very fast — it can easily insert from half a million to one million items per second on an average machine if pipelining is used. 🔥

But latency becomes an interesting parameter if we want to understand the delay of processing a message, in the context of blocking consumers in a consumer group — from the moment the message is produced via XADD, to the moment the message is obtained by the consumer because XREADGROUP returned with the message. ✨

🧚 How serving blocked consumers works 🧚

Before handing over the results of the tests that were performed, it's interesting to understand what model Valkey uses in order to route stream messages (and, in general, actually, how any blocking operation waiting for data is managed).

The blocked client is referenced in a hash table that maps keys for which there is at least one blocking consumer, to a list of consumers that are waiting for such key. This way, given a key that received data, we can resolve all the clients that are waiting for such data.
When a write happens — in this case, when the XADD command is called — it calls the signalKeyAsReady() function. This function will put the key into a list of keys that need to be processed, because such keys may have new data for blocked consumers. Note that such ready keys will be processed later, so in the course of the same event loop cycle, it's possible that the key will receive other writes.
Finally, before returning into the event loop, the ready keys are processed. For each key, the list of clients waiting for data is scanned, and if applicable, such clients will receive the new data that arrived. In the case of streams, the data is the messages in the applicable range requested by the consumer. 🎀

As you can see, basically, before returning to the event loop both the client calling XADD and the clients blocked to consume messages will have their reply in the output buffers — so the caller of XADD should receive the reply from Valkey at about the same time the consumers will receive the new messages. 🌸

This model is push-based, since adding data to the consumers' buffers will be performed directly by the action of calling XADD, so the latency tends to be quite predictable. 👑

📊 Latency tests results 📊

In order to check these latency characteristics, a test was performed using multiple instances of Ruby programs pushing messages having as an additional field the computer millisecond time, and Ruby programs reading the messages from the consumer group and processing them. The message processing step consisted of comparing the current computer time with the message timestamp, in order to understand the total latency. 🧪

Results obtained:

Processed between 0 and 1 ms -> 74.11%
Processed between 1 and 2 ms -> 25.80%
Processed between 2 and 3 ms -> 0.06%
Processed between 3 and 4 ms -> 0.01%
Processed between 4 and 5 ms -> 0.02%

So 99.9% of requests have a latency <= 2 milliseconds, with the outliers that remain still very close to the average. 🌈✨

Adding a few million unacknowledged messages to the stream does not change the gist of the benchmark, with most queries still processed with very short latency. 💅

A few remarks:

Here we processed up to 10k messages per iteration, this means that the COUNT parameter of XREADGROUP was set to 10000. This adds a lot of latency, but is needed in order to allow the slow consumers to be able to keep with the message flow. So you can expect a real-world latency that is a lot smaller. 🫶
The system used for this benchmark is very slow compared to today's standards. 😭

and that's a wrap bestie 🎀 this whole page is from Valkey 9.0.0, dated 2025-10-21, and it lives at VALKEY-STREAMS-INTRO(7) — go stream something iconic 💖✨🦋

📦 where this page comes from

fedora 44: valkey-doc 9.0.3-2.fc44 · CC-BY-SA-4.0
fedora 44: valkey-doc 9.0.6-1.fc44 · CC-BY-SA-4.0
💖 girlified by deepseek-ai/DeepSeek-V4.1-Flash on 2026-10-18: a language model's rewrite of the original, under its terms