标签：消费 15 Stream stream 队列精讲 name user 消息

Redis系列1：深刻理解高性能Redis的本质
 Redis系列2：数据持久化提高可用性
 Redis系列3：高可用之主从架构
 Redis系列4：高可用之Sentinel(哨兵模式）
Redis系列5：深入分析Cluster 集群模式
 追求性能极致：Redis6.0的多线程模型
 追求性能极致：客户端缓存带来的革命
 Redis系列8：Bitmap实现亿万级数据计算
 Redis系列9：Geo 类型赋能亿级地图位置计算
 Redis系列10：HyperLogLog实现海量数据基数统计
 Redis系列11：内存淘汰策略
 Redis系列12：Redis 的事务机制
 Redis系列13：分布式锁实现
 Redis系列14：使用List实现消息队列

1 介绍

我们上一篇介绍了如何使用List实现消息队列么，但是我们也看到很多局限性，如下：

不支持消息确认机制，没有很好的ACK应答
不支持消息回溯，无法排查问题和做消息分析
List遵循FIFO机制，所以存在消息堆积的风险。
查询效率低，作为线性结构，List中定位一个数据需要进行遍历，O(N)的时间复杂度。
不存在消费组（Consumer Group）的概念，无法进行分组消费和批量消费

Redis中有三种消息队列模式：

名称	简要说明
List	不支持消息确认机制（Ack），不支持消息回朔
pubSub	不支持消息确认机制（Ack），不支持消息回朔，不支持消息持久化
stream	支持消息确认机制（Ack），支持消息回朔，支持消息持久化，支持消息阻塞

可以看出，作为Redis 5.0 引入的专门为消息队列设计的数据类型，Stream 功能更加健全，更适合做消息队列分发。
Stream 可以包含 0个到 n个元素的有序队列，并根据ID的大小进行排序。

Stream类型消息队列的具备以下命令特点：

可以序列化生成消息ID，方便索引、排序
消息可回朔
支持Consumer Groups 消费组：多消费者消息争抢，加快消费速度
可以阻塞读取消息和非阻塞读取消息
没有消息漏读风险
有ACK消息确认机制，保证消息至少被消费一次
支持多播模式：可以让队列从逻辑上分组进行隔离消费

这些特性，基本达到了一个消息中间件的基本能力，比如：

类似 Kafka 的 Consumer Groups 的概念，它也具备了消费组的能力。
类似 Rocket MQ的持久化能力，以及高可用的文件存储机制，它也具备了消息的持久化和主从复制机制，可以记录访问位置，方便后续其他时间段继续访问，避免数据丢失。
详细的stream操作见官网文档：https://redis.io/docs/data-types/streams-tutorial/

2 XADD 消息写入

即讲消息添加到队列中，语法如下：

# 队列名称后面的队列id如果用 * 号表示 ，这代表让 Redis 为插入的消息自动生成唯一 序列化ID，当然也可以自己指定。
# 后面可以包含多个键值对，代表多个消息元素
XADD 队列名称 队列id key1 value1 [key2 value2 ....]

注释比较清楚，以下举例说明：

> xadd stream_user * user_id 1 user_name brand age 18
"1680926230000-0"

不指配*，这可以直接指定顺序Id

> XADD stream_user 0-1 user_name lili
0-1
> XADD stream_user 0-2 user_name brand
0-2
> XADD stream_user 0-* user_name candy
0-3

队列的消息ID 由两部分组成：

毫秒级别的当前时间的时间戳；
顺序编号。从 0 为起始值，用于区分同一时间内产生的多个ID，如果同一个时间戳内生成多ID，按序号顺序增长，这种方式可解决顺序识别和时间回拨问题。

通过这种时间戳 + 顺序编号的模式，变成数据Append的模式，这种流式记性数据顺序推送的方式符合MQ的基本消费逻辑，也为后面的有序性消费提供基本条件。

2 XREAD 消息阅读

即讲消息从队列中读取出来（消费），语法如下：

# COUNT：指的是对于每个Steam流中最多读取几个元素；
# BLOCK：当配置时阻塞读取，队列中没有消息即阻塞等待， 单位是ms，0 表示无限等待，类似MQ中的订阅，等待新消息出现。
# key：表示stream的名称
# ID：消息 id，读取消息的时候可以指定Id，并且指定某个Id的第一条甚至第n条开始读取，图中0-0 则表示从队列Id为0的队列的第1个元素开始读取。

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

注释比较清楚，以下举例说明：

XREAD COUNT 1 BLOCK 0 STREAMS stream_user 0-0
1) 1) "stream_user"
    2)  1) 1) "1680926230000-0"
         2) 1) "user_name"
             2) "brand"
             3) "age"
             4) "18"

如何顺序性消费：我们每次读取之后都会返回消息Id和序号，比如上面的 1680926230000-0,所以在下一次调用的时候，可以用上一次返回的ID序号作为参数，就可以从指定位置上进行消费。
问题：XREAD之后数据并没有删除，所以没记住读取的位置，下次可能重复阅读，造成重复消费。所以需要消费确认机制（即ACK）。

3 消费者组模式（Consumer Group）

消息队列很重要的一个能力就是分组消费（Consumer Group），无论是Kafka 还是 RabbitMQ。他允许队列从逻辑上进行分组来保证隔离消费。
这是典型的多播模，如下图所示：

它有如下特点：

Redis Stream 实际结构是一个链式的队列，一个消息由消息Id和消息内容组成，消息Id具有唯一性；
消费组的状态是独立的，像图中的GroupA、GroupB、GroupC，Stream 消息可以被这几个组消费；
同时一个消费者组可以有多个消费者，但是他们的竞选关系，任意消费者消费之后就会导致 last_deliverd_id 偏移，这样避免了重复消费。
每个消费者都携带pending_ids 变量，记录读取但还未消费（未被ack）的消息，来保证消息有且仅有一次被消费。

消费组实现的消息队列主要有3类指令，如下：

XGROUP：用于创建消费群组，包括注销和其他管理职能。
XREADGROUP：消费者群组，通过这些组从流中有序读取数据。
XACK：通过该命令，消费者将处理的完的消息标记为已正确完成。

3.1 写入队列数据

咱们先做一下数据准备，创建队列，并往里面写入一些数据，如下：

> xadd stream_user * user_id 1 user_name brand age 18
"1681126033000-0"
> xadd stream_user * user_id 2 user_name jay age 19
"1681126222000-0"
> xadd stream_user * user_id 3 user_name candy age 20
"1681126235000-0"
> xadd stream_user * user_id 4 user_name lili age 21
"1681126251000-0"
> xadd stream_user * user_id 5 user_name hanry age 22
"1681126263000-0"

3.2 创建消费者群组

这个的做法就是在队列中创建消费者组，然后指定消费的位置。
语法如下：

# stream_name：队列名称
# consumer_group：消费者组
# msgIdStartIndex：消息Id开始位置
# msgIdStartIndex：消息Id结束位置
xgroup create stream_name consumer_group msgIdStartIndex-msgIdStartIndex

下面是具体实现示例，为队列 stream_user 创建了消费组1（consumer_group1）和消费组2（consumer_group2）：

> xgroup create stream_user consumer_group1 0-0
OK
> xgroup create stream_user consumer_group2 0-0
OK

3.3 读取消费组信息

消费队列消息的语法如下:

# groupName: 消费者群组名
# consumerName: 消费者名称
# COUNT number: count 消费个数
# BLOCK ms: 阻塞时间，如果为 0 则代表无线阻塞
# streamName: 队列名称 
# id: 消息消费ID
# []：代表可选参数
# `>`：放在命令参数的最后面，表示从尚未被消费的消息开始读取；

XREADGROUP GROUP groupName consumerName [COUNT number] [BLOCK ms] STREAMS streamName [stream ...] id [id ...]

实现示例：消费组 consumer_group1 的消费者 consumer1 从 stream_user 中以阻塞的方式读取一条消息：

XREADGROUP GROUP consumer_group1 consumer1 COUNT 1 BLOCK 0 STREAMS stream_user >
1) 1) "stream_user"
    2) 1) 1) "1681126033000-0"
        2) 1) "user_id"
            2) "1"
            3) "user_name"
            4) "brand"
            5) "age"
            6) "18"

这边需要主意的是，同一个消费组内，消息只能单次消费，如果被消费组内消费过了，就不会被同组的其他消费组读取到。
如下：

XREADGROUP GROUP consumer_group1 consumer2 COUNT 1 BLOCK 0 STREAMS stream_user >
1) 1) "stream_user"
    2) 1) 1) "1681126222000-0"
        2) 1) "user_id"
            2) "2"
            3) "user_name"
            4) "jay"
            5) "age"
            6) "19"

上面 user_name 为 brand 的数据已经被consumer1消费了，所以consumer2 就读不到了，只能读取到下一条 user_name 为 jay 的数据。
多个消费者可以达到流量分摊的目的，为大业务流量的场景做负载和分流。如下图，多个消费者相对平均的进行消息消费。

3.4 XPENDING 检查已读取但未ACK的数据

有时候会出现这种情况，就是消费者组或者消费者发生了故障，甚至整个消费者都故障重启了，那么如何避免消息丢失呢，那就是将读取到的但是还没消费的数据进行暂存。
Redis在Stream内部实现了一个待决队列（pending List），消费者读取之后且没有进行ACK的数据都保存在这里。
这种情况就是：

消费者使用 XREADGROUP 读取消息
读取完成之后，发生故障或者异常，没有给 Stream 发送 XACK 命令，消息依然保留在Stream 的 pending List中。

比如查看 stream_user 中的消费组 consumer_group1 中各个消费者已读取未确认的消息信息：

XPENDING stream_user consumer_group1
1) (integer) 2   # 未确认消息条数
2) "1681126235000-0"   # consumer_group1 消费组中所有消费者读取的最小ID
3) "1681126251000-0"   # consumer_group1 消费组中所有消费者读取的最大ID
4) 1) 1) "consumer1"
        2) "1"
   2)  1) "consumer2"
        2) "1"

3.5 消息消费完成之后确认（ACK）

正如3.4中所说的相关内容消费完之后，需要 ACK 通知 Streams，然后Stream除消息。否则就会造成消息变成待决队列中，可能造成重复消费的情况。
执行命令语法如下：

# XACK stream_name group_name ID [ID ...]
# stream_name：队列名称
# group_name：消费组名称
# ID：消费ID，可多选

XACK stream_user consumer_group1 1681126235000-0 1681126251000-0
(integer) 2

ack的本意就是对消费完成的消息进行确认，业务处理没有问题之后进行一个check的过程，代表这个消息已经被消费完了。流程如下：

4 使用Redission实现Stream队列能力

4.1 添加maven依赖和配置基本连接

# maven信息
<dependency>
  <groupId>org.redisson</groupId>
  <artifactId>redisson-spring-boot-starter</artifactId>
  <version>3.16.8</version>
</dependency>

# 基本配置
spring:
  application:
    name: redission_test
  redis:
    host: x.x.x.x
    port: 6379
    ssl: false
    password: xxxx.xxxx

4.2 Java程序实现

@Slf4j
@Service
public class StreamQueueService {

    @Autowired
    private RedissonClient redissonClient;
	
  /**
     * 生产消息内容
     *
     * @param msg
     * @return
     */
    @Override
    public void produceMsg(String msg) {
        RStream<String, String> stream = redissonClient.getStream("stream_user");
        stream.add("user_id", "1");
        stream.add("user_name", "brand");
        stream.add("age", "18");
    }

   /**
     * 消费消息内容
     */
    @Override
    public void consumeMessage() {
        // 根据队列名称获取消息队列
        RStream<String, String> stream = redissonClient.getStream("stream_user");
        // 创建消费者小组
        stream.createGroup("consumer_group1", StreamMessageId.ALL);
        // 消费者读取消息
        Map<StreamMessageId, Map<String, String>> msgs
                = stream.readGroup("consumer_group1", "consumer1");
        for (Map.Entry<StreamMessageId, Map<String, String>> entry : msgs.entrySet()) {
            Map<String, String> msg = entry.getValue();
            log.info(msg);
            // todo：处理消息的业务逻辑代码
            stream.ack("consumer_group1", entry.getKey());
        }
    }
}

5 总结

相对List，Stream的能力有比较大的提升：

支持消息确认机制（ACK应答确认）
支持消息回溯，方便排查问题和做消息分析
存在消费组（Consumer Group）的概念，可以进行分组消费和批量消费，可以负载多个消费实例

标签：消费,15,Stream,stream,队列,精讲,name,user,消息
From： https://www.cnblogs.com/wzh2010/p/17205395.html

Redis系列15：使用Stream实现消息队列（精讲）

1 介绍

2 XADD 消息写入

2 XREAD 消息阅读

3 消费者组模式（Consumer Group）

3.1 写入队列数据

3.2 创建消费者群组

3.3 读取消费组信息

3.4 XPENDING 检查已读取但未ACK的数据

3.5 消息消费完成之后确认（ACK）

4 使用Redission实现Stream队列能力

4.1 添加maven依赖和配置基本连接

4.2 Java程序实现

5 总结

相关文章

赞助商

阅读排行

Redis系列15：使用Stream实现消息队列（精讲）

1 介绍

2 XADD 消息写入

2 XREAD 消息阅读

3 消费者组模式（Consumer Group）

3.1 写入队列数据

3.2 创建消费者群组

3.3 读取消费组信息

3.4 XPENDING 检查已读取但未ACK的数据

3.5 消息消费完成之后确认（ACK）

4 使用Redission实现Stream队列能力

4.1 添加maven依赖 和 配置基本连接

4.2 Java程序实现

5 总结

相关文章

赞助商

阅读排行

4.1 添加maven依赖和配置基本连接