Redis之stream类型解读

叁歲伎倆 2023-10-15 08:36 134阅读 0赞

**目录**

基本介绍

数据结构

消息

消费组

消费者

基本使用命令

概述

xadd 命令

xtrim 命令

xdel 命令

xlen 命令

xrange 命令

xread 命令

xgroup 命令

xreadgroup 命令

xack 命令

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

#### 基本介绍 ####

Redis stream（流）是一种数据结构，其作用类似于仅追加日志，但也实现了多个操作来克服典型仅追加日志的一些限制。其中包括O（1）时间的随机访问和复杂的消费策略，如消费者群体。 您可以使用流实时记录和同时联合事件。

Redis 为每个stream（流）条目生成一个唯一的 ID。可以在以后使用这些 ID 检索其关联的条目，或读取和处理流中的所有后续条目。

> Redis Stream 主要用于消息队列（MQ，Message Queue），Redis 本身是有一个 Redis 发布订阅 (pub/sub) 来实现消息队列的功能，但它有个缺点就是消息无法持久化，如果出现网络断开、Redis 宕机等，消息就会被丢弃。Redis Stream 提供了消息的持久化和主备复制功能，可以让任何客户端访问任何时刻的数据，并且能记住每一个客户端的访问位置，还能保证消息不丢失。stream 有一个消息链表，将所有加入的消息都串起来，每个消息都有一个唯一的 ID 和对应的内容。消息是持久化的，Redis 重启后，内容还在。

#### 数据结构 ####

Redis Stream 的结构如下所示，它有一个消息链表，将所有加入的消息都串起来，每个消息都有一个唯一的 ID 和对应的内容：

![d032155a4a0c4676937672e83147cebe.png][]

##### **消息** #####

每个 Stream 都有唯一的名称，它就是 Redis 的 key，在我们首次使用 xadd 指令追加消息时自动创建。`XADD key ID field value[field value ...]`

*  消息 ID：消息 ID 的形式是 timestampInMillis-sequence，例如 1527846880572-5，它表示当前的消息在毫米时间戳 1527846880572 时产生，并且是该毫秒内产生的第5条消息。消息 ID 可以由服务器自动生成，也可以由客户端自己指定，但是形式必须是整数-整数，而且必须是后面加入的消息的 ID 要大于前面的消息 ID。
 *  消息内容：消息内容就是键值对，形如 hash 结构的键值对。

##### **消费组** #####

每个 Stream 都可以挂多个消费组（Consumer Group），每个消费组会有个游标*last\_delivered\_id*在 Stream 数组之上往前移动，表示当前消费组已经消费到哪条消息了。每个消费组都有一个 Stream 内唯一的名称，消费组不会自动创建，它需要单独的指令`xgroup create`进行创建，需要指定从 Stream 的某个消息*ID*开始消费，这个 ID 用来初始化 last\_delivered\_id 变量。每个消费组的状态都是独立的，相互不受影响。也就是说同一份 Stream 内部的消息会被每个消费组都消费到。同一个消费组可以挂接多个消费者（Consumer），这些消费者之间是竞争关系，任意一个消费者读取了消息都会使游标 last\_delivered\_id 往前移动。每个消费者有一个组内唯一名称。

##### **消费者** #####

消费者内部会有一个状态变量*pending\_ids*，维护消费者的未确认的 id。 pending\_ids 记录了当前已经被客户端读取的消息，但是还没有*ack*(Acknowledge character：确认字符）。如果客户端没有 ack，这个变量里面的消息 ID 就会越来越多，一旦某个消息被 ack，它就开始减少。这个 pending\_ids 变量在 Redis 官方被称为 PEL，也就是 Pending Entries List，这是一个核心的数据结构，它用来确保客户端至少消费了消息一次，而不会在网络传输的中途丢失了而没被处理。

#### 基本使用命令 ####

##### 概述 #####

**消息队列相关命令：**

*  **XADD** \- 添加消息到末尾
 *  **XTRIM** \- 对流进行修剪，限制长度
 *  **XDEL** \- 删除消息
 *  **XLEN** \- 获取流包含的元素数量，即消息长度
 *  **XRANGE** \- 获取消息列表，会自动过滤已经删除的消息
 *  **XREVRANGE** \- 反向获取消息列表，ID 从大到小
 *  **XREAD** \- 以阻塞或非阻塞方式获取消息列表

**消费者组相关命令：**

*  **XGROUP CREATE** \- 创建消费者组
 *  **XREADGROUP GROUP** \- 读取消费者组中的消息
 *  **XACK** \- 将消息标记为"已处理"
 *  **XGROUP SETID** \- 为消费者组设置新的最后递送消息ID
 *  **XGROUP DELCONSUMER** \- 删除消费者
 *  **XGROUP DESTROY** \- 删除消费者组
 *  **XPENDING** \- 显示待处理消息的相关信息
 *  **XCLAIM** \- 转移消息的归属权
 *  **XINFO** \- 查看流和消费者组的相关信息；
 *  **XINFO GROUPS** \- 打印消费者组的信息；
 *  **XINFO STREAM** \- 打印流信息

##### xadd 命令 #####

XADD 命令将指定的流条目追加到指定 key 的流中。如果 key 不存在，将使用流的条目自动创建 key。

> 一个条目是由一组键值对组成的，它基本上是一个小的字典。键值对以用户给定的顺序存储，并且读取流的命令（如 XRANGE 或者 XREAD）可以保证按照通过 XADD 添加的顺序返回。

XADD key ID field value [field value ...]

*  **key** ：队列名称，如果不存在就创建
 *  **ID** ：消息 id，我们使用 \* 表示由 redis 生成，可以自定义，但是要自己保证递增性。
 *  **field value** ： 记录。

redis> XADD mystream * name Sara surname OConnor
    "1601372323627-0"
    redis> XADD mystream * field1 value1 field2 value2 field3 value3
    "1601372323627-1"
    redis> XLEN mystream
    (integer) 2
    redis> XRANGE mystream - +
    1) 1) "1601372323627-0"
       2) 1) "name"
          2) "Sara"
          3) "surname"
          4) "OConnor"
    2) 1) "1601372323627-1"
       2) 1) "field1"
          2) "value1"
          3) "field2"
          4) "value2"
          5) "field3"
          6) "value3"

> 返回值：该命令返回添加的条目的 ID。如果 ID 参数传的是\*，那么 ID 是自动生成的，否则，命令仅返回用户在插入期间指定的相同的 ID。

##### xtrim 命令 #####

XTRIM 将流裁剪为指定数量的项目，如有需要，将驱逐旧的项目（ID较小的项目）。

XTRIM key MAXLEN [~] count

*  **key** ：队列名称
 *  **MAXLEN** ：长度
 *  **count** ：数量

127.0.0.1:6379> XADD mystream * field1 A field2 B field3 C field4 D
    "1601372434568-0"
    127.0.0.1:6379> XTRIM mystream MAXLEN 2
    (integer) 0
    127.0.0.1:6379> XRANGE mystream - +
    1) 1) "1601372434568-0"
       2) 1) "field1"
          2) "A"
          3) "field2"
          4) "B"
          5) "field3"
          6) "C"
          7) "field4"
          8) "D"

> 返回值：返回从流中删除的条目数。

##### xdel 命令 #####

从指定流中移除指定的条目，并返回成功删除的条目的数量。在传递的ID不存在的情况下，返回的数量可能与传递的ID数量不同。

XDEL key ID[ID ...]

*  **key：队列名称。**
 *  **ID：消息 ID**

> XADD mystream * a 1
    1538561698944-0
    > XADD mystream * b 2
    1538561700640-0
    > XADD mystream * c 3
    1538561701744-0
    > XDEL mystream 1538561700640-0
    (integer) 1
    127.0.0.1:6379> XRANGE mystream - +
    1) 1) 1538561698944-0
       2) 1) "a"
          2) "1"
    2) 1) 1538561701744-0
       2) 1) "c"
          2) "3"

> 返回值：删除的条目数量。

##### xlen 命令 #####

返回流中的条目数。如果指定的key不存在，则此命令返回0，就好像该流为空。

> 与其他的Redis类型不同，零长度流是可能的，所以你应该调用`TYPE`或者`EXISTS`来检查一个key是否存在。一旦内部没有任何的条目（例如调用`XDEL`后），流不会被自动删除，因为可能还存在与其相关联的消费者组。

redis> XADD mystream * item 1
    "1601372563177-0"
    redis> XADD mystream * item 2
    "1601372563178-0"
    redis> XADD mystream * item 3
    "1601372563178-1"
    redis> XLEN mystream
    (integer) 3

> 返回值：流中包含的条目数量

##### xrange 命令 #####

xrange命令返回流中满足给定ID范围的条目。范围由最小和最大ID指定。所有ID在指定的两个ID之间或与其中一个ID相等（闭合区间）的条目将会被返回。

XRANGE key start end [COUNT count]

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

redis> XADD writers * name Virginia surname Woolf
    "1601372577811-0"
    redis> XADD writers * name Jane surname Austen
    "1601372577811-1"
    redis> XADD writers * name Toni surname Morrison
    "1601372577811-2"
    redis> XADD writers * name Agatha surname Christie
    "1601372577812-0"
    redis> XADD writers * name Ngozi surname Adichie
    "1601372577812-1"
    redis> XLEN writers
    (integer) 5
    redis> XRANGE writers - + COUNT 2
    1) 1) "1601372577811-0"
       2) 1) "name"
          2) "Virginia"
          3) "surname"
          4) "Woolf"
    2) 1) "1601372577811-1"
       2) 1) "name"
          2) "Jane"
          3) "surname"
          4) "Austen"

> 返回值：该命令返回ID与指定范围匹配的条目。返回的条目是完整的，这意味着ID和所有组成条目的字段都将返回。此外，返回的条目及其字段和值的顺序与使用`XADD`添加它们的顺序完全一致。

##### xread 命令 #####

从一个或者多个流中读取数据，仅返回ID大于调用者报告的最后接收ID的条目。此命令有一个阻塞选项，用于等待可用的项目，类似于`BRPOP`或者`BZPOPMIN`等等。

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

*  count：数量。
 *  milliseconds：可选，阻塞毫秒数，没有设置就是非阻塞模式。
 *  key ：队列名。
 *  id：消息 ID。

redis> XREAD COUNT 2 STREAMS mystream writers 0-0 0-0
    1) 1) "mystream"
       2) 1) 1) 1526984818136-0
             2) 1) "duration"
                2) "1532"
                3) "event-id"
                4) "5"
                5) "user-id"
                6) "7782813"
          2) 1) 1526999352406-0
             2) 1) "duration"
                2) "812"
                3) "event-id"
                4) "9"
                5) "user-id"
                6) "388234"
    2) 1) "writers"
       2) 1) 1) 1526985676425-0
             2) 1) "name"
                2) "Virginia"
                3) "surname"
                4) "Woolf"
          2) 1) 1526985685298-0
             2) 1) "name"
                2) "Jane"
                3) "surname"
                4) "Austen"

> 返回值：该命令返回一个结果数组：返回数组的每个元素都是一个由两个元素组成的数组（键名和为该键报告的条目）。报告的条目是完整的流条目，具有ID以及所有字段和值的列表。返回的条目及其字段和值的顺序与使用`XADD`添加它们的顺序完全一致。
> 
> 当使用**BLOCK**时，超时时将返回一个空回复（`nil`）。

##### xgroup 命令 #####

使用 XGROUP CREATE 创建消费者组，语法格式：

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

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

从头开始消费:

XGROUP CREATE mystream consumer-group-name 0-0

从尾部开始消费:

XGROUP CREATE mystream consumer-group-name $

##### xreadgroup 命令 #####

使用 XREADGROUP GROUP 读取消费组中的消息，语法格式：

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

*  **group** ：消费组名
 *  **consumer** ：消费者名。
 *  **count** ： 读取数量。
 *  **milliseconds** ： 阻塞毫秒数。
 *  **key** ： 队列名。
 *  **ID** ： 消息 ID。