Redis Stream β a new abstract data type introduced in Redis with the release of version 5.0
Conceptually, Redis Stream is a List to which you can add entries. Each entry has a unique identifier. By default, the identifier is generated automatically and includes a timestamp. Therefore, you can query ranges of entries by time or receive new data as it arrives in the stream, similar to how the Unix command 'tail -f' reads a log file and waits for new data. Note that multiple clients can listen to the stream simultaneously, just as many 'tail -f' processes can read a file at the same time without conflicting.
To understand all the advantages of the new data type, let's briefly recall the long-standing Redis structures that partially replicate the functionality of Redis Stream.
Redis PUB/SUB
Redis Pub/Sub β a simple messaging system already built into your key-value storage. However, simplicity comes at a cost:
- If a publisher crashes for any reason, it loses all its subscribers
- A publisher needs to know the exact address of all its subscribers
- A publisher can overwhelm its subscribers with work if data is published faster than it can be processed
- A message is removed from the publisher's buffer immediately after publication, regardless of how many subscribers it was delivered to and how quickly they managed to process it.
- All subscribers will receive the message simultaneously. Subscribers must coordinate among themselves how to handle the same message.
- There is no built-in mechanism for confirming successful processing of a message by a subscriber. If a subscriber receives the message and crashes during processing, the publisher will not know.
Redis List
Redis List is a data structure that supports blocking read commands. You can add and read messages from the beginning or end of the list. Based on this structure, you can create a decent stack or queue for your distributed system, which will often be sufficient. The main differences from Redis Pub/Sub are:
- A message is delivered to a single client. The first client that is blocked on reading will receive the data first.
- Clint must initiate the read operation for each message himself. List knows nothing about clients.
- Messages are stored until they are read or explicitly deleted. If you have configured the Redis server to flush data to disk, the reliability of the system increases significantly.
Introduction to Stream
Adding a record to the stream
The command XADD adds a new entry to the stream. An entry is not just a string; it consists of one or more key-value pairs. Thus, each entry is already structured and resembles the structure of a CSV file.
> XADD mystream * sensor-id 1234 temperature 19.8
1518951480106-0
In the example above, we are adding two fields to the stream named (key) 'mystream': 'sensor-id' and 'temperature' with values '1234' and '19.8', respectively. As the second argument, the command takes an identifier that will be assigned to the entry β this identifier uniquely identifies each entry in the stream. However, in this case, we passed *, because we want Redis to generate a new identifier for us. Each new identifier will increment. Therefore, every new entry will have a larger identifier compared to previous entries.
Identifier format
The identifier of the entry returned by the command XADD, consists of two parts:
{millisecondsTime}-{sequenceNumber}
millisecondsTime β Unix time in milliseconds (time server Redis). However, if the current time turns out to be the same as or less than the time of the previous entry, the timestamp of the previous entry is used. Thus, if the server time goes back in time, the new identifier will still retain its incrementing property.
sequenceNumber is used for entries created in the same millisecond. sequenceNumber will be incremented by 1 relative to the previous entry. Since sequenceNumber is 64 bits in size, in practice you should not run into a limit on the number of entries that can be generated within a single millisecond.
At first glance, the format of such identifiers may seem strange. A skeptical reader might wonder why time is part of the identifier. The reason is that Redis streams support range queries by identifiers. Since the identifier is related to the creation time of the entry, this allows for querying time ranges. We will look at a specific example when we move on to studying the command. XRANGE.
If for any reason a user needs to specify their own identifier, which for example is associated with some external system, we can pass it to the command XADD instead of the asterisk as shown below:
> XADD somestream 0-1 field value
0-1
> XADD somestream 0-2 foo bar
0-2
Note that in this case you must keep track of the increment of the identifier yourself. In our example, the minimum identifier is "0-1", so the command will not accept another identifier that is equal to or less than "0-1".
> XADD somestream 0-1 foo bar
(error) ERR The ID specified in XADD is equal or smaller than the target stream top item
The number of entries in the stream
You can get the number of entries in the stream simply by using the command XLEN. For our example, this command will return the following value:
> XLEN somestream
(integer) 2
Range queries β XRANGE and XREVRANGE
To query data by range, we need to specify two identifiers β the start and end of the range. The returned range will include all elements, including the boundaries. There are also two special identifiers "-" and "+", which correspond to the smallest (first entry) and largest (last entry) identifier in the stream, respectively. The example below will output all entries in the stream.
> XRANGE mystream - +
1) 1) 1518951480106-0
2) 1) "sensor-id"
2) "1234"
3) "temperature"
4) "19.8"
2) 1) 1518951482479-0
2) 1) "sensor-id"
2) "9999"
3) "temperature"
4) "18.2"
Each returned entry is an array of two elements: the identifier and a list of key-value pairs. We have already mentioned that the identifiers of entries are time-related. Therefore, we can query a range of a specific time interval. However, we can specify in the request not a full identifier, but only Unix time, omitting the part related to sequenceNumberThe omitted part of the identifier will automatically be set to zero at the beginning of the range and to the maximum possible value at the end of the range. Below is an example of how to request a range of two milliseconds.
> XRANGE mystream 1518951480106 1518951480107
1) 1) 1518951480106-0
2) 1) "sensor-id"
2) "1234"
3) "temperature"
4) "19.8"
We have only one entry in this range, but in real datasets, the returned result can be massive. For this reason, XRANGE the COUNT option is supported. By specifying a number, we can simply retrieve the first N entries. If we need the next N entries (pagination), we can use the last retrieved identifier, increment it sequenceNumber by one, and request again. Let's look at this in the following example. We start adding 10 items using XADD (let's assume the mystream has already been filled with 10 items). To begin iteration, retrieving 2 items per command, we start with the full range but with COUNT set to 2.
> XRANGE mystream - + COUNT 2
1) 1) 1519073278252-0
2) 1) "foo"
2) "value_1"
2) 1) 1519073279157-0
2) 1) "foo"
2) "value_2"
To continue the iteration with the next two items, we need to take the last retrieved identifier, which is 1519073279157-0, and add 1 to it. sequenceNumber.
The resulting identifier, in this case, 1519073279157-1, can now be used as the new starting range argument for the next call. XRANGE:
> XRANGE mystream 1519073279157-1 + COUNT 2
1) 1) 1519073280281-0
2) 1) "foo"
2) "value_3"
2) 1) 1519073281432-0
2) 1) "foo"
2) "value_4"
And so on. Since the complexity XRANGE is O(log (N)) for the search and then O(M) for returning M elements, each iteration step is quick. Thus, using XRANGE we can efficiently iterate through streams.
The command XREVRANGE is the equivalent XRANGE, but returns elements in reverse order:
> XREVRANGE mystream + - COUNT 1
1) 1) 1519073287312-0
2) 1) "foo"
2) "value_10"
Note that the command XREVRANGE takes the range arguments start and stop in reverse order.
Reading new entries using XREAD
There is often a task to subscribe to a stream and receive only new messages. This concept may seem similar to Redis Pub/Sub or blocking Redis List, but there are fundamental differences in how to use Redis Stream:
- By default, each new message is delivered to every subscriber. This behavior is different from a blocking Redis List, where a new message will be read by only one subscriber.
- While in Redis Pub/Sub all messages are forgotten and never saved, in Streams all messages are preserved indefinitely (unless the client explicitly calls for deletion).
- Redis Streams allows you to control access to messages within a single stream. A specific subscriber can see only their personal message history.
You can subscribe to a stream and receive new messages using the command XREAD. This is somewhat more complex than XRANGE, so we will first start with simpler examples.
> XREAD COUNT 2 STREAMS mystream 0
1) 1) "mystream"
2) 1) 1) 1519073278252-0
2) 1) "foo"
2) "value_1"
2) 1) 1519073279157-0
2) 1) "foo"
2) "value_2"
In the example above, a non-blocking form is specified. XREADNote that the COUNT option is not mandatory. In fact, the only required option for the command is the STREAMS option, which specifies the list of streams along with the corresponding maximum identifier. We wrote "STREAMS mystream 0" β we want to receive all records from the mystream with an ID greater than "0-0". As seen in the example, the command returns the stream name since we can subscribe to multiple streams simultaneously. We could write, for instance, "STREAMS mystream otherstream 0 0". Note that after the STREAMS option, we first need to provide the names of all the required streams and then the list of identifiers.
In this simple form, the command does nothing particularly special compared to XRANGE. However, the interesting part is that we can easily turn XREAD into a blocking command by specifying the BLOCK argument:
> XREAD BLOCK 0 STREAMS mystream $
In the example above, a new BLOCK option is indicated with a timeout of 0 milliseconds (which means infinite waiting). Moreover, instead of passing a regular identifier for the mystream, a special identifier $ was provided. This special identifier means that XREAD should use the maximum identifier in the mystream. So we will receive only new messages starting from the moment we began listening. In a sense, this is similar to the Unix command "tail -f".
Note that when using the BLOCK option, we don't necessarily need a special identifier $. We can use any existing identifier in the stream. If the command can handle our request immediately, without blocking, it will do so; otherwise, it will block.
Blocking XREAD can also listen to multiple streams at once; you just need to specify their names. In this case, the command will return the record from the first stream that received the data. The first subscriber blocked for that stream will receive the data first.
Consumer Groups
In certain tasks, we want to restrict subscriber access to messages within a single stream. An example of when this might be useful is a message queue with workers that will receive different stream messages, allowing for message processing to be scaled.
If we imagine having three subscribers C1, C2, C3 and a stream containing messages 1, 2, 3, 4, 5, 6, 7, the message servicing will occur as shown in the diagram below:
1 -> C1
2 -> C2
3 -> C3
4 -> C1
5 -> C2
6 -> C3
7 -> C1
To achieve this effect, Redis Stream uses a concept called Consumer Group. This concept is similar to a pseudo-subscriber that receives data from the stream but is actually serviced by multiple subscribers within the group, providing certain guarantees:
- Each message is delivered to different subscribers within the group.
- Within the group, subscribers are identified by a name that is a case-sensitive string. If a subscriber temporarily leaves the group, they can rejoin by their unique name.
- Each Consumer Group follows the concept of 'first unacknowledged message.' When a subscriber requests new messages, they can only receive messages that have never been delivered to any subscriber within the group before.
- There is a command for explicitly acknowledging the successful processing of a message by a subscriber. Until this command is called, the requested message remains in a 'pending' status.
- Within a Consumer Group, each subscriber can request the history of messages that have specifically been delivered to them but have not yet been processed (in a 'pending' status).
In a sense, the state of the group can be represented as follows:
+----------------------------------------+
| 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)
+----------------------------------------+
Now it's time to get acquainted with the main commands for Consumer Group, namely:
- XGROUP is used to create, destroy, and manage groups
- XREADGROUP is used to read from the stream via the group
- XACK β this command allows the subscriber to mark a message as successfully processed
Creating a Consumer Group
Assuming the stream mystream already exists. Then the command to create the group will look like this:
> XGROUP CREATE mystream mygroup $
OK
When creating a group, we need to provide the identifier from which the group will start receiving messages. If we want to receive only new messages, we can use the special identifier $ (as in our example above). If instead of the special identifier we specify 0, then all messages from the stream will be available to the group.
Now, when the group is created, we can immediately start reading messages using the command XREADGROUP. This command is very similar to XREAD and supports an optional BLOCK option. However, there is a mandatory GROUP option that must always be specified with two arguments: the group name and the subscriber name. The COUNT option is also supported.
Before reading the stream, let's add some messages to it:
> XADD mystream * message apple
1526569495631-0
> XADD mystream * message orange
1526569498055-0
> XADD mystream * message strawberry
1526569506935-0
> XADD mystream * message apricot
1526569535168-0
> XADD mystream * message banana
1526569544280-0
And now let's try to read this stream through the group:
> XREADGROUP GROUP mygroup Alice COUNT 1 STREAMS mystream >
1) 1) "mystream"
2) 1) 1) 1526569495631-0
2) 1) "message"
2) "apple"
The command above literally states the following:
"I, Alice the subscriber, member of the group mygroup, want to read one message from the stream mystream that has never been delivered to anyone before."
Every time a subscriber performs an operation with a group, they must specify their name, uniquely identifying themselves within the group. There is another very important detail in the command above β the special identifier ">>". This special identifier filters messages, leaving only those that have not yet been delivered.
In special cases, you may also specify a real identifier, such as 0 or any other valid identifier. In this case, the command XREADGROUP will return the history of messages with a status of "pending" that have been delivered to the specified subscriber (Alice) but have not yet been acknowledged using the command XACK.
We can verify this behavior by immediately specifying the identifier 0, without the option COUNT. We will simply see the single pending message, which is the message with the apple:
> XREADGROUP GROUP mygroup Alice STREAMS mystream 0
1) 1) "mystream"
2) 1) 1) 1526569495631-0
2) 1) "message"
2) "apple"
However, if we acknowledge the message as successfully processed, it will no longer be displayed:
> XACK mystream mygroup 1526569495631-0
(integer) 1
> XREADGROUP GROUP mygroup Alice STREAMS mystream 0
1) 1) "mystream"
2) (empty list or set)
Now it's Bob's turn to read something:
> XREADGROUP GROUP mygroup Bob COUNT 2 STREAMS mystream >
1) 1) "mystream"
2) 1) 1) 1526569498055-0
2) 1) "message"
2) "orange"
2) 1) 1526569506935-0
2) 1) "message"
2) "strawberry"
Bob, a member of the mygroup group, requested no more than two messages. The command only reports undelivered messages due to the special identifier ">>". As you can see, the message "apple" does not appear since it has already been delivered to Alice; therefore, Bob receives "orange" and "strawberry."
Thus, Alice, Bob, and any other subscriber of the group can read different messages from the same stream. They can also read their history of unprocessed messages or acknowledge messages as processed.
There are a few things to keep in mind:
- Once a subscriber reads a message with the command XREADGROUP, that message transitions to a "pending" state and is tied to that specific subscriber. Other group subscribers will not be able to read this message.
- Subscribers are automatically created upon the first mention, without the need for explicit creation.
- Using XREADGROUP You can read messages from multiple streams simultaneously; however, for this to work, you need to create groups with the same name for each stream using XGROUP
Recovery from Failure
A subscriber can recover from a failure and read their list of messages with a status of 'pending'. However, in the real world, subscribers may ultimately fail. What happens to the pending messages of the subscriber if they cannot recover from the failure?
The Consumer Group offers a feature that is specifically used for such cases β when it is necessary to change the ownership of messages.
First, you need to call the command XPENDING, which displays all messages in the group with a status of 'pending'. In its simplest form, the command is called with only two arguments: the stream name and the group name:
> XPENDING mystream mygroup
1) (integer) 2
2) 1526569498055-0
3) 1526569506935-0
4) 1) 1) "Bob"
2) "2"
The command outputted the number of unprocessed messages for the entire group and for each subscriber. We only have Bob with two unprocessed messages because the single message requested by Alice was acknowledged with XACK.
We can request additional information using more arguments:
XPENDING {key} {groupname} [{start-id} {end-id} {count} [{consumer-name}]]
{start-id} {end-id} β range of IDs (you can use '-' and '+')
{count} β number of delivery attempts
{consumer-name} β group name
> XPENDING mystream mygroup - + 10
1) 1) 1526569498055-0
2) "Bob"
3) (integer) 74170458
4) (integer) 1
2) 1) 1526569506935-0
2) "Bob"
3) (integer) 74170458
4) (integer) 1
Now we have details for each message: ID, subscriber name, idle time in milliseconds, and finally, the number of delivery attempts. We have two messages from Bob, and they have been idle for 74170458 milliseconds, about 20 hours.
Note that nothing stops us from checking what the message content was by simply using XRANGE.
> XRANGE mystream 1526569498055-0 1526569498055-0
1) 1) 1526569498055-0
2) 1) "message"
2) "orange"
We just need to repeat the same ID twice in the arguments. Now that we have some idea, Alice may decide that after 20 hours of idleness, Bob is unlikely to recover, and it's time to request these messages and resume processing instead of Bob. For this, we use the command XCLAIM:
XCLAIM {key} {group} {consumer} {min-idle-time} {ID-1} {ID-2} ... {ID-N}
With this command, we can retrieve a 'foreign' message that has not yet been processed by changing the owner to {consumer}. However, we can also provide a minimum idle time of {min-idle-time}. This helps to avoid a situation where two clients try to simultaneously change the owner of the same messages.
Client 1: XCLAIM mystream mygroup Alice 3600000 1526569498055-0
Client 2: XCLAIM mystream mygroup Lora 3600000 1526569498055-0
The first client will reset the idle time and increase the delivery counter. Thus, the second client will not be able to request it.
> XCLAIM mystream mygroup Alice 3600000 1526569498055-0
1) 1) 1526569498055-0
2) 1) "message"
2) "orange"
The message has been successfully claimed by Alice, who can now process the message and acknowledge it.
From the example above, it is clear that a successful request returns the content of the message itself. However, this is not mandatory. The JUSTID option can be used to return only message identifiers. This is useful if you are not interested in the message details and want to improve system performance.
Delivery counter
The counter you observe in the output XPENDING β is the number of deliveries for each message. This counter increments in two ways: when a message is successfully claimed via XCLAIM or when a call is made to XREADGROUP.
It is normal for some messages to be delivered multiple times. The main thing is that all messages are eventually processed. Sometimes, issues may arise when processing a message due to corruption of the message itself, or the processing of the message may cause an error in the handler's code. In such cases, it may turn out that no one will be able to process this message. Since we have a delivery attempt counter, we can use this counter to detect such situations. Therefore, as soon as the delivery counter reaches a large number that you specified, it would probably be wiser to place such a message in a different stream and send a notification to the system administrator.
Stream status
The command XINFO is used to request various information about the stream and its groups. For example, the basic syntax of the command looks as follows:
> XINFO STREAM mystream
1) length
2) (integer) 13
3) radix-tree-keys
4) (integer) 1
5) radix-tree-nodes
6) (integer) 2
7) groups
8) (integer) 2
9) first-entry
10) 1) 1524494395530-0
2) 1) "a"
2) "1"
3) "b"
4) "2"
11) last-entry
12) 1) 1526569544280-0
2) 1) "message"
2) "banana"
The command above displays general information about the specified stream. Now for a slightly more complex example:
> XINFO GROUPS mystream
1) 1) name
2) "mygroup"
3) consumers
4) (integer) 2
5) pending
6) (integer) 2
2) 1) name
2) "some-other-group"
3) consumers
4) (integer) 1
5) pending
6) (integer) 0
The command above displays general information about all groups of the specified stream.
> XINFO CONSUMERS mystream mygroup
1) 1) name
2) "Alice"
3) pending
4) (integer) 1
5) idle
6) (integer) 9104628
2) 1) name
2) "Bob"
3) pending
4) (integer) 1
5) idle
6) (integer) 83841983
The command above displays information about all subscribers of the specified stream and group.
If you forget the command syntax, just refer to the command help:
> XINFO HELP
1) XINFO {subcommand} arg arg ... arg. Subcommands are:
2) CONSUMERS {key} {groupname} -- Show consumer groups of group {groupname}.
3) GROUPS {key} -- Show the stream consumer groups.
4) STREAM {key} -- Show information about the stream.
5) HELP -- Print this help.
Stream size limitation
Many applications do not want to collect data in a stream forever. It is often useful to have a maximum number of messages in the stream. In other cases, it is beneficial to transfer all messages from the stream to another persistent storage once the specified stream size is reached. You can limit the stream size using the MAXLEN parameter in the command. XADD:
> XADD mystream MAXLEN 2 * value 1
1526654998691-0
> XADD mystream MAXLEN 2 * value 2
1526654999635-0
> XADD mystream MAXLEN 2 * value 3
1526655000369-0
> XLEN mystream
(integer) 2
> XRANGE mystream - +
1) 1) 1526654999635-0
2) 1) "value"
2) "2"
2) 1) 1526655000369-0
2) 1) "value"
2) "3"
When using MAXLEN, old entries are automatically removed once the specified length is reached, keeping the stream at a constant size. However, trimming in this case is not the most efficient way in Redis memory. The situation can be improved as follows:
XADD mystream MAXLEN ~ 1000 * ... entry fields here ...
The argument ~ in the above example means that we do not have to limit the length of the stream to a specific value. In our example, this could be any number greater than or equal to 1000 (for example, 1000, 1010, or 1030). We have explicitly indicated that we want our stream to store at least 1000 entries. This makes memory management much more efficient within Redis.
There is also a separate command XTRIM, which performs the same function:
> XTRIM mystream MAXLEN 10
> XTRIM mystream MAXLEN ~ 10
Persistent storage and replication
Redis Stream asynchronously replicates to slave nodes and is saved in AOF files (a snapshot of all data) and RDB files (a log of all write operations). Consumer Groups state replication is also supported. Thus, if a message is in "pending" status on the master node, it will have the same status on the slave nodes.
Removing individual elements from the stream
There is a special command to delete messages XDEL. The command takes the stream name followed by the message identifiers that need to be deleted:
> XRANGE mystream - + COUNT 2
1) 1) 1526654999635-0
2) 1) "value"
2) "2"
2) 1) 1526655000369-0
2) 1) "value"
2) "3"
> XDEL mystream 1526654999635-0
(integer) 1
> XRANGE mystream - + COUNT 2
1) 1) 1526655000369-0
2) 1) "value"
2) "3"
When using this command, it should be noted that memory will not be freed immediately.
Zero-length streams
The difference between streams and other Redis data structures is that when other data structures no longer have elements, as a side effect, the data structure itself is removed from memory. For example, a sorted set will be completely deleted when calling ZREM removes the last element. Instead, streams are allowed to remain in memory even when they contain no elements.
Conclusion
Redis Stream is perfect for creating message brokers, message queues, unified logs, and chat systems that store history.
As Niklaus Wirth once said One Microsoft developer believes that ReactOS could not have evolved without borrowing code from Windows
Source: habr.com
