XREADGROUP(3valkey)

📦
version 9.0.6 · 2 versions
, in fedora 44

💖 XREADGROUP 💖

okay bestie, welcome to the Valkey Command Manual page for xreadgroup, from the 3valkey manual 💅 she's the stream-reader-with-consumer-groups girlie and she's kind of a lot, but in a good way.

✨ NAME ✨

xreadgroup — returns new or historical messages from a stream for a consumer in a group. Blocks until a message is available otherwise. 🦋

🌸 SYNOPSIS 🌸

XREADGROUP GROUP group consumer [COUNT count] [BLOCK milliseconds]
[NOACK] STREAMS key [key...]  ID [ID...]

🎀 DESCRIPTION 🎀

okay so XREADGROUP is the ✨special✨ version of XREAD — literally same energy, except it's got consumer group support. ngl bestie you're probably gonna have to understand XREAD first before this page makes any sense at all. no cap.

Also? if you're new to streams, we highly recommend reading our valkey-streams-intro(7) introduction to Streams 🌈. Please make sure you actually understand the concept of a consumer group in that intro, so following how this command works is way simpler for you. it's giving homework but it's giving necessary homework.

🧚 Consumer groups in 30 seconds 🧚

The difference between this command and the vanilla XREAD is that this one supports consumer groups. periodt.

Without consumer groups, just using XREAD, all the clients are served with all the entries arriving in a stream. But with consumer groups + XREADGROUP, you can create groups of clients that consume different parts of the messages arriving in a given stream 👑. Like, if the stream gets the new entries A, B, and C, and there are two consumers reading via a consumer group, one client will get (for instance) messages A and C, and the other gets message B, and so forth. slay, division of labor.

Within a consumer group, a given consumer (that's just a client consuming messages from the stream) has to identify itself with a unique consumer name. Which is just a string. fr fr, that's it.

One of the guarantees of consumer groups is that a given consumer can only see the history of messages that were delivered to it, so a message has just a single owner. BUT — and this is the iconic part — there's a special feature called message claiming that lets other consumers claim messages in case there's a non recoverable failure of some consumer 💀. To make that whole semantic work, consumer groups require explicit acknowledgment of the messages successfully processed by the consumer, via the XACK command. This is needed because the stream will track, for each consumer group, who is processing what message.

So here's how to vibe check whether you want a consumer group or not:

  1. If you have a stream and multiple clients, and you want all the clients to get all the messages — you do not need a consumer group.
  2. If you have a stream and multiple clients, and you want the stream to be partitioned or sharded across your clients, so that each client gets a sub set of the messages arriving in a stream — you do need a consumer group.

🦋 Differences between XREAD and XREADGROUP 🦋

From the point of view of the syntax, the commands are almost the same — however XREADGROUP requires a special and mandatory option:

GROUP <group-name> <consumer-name>

The group name is just the name of a consumer group associated to the stream. The group is created using the XGROUP command. The consumer name is the string the client uses to identify itself inside the group. The consumer is auto created inside the consumer group the first time it is saw 👀. Different clients should select a different consumer name.

When you read with XREADGROUP, the server will remember that a given message was delivered to you: the message gets stored inside the consumer group in what's called a Pending Entries List (PEL) — that's a list of message IDs delivered but not yet acknowledged. iykyk.

The client will have to acknowledge the message processing using XACK in order for the pending entry to be removed from the PEL. The PEL can be inspected using the XPENDING command 👑.

The NOACK subcommand can be used to avoid adding the message to the PEL in cases where reliability is not a requirement and the occasional message loss is acceptable 😭. This is equivalent to acknowledging the message when it is read.

The ID to specify in the STREAMS option when using XREADGROUP can be one of the following two:

  • The special > ID, which means the consumer wants to receive only messages that were never delivered to any other consumer. It just means, give me new messages ✨.
  • Any other ID — that is, 0 or any other valid ID or incomplete ID (just the millisecond time part) — will have the effect of returning entries that are pending for the consumer sending the command, with IDs greater than the one provided. So basically if the ID is not >, then the command will just let the client access its pending entries: messages delivered to it, but not yet acknowledged. Note that in this case, both BLOCK and NOACK are ignored.

Like XREAD, the XREADGROUP command can be used in a blocking way. There are no differences in this regard.

💜 What happens when a message is delivered to a consumer? 💜

Two things:

  1. If the message was never delivered to anyone — that is, if we're talking about a new message — then a PEL (Pending Entries List) is created.
  2. If instead the message was already delivered to this consumer, and it's just re-fetching the same message again, then the last delivery counter is updated to the current time, and the number of deliveries is incremented by one. You can access those message properties using the XPENDING command.

🌈 Usage example 🌈

Normally you use the command like that in order to get new messages and process them. In pseudo-code:

WHILE true
    entries = XREADGROUP GROUP $GroupName $ConsumerName BLOCK 2000 COUNT 10 STREAMS mystream >
    if entries == nil
        puts "Timeout... try again"
        CONTINUE
    end

    FOREACH entries AS stream_entries
        FOREACH stream_entries as message
            process_message(message.id,message.fields)

            # ACK the message as processed
            XACK mystream $GroupName message.id
        END
    END
END

In this way the example consumer code will fetch only new messages, process them, and acknowledge them via XACK ✨. However the example code above is not complete, because it does not handle recovering after a crash 💀. What happens if we crash in the middle of processing messages is that our messages will remain in the pending entries list — so we can access our history by giving XREADGROUP initially an ID of 0, and performing the same loop. Once providing an ID of 0, the reply is an empty set of messages, we know that we processed and acknowledged all the pending messages: we can start to use > as ID, in order to get the new messages and rejoin the consumers that are processing new things. main character energy, truly.

To see how the command actually replies, please check the XREAD command page 🫶.

🍓 What happens when a pending message is deleted? 🍓

Entries may be deleted from the stream due to trimming or explicit calls to XDEL at any time. By design, Valkey doesn't prevent the deletion of entries that are present in the stream's PELs. When this happens, the PELs retain the deleted entries' IDs, but the actual entry payload is no longer available. Therefore, when reading such PEL entries, Valkey will return a null value in place of their respective data.

Example:

> XADD mystream 1 myfield mydata
"1-0"
> XGROUP CREATE mystream mygroup 0
OK
> XREADGROUP GROUP mygroup myconsumer STREAMS mystream >
1) 1) "mystream"
   2) 1) 1) "1-0"
         2) 1) "myfield"
            2) "mydata"
> XDEL mystream 1-0
(integer) 1
> XREADGROUP GROUP mygroup myconsumer STREAMS mystream 0
1) 1) "mystream"
   2) 1) 1) "1-0"
         2) (nil)

Reading the valkey-streams-intro(7) Streams introduction is highly suggested in order to understand more about the streams overall behavior and semantics 🌸.

💌 REPLY 💌

⭐ RESP2 ⭐

One of the following:

  • valkey-protocol(7) Array reply: an array where each element is an array composed of a two elements containing the key name and the entries reported for that key. The entries reported are full stream entries, having IDs and the list of all the fields and values. Field and values are guaranteed to be reported in the same order they were added by XADD.
  • valkey-protocol(7) Nil reply: if the BLOCK option is given and a timeout occurs, or if there is no stream that can be served.

⭐ RESP3 ⭐

One of the following:

  • valkey-protocol(7) Map reply: A map of key-value elements where each element is composed of the key name and the entries reported for that key. The entries reported are full stream entries, having IDs and the list of all the fields and values. Field and values are guaranteed to be reported in the same order they were added by XADD.
  • valkey-protocol(7) Null reply: if the BLOCK option is given and a timeout occurs, or if there is no stream that can be served.

👑 COMPLEXITY 👑

For each stream mentioned: O(M) with M being the number of elements returned. If M is constant (e.g. always asking for the first 10 elements with COUNT), you can consider it O(1). On the other side when XREADGROUP blocks, XADD will pay the O(N) time in order to serve the N clients blocked on the stream getting new data. math girlie behavior, honestly.

🔥 ACL CATEGORIES 🔥

@blocking @slow @stream @write

🦋 HISTORY 🦋

  • Available since: 5.0.0

💖 SEE ALSO 💖

xack(3valkey), xadd(3valkey), xautoclaim(3valkey), xclaim(3valkey), xdel(3valkey), xgroup(3valkey), xgroup-create(3valkey), xgroup-createconsumer(3valkey), xgroup-delconsumer(3valkey), xgroup-destroy(3valkey), xgroup-help(3valkey), xgroup-setid(3valkey), xinfo(3valkey), xinfo-consumers(3valkey), xinfo-groups(3valkey), xinfo-help(3valkey), xinfo-stream(3valkey), xlen(3valkey), xpending(3valkey), xrange(3valkey), xread(3valkey), xrevrange(3valkey), xsetid(3valkey), xtrim(3valkey)

and that's the whole page bestie 🫶 this was Valkey 9.0.0, dated 2025-10-21 — go forth and consume your streams responsibly, mwah 💋✨

📦 where this page comes from