利用Redis Stream实现一个可靠的消息队列

标签：读取 Stream 队列 Redis id 消息 key ID mystream

Redis Streams 简介

Redis Streams 是 Redis 5.0 引入的一种新数据结构，专门用于处理日志和消息流数据。它结合了多种数据结构的优点，提供了高效的消息存储和消费机制。Redis Streams 可以用于实时数据处理、事件驱动的系统、日志聚合和消息队列等场景。

主要特点

持久化：redis的publish/subscribe功能发送的消息是无法持久话的，stream消息存储在 Redis 中，并可以持久化到磁盘，确保数据不会丢失。
消费者组：支持多消费者组，每个组内可以有多个消费者，实现消息的并发处理。
阻塞读取：消费者可以阻塞等待新消息，实现实时处理。
灵活的消息确认机制：消费者处理完消息后需要确认（ACK），确保消息不会丢失。
高效的内存使用：采用压缩存储，优化内存使用。

使用场景

实时日志处理：如日志聚合、监控和告警系统。
消息队列：分布式系统中的消息传递和处理。
事件溯源：记录和重放系统中的事件。
数据管道：流数据处理和传输，如实时分析和 ETL。

核心概念

流（Stream）：消息的有序集合，每个消息都有一个唯一的 ID。
消息（Entry）：由多个字段组成的键值对。
消息 ID：每条消息都有一个唯一的 ID，由时间戳和序列号组成。
消费者组（Consumer Group）：用于实现消息的并发处理，每个组内有多个消费者。
待处理消息（Pending Entry List, PEL）：存储每个消费者组中未确认处理的消息。

队列结构图

创建流并添加消息

XADD

将消息添加到流中。

XADD stream-name [ID or *] field1 value1 [field2 value2 ...]

stream-name: 流的名称。
ID or *: 消息的 ID。使用 * 表示自动生成唯一 ID。
field1 value1 [field2 value2 ...]: 消息的字段和值。

示例：

XADD mystream * sensor-id 1234 temperature 19.8

读取消息

XRANGE

获取指定范围内的消息。

XRANGE stream-name start end [COUNT count]

stream-name: 流的名称。
start: 起始 ID。- 表示最小 ID。
end: 结束 ID。+ 表示最大 ID。
COUNT count: （可选）返回的最大消息数量。

示例：

XRANGE mystream - +

XREVRANGE

逆序获取指定范围内的消息。

XREVRANGE stream-name end start [COUNT count]

stream-name: 流的名称。
end: 结束 ID。+ 表示最大 ID。
start: 起始 ID。- 表示最小 ID。
COUNT count: （可选）返回的最大消息数量。

示例：

XREVRANGE mystream + -

XREAD

XREAD 命令是 Redis 中用于从一个或多个流（stream）读取数据的命令。它是一种阻塞或非阻塞的读取操作，可以用于构建消息队列、事件处理等场景。以下是 XREAD 命令的详细讲解：

基本语法

XREAD [COUNT count] [BLOCK milliseconds] STREAMS key [key ...] ID [ID ...]

参数说明

COUNT count：可选参数，用于限制返回的条目数。如果不指定，默认会返回尽可能多的条目。
BLOCK milliseconds：可选参数，用于阻塞读取。如果指定，命令会在没有新条目可读时阻塞指定的毫秒数。如果不指定，命令会立即返回现有的条目（如果有）。
STREAMS key [key ...]：指定要读取的一个或多个流的键。
ID [ID ...]：指定每个流的起始 ID，表示从哪个 ID 之后开始读取。

示例

假设我们有两个流 mystream1 和 mystream2，可以使用以下命令从这两个流中读取数据：

XREAD COUNT 2 STREAMS mystream1 mystream2 0 0

这条命令会从 mystream1 和 mystream2 中读取数据，并且每个流最多读取 2 条消息。0 表示从流的开头开始读取。

阻塞读取示例

使用阻塞读取时，如果没有新消息，命令会阻塞指定的毫秒数：

XREAD BLOCK 5000 STREAMS mystream1 0

这条命令会在 mystream1 中阻塞最多 5 秒钟，等待新消息的到来。

读取特定 ID 之后的消息

可以指定从某个特定 ID 之后开始读取消息：

XREAD STREAMS mystream1 1526569495631-0

这条命令会从 mystream1 中读取 ID 为 1526569495631-0 之后的消息。

返回结果格式

XREAD 命令的返回结果是一个包含流数据的列表，每个流的数据以包含流名和条目的元组表示。条目是一个列表，包含 ID 和消息字段-值对。例如：

1) 1) "mystream1"
   2) 1) 1) "1526569495631-0"
         2) 1) "field1"
            2) "value1"
            3) "field2"
            4) "value2"

注意事项

XREAD 命令支持从多个流中读取数据，这在构建复杂的消息处理逻辑时非常有用。
使用 BLOCK 参数时要小心，以避免不必要的长时间阻塞，尤其是在高负载的生产环境中。

通过 XREAD 命令，Redis 提供了强大的流数据处理能力，可以满足各种实时数据处理和消息队列的需求。

消费者操作

XGROUP CREATE

创建消费组。消费组用于管理消费者和队列读取记录。Stream中的消费组有两个特点：

从资源结构上说消费者从属于一个消费组
一个队列可以拥有多个消费组。不同消费组之间读取队列互不干扰

语法格式：

XGROUP [CREATE key groupname id-or-$] [SETID key groupname id-or-$] [DESTROY key groupname] [DELCONSUMER key groupname consumername]

key：队列名称，如果不存在就创建
groupname：组名
id: $表示从尾部开始消费，只接受新消息，当前Stream消息会全部忽略

命令使用：

为队列mystream创建一个消费组 mqGroup，从第一个消息开始读

127.0.0.1:6379> XGROUP CREATE mystream mqGroup 0
OK

XREADGROUP

读取队列的消息。在读取消息时需要指定消费者，只需要指定名字，不用预先创建。

语法格式：

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

group：消费组名
consumer：消费者名
count：读取数量
BLOCK milliseconds：阻塞读以及阻塞毫秒数。默认非阻塞。和XREAD类似
key：队列名
id：消息ID。ID可以填写特殊符号>，表示未被组内消费的起始消息

命令使用：
创建消费者c1和c2，各读取一条消息

127.0.0.1:6379> XREADGROUP GROUP mqGroup c1 COUNT 1 STREAMS mystream >
            
127.0.0.1:6379> XREADGROUP GROUP mqGroup c2 count 1 streams mystream >

可以进行组内消费的基本原理是，STREAM类型会为每个组记录一个最后读取的消息ID（last_delivered_id），这样在组内消费时，就可以从这个值后面开始读取，保证不重复消费。

消费组消费时，还有一个必须要考虑的问题，就是若某个消费者，消费了某条消息，但是并没有处理成功时（例如消费者进程宕机），这条消息可能会丢失，因为组内其他消费者不能再次消费到该消息了

XPENDING

为了解决组内消息读取但处理期间消费者崩溃带来的消息丢失问题，Stream 设计了 Pending 列表，用于记录读取但并未确认完毕的消息。
语法格式：

XPENDING key group [[IDLE min-idle-time] start end count [consumer]]

key：队列名
group：消费组名
start：开始值，-表示最小值
end：结束值，+表示最大值
count：数量

命令使用：

首先查看队列中的消息数量有3个，然后查看已读取未处理的消息有两个。

127.0.0.1:6379> xlen mystream
(integer) 3

127.0.0.1:6379> xpending mystream mqGroup
1) (integer) 2
2) "1721185317453-0"
3) "1721185327505-0"
4) 1) 1) "c1"
      2) "1"
   2) 1) "c2"
      2) "1"

队列中一共三条信息，有两条被消费但未处理完毕，也就是上面XREADGROUP消费的两条。一个是消费者c1，另一个是c2。

获取未确认的详细信息

127.0.0.1:6379> xpending mystream mqGroup - + 10

XACK

对于已读取未处理的消息，使用命令 XACK 完成告知消息处理完成
XACK 命令确认消费的信息，一旦信息被确认处理，就表示信息被完善处理。

语法格式：

XACK key group id [id ...]

key: stream 名
group：消费组
id：消息ID

命令使用：

确认消息1674985213802-0

127.0.0.1:6379> XACK mystream mqGroup 1674985213802-0
(integer) 1
127.0.0.1:6379>

XCLAIM

某个消费者读取了消息但没有处理，这时消费者宕机或重启等就会导致该消息失踪。那么就需要该消息转移给其他的消费者处理，就是消息转移。XCLAIM来实现消息转移的操作。

语法格式：

XCLAIM key group consumer min-idle-time id [id ...] [IDLE ms]
  [TIME unix-time-milliseconds] [RETRYCOUNT count] [FORCE] [JUSTID]
  [LASTID id]

key：队列名称
group ：消费组
consumer：消费组里的消费者
min-idle-time 最小时间。空闲时间大于min-idle-time的消息才会被转移成功
id：消息的ID

转移除了要指定ID外，还需要指定min-idle-time，min-idle-time是最小空闲时间，该值要小于消息的空闲时间，这个参数是为了保证是多于多长时间的消息未处理的才被转移。比如超过24小时的处于pending未xack的消息要进行转移
同时min-idle-time还有一个功能是能够避免两个消费者同时转移一条消息。被转移的消息的IDLE会被重置为0。假设两个消费者都以2min来转移，第一个成功之后IDLE被重置为0，第二个消费者就会因为min-idle-time大与空闲时间而是失败。

命令使用：
目前未确认的消息

127.0.0.1:6379> xpending mystream mqGroup - + 10
1) 1) "1721185327505-0"
   2) "c2"
   3) (integer) 263196
   4) (integer) 1

id: 1721185327505-0
空闲时间：263196，单位ms
读取次数：1

将c2未处理的消息转移给c1。

127.0.0.1:6379> XCLAIM mystream mqGroup c1 3600000 1721185327505-0

查看未确认的消息
消息已经从c2转移给c1，IDLE重置，读取次数加1。转移之后就可以继续处理这条消息。

127.0.0.1:6379> xpending mystream mqGroup - + 10
1) 1) "1721185327505-0"
   2) "c1"
   3) (integer) 63196
   4) (integer) 2

通常转移操作的完整流程是：

先用xpending命令找出所有未确认的消息
再用xclaim命令转移所有未确认消息

XINFO

Stream提供了XINFO来实现对服务器信息的监控

查看队列信息

127.0.0.1:6379> xinfo stream mystream

消费组信息

127.0.0.1:6379> xinfo groups mystream

消费者组成员信息

127.0.0.1:6379> xinfo consumers mystream mqGroup

完整 Java 示例代码

生产者（Producer）

import redis.clients.jedis.Jedis;
import java.util.Map;

public class Producer {
    public static void main(String[] args) {
        Jedis jedis = new Jedis("localhost", 6379);
        
        for (int i = 1; i <= 10; i++) {
            String messageId = jedis.xadd("mystream", null, Map.of(
                "sensor-id", String.valueOf(i),
                "temperature", String.valueOf(20 + i)
            ));
            System.out.println("Added message with ID: " + messageId);
        }
        
        jedis.close();
    }
}

消费者（Consumer）

import redis.clients.jedis.Jedis;
import redis.clients.jedis.StreamEntryID;
import redis.clients.jedis.StreamEntry;
import java.util.AbstractMap;
import java.util.List;
import java.util.Map;

public class Consumer {
    public static void main(String[] args) {
        Jedis jedis = new Jedis("localhost", 6379);
        
        while (true) {
            List<Map.Entry<String, List<StreamEntry>>> streams = jedis.xreadGroup(
                "mygroup", "consumer1", 1, 0, false, new AbstractMap.SimpleEntry<>("mystream", StreamEntryID.UNRECEIVED

_ENTRY));
            
            if (streams != null && !streams.isEmpty()) {
                for (Map.Entry<String, List<StreamEntry>> stream : streams) {
                    for (StreamEntry entry : stream.getValue()) {
                        String id = entry.getID().toString();
                        Map<String, String> fields = entry.getFields();
                        
                        System.out.println("Processing message with ID: " + id);
                        fields.forEach((key, value) -> System.out.println(key + ": " + value));
                        
                        // 模拟消息处理
                        try {
                            Thread.sleep(1000);
                        } catch (InterruptedException e) {
                            Thread.currentThread().interrupt();
                        }
                        
                        // 确认消息已处理
                        jedis.xack("mystream", "mygroup", new StreamEntryID(id));
                        System.out.println("Acknowledged message with ID: " + id);
                    }
                }
            }
        }
    }
}

通过这些示例代码和命令，您可以创建一个基于 Redis Streams 的消息队列系统，处理和管理实时数据流。

标签：读取,Stream,队列,Redis,id,消息,key,ID,mystream
From： https://blog.csdn.net/2401_85380274/article/details/140491343

利用Redis Stream实现一个可靠的消息队列

Redis Streams 简介

主要特点

使用场景

核心概念

队列结构图

创建流并添加消息

XADD

读取消息

XRANGE

XREVRANGE

XREAD

基本语法

参数说明

示例

阻塞读取示例

读取特定 ID 之后的消息

返回结果格式

注意事项

消费者操作

XGROUP CREATE

XREADGROUP

XPENDING

XACK

XCLAIM

XINFO

完整 Java 示例代码

生产者（Producer）

消费者（Consumer）

相关文章

赞助商

阅读排行