实现消费者组逻辑

In the previous lesson, we learned what Redis Streams are. Now, let's dive into Consumer Groups! They are a powerful feature that lets multiple clients process messages from a stream cooperatively.

Imagine a team of workers processing tasks from a single to-do list. Each worker gets unique tasks, and if one fails, others can pick up pending tasks. That's what Consumer Groups enable!

Before consumers can join, you need to create a Consumer Group for your stream. This is done using the XGROUP CREATE command.

It tells Redis: "Hey, for this stream, make a new group." You also specify an ID, which is typically 0 or $ to start reading from the beginning or end of the stream.

Let's create a group named mygroup for a stream called mystream. The $ means "start reading from the latest entry." The MKSTREAM option is crucial: it creates the stream if it doesn't already exist!

Try it out:

Once a group is created, consumers can start reading messages using the XREADGROUP command. This command is designed specifically for consumer groups.

It ensures that each message is delivered to only one consumer within the group. If a consumer fails to acknowledge a message, it remains pending and can be claimed later.

The XREADGROUP command has several important parameters:

• GROUP <groupname> <consumername>: Specifies which group and consumer are reading.
• COUNT <N>: Optional. Limits the number of messages to read.
• BLOCK <milliseconds>: Optional. Blocks the client if no messages are available.
• STREAMS <streamname> <ID>: The stream to read from and the ID. Use > to get new messages that haven't been delivered to any other consumer in the group yet.

First, let's add a few messages to our mystream:

Now, let's have a consumer named consumer-1 from mygroup read from mystream. The > ID means "new, unread messages."

After a consumer successfully processes a message, it's crucial to acknowledge it. This tells Redis that the message has been handled and can be removed from the consumer's Pending Entries List (PEL).

Acknowledgment prevents the same message from being re-delivered to another consumer if the current one crashes or goes offline before completing the task.

The command to acknowledge messages is XACK. Its syntax is straightforward:

You provide the stream name, the consumer group name, and one or more message IDs that have been successfully processed.

Let's assume you received message IDs 1678881234567-0 and 1678881234568-0 from the previous read. You would acknowledge them like this (replace with your actual IDs):

You've learned how to create groups, read messages, and acknowledge them. Which of the following statements about Redis Consumer Groups is TRUE?

Great job! You've mastered the fundamentals of implementing Redis Consumer Group logic:

• We used XGROUP CREATE to set up a new group for a stream.
• You learned to read messages cooperatively as a consumer with XREADGROUP.
• We saw how XACK is vital for acknowledging processed messages, managing the Pending Entries List (PEL).

Next, we'll explore how to handle pending messages and consumer failures more robustly!