⑨【Stream】Redis流是什么？怎么用？： Stream [使用手册]

爱被打了一巴掌 2024-02-17 10:32 45阅读 0赞

![在这里插入图片描述][8d8e0d8a542946af858489a536162608.gif_pic_center]

> 个人简介：Java领域新星创作者；阿里云技术博主、星级博主、专家博主；正在Java学习的路上摸爬滚打，记录学习的过程~  
> 个人主页：[.29.的博客][.29.]  
> 学习社区：[进去逛一逛~][Link 1]

> [![在这里插入图片描述][d96175b2bec4472ab4ea0ada46da4af4.png_pic_center]][d96175b2bec4472ab4ea0ada46da4af4.png_pic_center 1]

#### ⑨Redis Stream基本操作命令汇总 ####

*  一、Redis流 （Stream）
 *  二、Redis Stream 基本操作命令
 *   *  1.队列相关命令
     *   *  ① xadd 向Stream队列添加消息
         *  ② xrange 获取指定队列的消息列表
         *  ③ xrevrange 获取指定队列的消息列表
         *  ④ xdel 删除消息列表
         *  ⑤ xlen 获取Stream队列消息的长度
         *  ⑥ xtrim 对Stream的长度进行截取
         *  ⑦ xread 获取消息，只会返回大于指定id的消息
     *  2.消费组相关命令
     *   *  ① xgroup create 创建消费组
         *  ② xreadgroup GROUP 指定消费组中消费者来消费消息
         *  ③ xpending 查询已读取但尚未确认的消息
         *  ④ xack 向消息队列确认消息已经处理完成
         *  ⑤ xinfo stream 查询流的详细信息
         *  重点问题

--------------------

## 一、Redis流 （Stream） ##

`Redis 5.0 之前，实现消息队列的两种方案`：

> 方案一：List实现
> 
> ![在这里插入图片描述][4f0591e390e94a4ebf2fc30c4f707573.png]

> 方案二：发布订阅（Pub/Sub）
> 
> ![在这里插入图片描述][033c45cb659647d9811cfba94ec2d5cd.png]

--------------------

`Redis 5.0 后`：

*  Redis Steam 是redis 5.0 新增的一种数据结构
 *  Redis Stram**可以用来实现消息队列**，它支持消息的持久化、支持自动生成全局唯一ID、支持ack确认消息的模式、支持消费组模式等，让消息队列更加稳定和可靠

`Stream 结构`：

![在这里插入图片描述][e98195ec4bcb405589ac6324a564367f.png]

*  **Stream本质是一个消息链表，将所有加入的消息都串起来，每个消息都有一个唯一的ID和对应的内容**
 *  `Message Content` : **消息内容**
 *  `Consumer group` ：**消费组**，通过XGROUP CREATE命令创建，一个消费组中可以有多个消费者
 *  `Last_delivered_id` ：**游标**，每个消费组会有一个游标Last\_delivered\_id，任意一个消费者读取了消息都会使得这个游标往前移动
 *  `Consumer` ：消费组中的消费者
 *  `Pending_ids` ：每个消费者都会有一个**状态变量，用于记录被当前消费者已读取但未被ack确认的消息ID**，如果客户端没有ack确认，这个变量里面的消息ID会愈来愈多，一旦某个消息被ack，它就开始减少。这个Pending\_ids变量在Redis官方被称为PEL(Pending Entries List)，记录了当前已经被客户端读取的但还未ack (Acknowledge character：确认字符）的消息，它用来确保客户端至少消费了消息一次，而不会在网络传输的中途丢失而导致没处理。

--------------------

## 二、Redis Stream 基本操作命令 ##

### 1.队列相关命令 ###

#### ① xadd 向Stream队列添加消息 ####

`xadd key [NOMKSTREAM] [MAXLEN|MINID [=|~] threshold [LIMIT count]] *|id field value [field value ...]`

# 消息队列：mystream ，不存在会先创建再添加消息
    # * ： * 号表示服务器自动生成 MessageID
    # id —— 29 ， name —— little29 ，两对键值对，也是添加到队列中的消息
    xadd mystream * id 29 name little29

> ![在这里插入图片描述][1263f61500864a44990d63b1dd80cd7b.png]  
> ![在这里插入图片描述][662519aaf7d844c69f0fa584c8d2b098.png]  
> ![在这里插入图片描述][62d614f9cb894073a94aa84ea7ee112a.png]

--------------------

#### ② xrange 获取指定队列的消息列表 ####

`xrange key start end [COUNT count]`

`start` : 代表开始值

`end` : 代表结束值

`count` ：表示最多获取多少个值

# -号：代表最小值
    # +号：代表最大值
    xrange mystream - +

> ![在这里插入图片描述][48bb45d89c074a03adfe86185d80a5de.png]

--------------------

#### ③ xrevrange 获取指定队列的消息列表 ####

*  与xrange的区别在于，获取消息队列元素的方向是相反的，end在前，start在后

`xrevrange key end start [COUNT count]`

`start` : 代表开始值

`end` : 代表结束值

`count` ：表示最多获取多少个值

# -号：代表最小值
    # +号：代表最大值
    xrange mystream + -

> ![在这里插入图片描述][10de90b39860448e9f22638061adce56.png]

--------------------

#### ④ xdel 删除消息列表 ####

*  根据指定的MessageID，删除一个或多个消息列表

`xdel key id [id ...]`

xdel mystream 1681006258096-0

> ![在这里插入图片描述][7f0132b9a0404c62959cd24684146aaa.png]

--------------------

#### ⑤ xlen 获取Stream队列消息的长度 ####

`xlen key`

xlen mystream

> ![在这里插入图片描述][ddbad89797c24d9b8e41c26490ebde2b.png]

--------------------

#### ⑥ xtrim 对Stream的长度进行截取 ####

`xtrim key MAXLEN|MINID [=|~] threshold [LIMIT count]`

`MAXLEN` : 允许的最大长度，超过就会对流进行截取

`MINID` : 允许的最小ID，从某个ID值开始，比这个ID值小的将会被抛弃

# 1681007772647-0作为最小id，id值小于它的会被抛弃
    xtrim mystream minid 1681007772647-0

> ![在这里插入图片描述][65636c16c20042b49ae14a04cfe57dee.png]

--------------------

#### ⑦ xread 获取消息，只会返回大于指定id的消息 ####

`xread [COUNT count] [BLOCK milliseconds] STREAMS key [key ...] id [id ...]`

`COUNT` : 最多读取多少条消息

`BLOCK` : 表示是否以阻塞的方式读取消息，默认不阻塞，如果milliseconds设置为0，表示永久阻塞

# $代表特殊ID，表示以当前Stream已经存储的最大的ID作为最后一个ID，当前Stream中不存在大于当前最大ID的消息，因此此时返回nil
    xread count 2 streams mystream $
    # 0-0代表从最小的ID开始获取Stream中的消息，当不指定count，将会返回Stream中的所有消息，注意也可以使用0（00/000也都是可以的……）
    xread count 2 streams mystream 0-0

> ![在这里插入图片描述][2a01f31a658540e9baccb4b060600fb4.png]  
> ![在这里插入图片描述][6e30303604f64502a75c4424f39ea634.png]

--------------------

### 2.消费组相关命令 ###

*  Stream中的消息一旦被消费组里的一个消费者读取了，就不能再被该消费组内的其他消费者读取了，即**同一个消费组里得消费者不能消费同一条消息。**
 *  但是，**不同消费组中的消费者可以消费同一条消息。**
 *  **消费组的`目的`：** 让组内多个消费者共同分担读取消息，所以，我们通常会让每个消费者读取部分消息，从而实现消息读取负载在多个消费者间是均衡分布的。

#### ① xgroup create 创建消费组 ####

`xgroup create key groupname id|$ [MKSTREAM] [ENTRIESREAD entries_read]`

*  创建消费者组的时候必须指定 ID, ID 为 0 表示从头开始消费，为 $ 表示只消费新的消息
 *  $表示从Stream尾部开始消费
 *  0表示从Stream头部开始消费

# 从头开始消费
    xgroup create mystream group1 0
    # 只消费新的消息
    xgroup create mystream group2 $

> ![在这里插入图片描述][1a9aaa34b74b4019adf1d1a3c5abb018.png]

--------------------

#### ② xreadgroup GROUP 指定消费组中消费者来消费消息 ####

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

*  命令末尾加上“>”,表示从第一条尚未被消费的消息开始读取

# 消费组group1中的消费者consumer1从mystream消息队列中读取所有消息
    # “>”：表示从第一条尚未被消费的消息开始读取
    xreadgroup GROUP group1 consumer1 STREAMS mystream >

> ![在这里插入图片描述][a0ae5da612d84bf0a08c383de5d8b77c.png]

--------------------

#### ③ xpending 查询已读取但尚未确认的消息 ####

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

# 查询指定消费组group1对mystream中已读取但未确认的消息
    xpending mystream group1

> ![在这里插入图片描述][64e0ef1d861741b4b8023788c25967f5.png]

--------------------

#### ④ xack 向消息队列确认消息已经处理完成 ####

`xack key group id [id ...]`

# 通过xack，向消息队列确认指定消息ID：1681007772647-0的消息已经处理完成
    xack mystream group1 1681007772647-0

> ![在这里插入图片描述][416d809993ce48ff88ecb9cc412441fe.png]

--------------------

#### ⑤ xinfo stream 查询流的详细信息 ####

`xinfo stream key [FULL [COUNT count]]`

# 查询流：mystream的详细信息
    xinfo stream mystream

> ![在这里插入图片描述][0cd949e56d6b4ec297e542a461c743cc.png]

#### 重点问题 ####

![在这里插入图片描述][b310fe73a0884dac9b23d44af1eca2dc.png]

--------------------

[![在这里插入图片描述][20a7f1b58dfb4660b75d7f021c157d57.png_pic_center]][20a7f1b58dfb4660b75d7f021c157d57.png_pic_center 1]

[8d8e0d8a542946af858489a536162608.gif_pic_center]: https://image.dandelioncloud.cn/pgy_files/images/2024/02/06/17d82b700cf243619683f7124e49bf6f.png
[.29.]: https://blog.csdn.net/ebb29bbe?spm=1018.2226.3001.5343
[Link 1]: https://bbs.csdn.net/forums/185241f773fb401b8ca7994dc3d02333?joinKey=cyru4ex9o7wq-mz6v1koj5x-1-add3ae8dc3827dbbd4360c79af92bbbe
[d96175b2bec4472ab4ea0ada46da4af4.png_pic_center]: https://img-blog.csdnimg.cn/d96175b2bec4472ab4ea0ada46da4af4.png#pic_center
[d96175b2bec4472ab4ea0ada46da4af4.png_pic_center 1]: https://image.dandelioncloud.cn/pgy_files/images/2024/02/06/cd13437a7d9641caa085dcfe0fcb8df3.png
[4f0591e390e94a4ebf2fc30c4f707573.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/02/06/4d920d36e6f14ec09125737f9557bc0f.png
[033c45cb659647d9811cfba94ec2d5cd.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/02/06/8b20d63d8d564725b17ef9fe8c3c4082.png
[e98195ec4bcb405589ac6324a564367f.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/02/06/52903ea70cae40d68d5329648114859b.png
[1263f61500864a44990d63b1dd80cd7b.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/02/06/3f161a9d64ad4c79b2d81324c565671e.png
[662519aaf7d844c69f0fa584c8d2b098.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/02/06/31a09a66e6734e0aa4c5d9991941085a.png
[62d614f9cb894073a94aa84ea7ee112a.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/02/06/478cf22a69fb4abbaf8bb80948117d0d.png
[48bb45d89c074a03adfe86185d80a5de.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/02/06/78dbe48c275b4f68915ee12f1d62f680.png
[10de90b39860448e9f22638061adce56.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/02/06/c3dc325dbe2e4f17b558d630b4ddfa6a.png
[7f0132b9a0404c62959cd24684146aaa.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/02/06/414aea8b4728463e80a5b755a80de7ab.png
[ddbad89797c24d9b8e41c26490ebde2b.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/02/06/281a834ac02542d39b83c9a5fe9f0795.png
[65636c16c20042b49ae14a04cfe57dee.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/02/06/1fc063781e974060b24908939bc8624d.png
[2a01f31a658540e9baccb4b060600fb4.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/02/06/d360e16b74cc47c2a3fe55161ae57a00.png
[6e30303604f64502a75c4424f39ea634.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/02/06/8fdae8349ded439f8373e9a55c2d8bd6.png
[1a9aaa34b74b4019adf1d1a3c5abb018.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/02/06/3f6c4857e9e14ce5a6247015a7664971.png
[a0ae5da612d84bf0a08c383de5d8b77c.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/02/06/f21e0cf7e40d46dcb53ea44780f54ba9.png
[64e0ef1d861741b4b8023788c25967f5.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/02/06/5f27275fd7424041abc85d8f3c68ca12.png
[416d809993ce48ff88ecb9cc412441fe.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/02/06/7186b302c59a4fd68804737059a4bd3a.png
[0cd949e56d6b4ec297e542a461c743cc.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/02/06/4147848ff77e4eaab0c7daca61277ec6.png
[b310fe73a0884dac9b23d44af1eca2dc.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/02/06/180a9c1aa3004e3cb44f857e955970b3.png
[20a7f1b58dfb4660b75d7f021c157d57.png_pic_center]: https://img-blog.csdnimg.cn/20a7f1b58dfb4660b75d7f021c157d57.png#pic_center
[20a7f1b58dfb4660b75d7f021c157d57.png_pic_center 1]: https://image.dandelioncloud.cn/pgy_files/images/2024/02/06/a633bbff7a6243b79651fffb42873f5c.png