Redis入门 - 数据类型：Stream类型详解

2022年7月13日

Redis
数据类型

Stream类型

大约 11 分钟

Redis入门 - 数据类型：Stream类型详解

背景

最近在看redis这方面的知识，发现在redis5中产生了一种新的数据类型Stream，它和kafka的设计有些类似，可以当作一个简单的消息队列来使用。

Stream类型特点

支持持久化。streams能持久化存储数据，不同于pub/sub机制和list 消息被消费后就会被删除，streams消费过的数据会被持久化的保存在历史中。
支持消息的多播、分组消费。这一点跟 pub/sub有些类似。
支持消息的有序性，可以通过消息的ID来控制消息的排序。
支持消费者组。streams 允许同一消费组内的消费者竞争消息，并提供了一系列机制允许消费者查看自己的历史消费消息。并允许监控streams的消费者组信息，消费者组内消费者信息，也可以监控streams内消息的状态。

Stream结构

解释：

消费者组：Consumer Group,即使用XGROUP CREATE命令创建的，一个消费者组中可以存在多个消费者，这些消费者之间是竞争关系。
- 同一条消息，只能被这个消费者组中的某个消费者获取。
- 多个消费者之间是相互独立的，互不干扰。
消费者： Consumer 消费消息。
last_delivered_id：这个id保证了在同一个消费者组中，一个消息只能被一个消费者获取。每当消费者组的某个消费者读取到了这个消息后，这个last_delivered_id的值会往后移动一位，保证消费者不会读取到重复的消息。
pending_ids：记录了消费者读取到的消息id列表，但是这些消息可能还没有处理，如果认为某个消息处理，需要调用ack命令。这样就确保了某个消息一定会被执行一次。
消息内容：是一个键值对的格式。
Stream 中消息的 ID：默认情况下，ID使用 * ，redis可以自动生成一个，格式为时间戳-序列号，也可以自己指定，一般使用默认生成的即可，且后生成的id号要比之前生成的大。

Stream命令

XADD 往Stream末尾添加消息

一、命令格式

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

二、举例

xadd 命令返回的是数据的id, xx-yy (xx指的是毫秒数,yy指的是在这个毫秒内的第几条消息)

1、向流中增加一条数据

127.0.0.1:6379> xadd stream-key * username zhangsan # 向stream-key这个流中增加一个 username 是zhangsan的数据 *表示自动生成id
"1635999858912-0" # 返回的是ID
127.0.0.1:6379> keys *
1) "stream-key" # 可以看到stream自动创建了
   127.0.0.1:6379>

2、向流中增加数据，不自动创建流

127.0.0.1:6379> xadd not-exists-stream nomkstream * username lisi # 因为指定了nomkstream参数，而not-exists-stream之前不存在，所以加入失败
(nil)
127.0.0.1:6379> keys *
(empty array)
127.0.0.1:6379>

3、手动指定ID的值

127.0.0.1:6379> xadd stream-key 1-1 username lisi # 此处id的值是自己传递的1-1,而不是使用*自动生成
"1-1" # 返回的是id的值
127.0.0.1:6379>

4、设置一个固定大小的Stream

1）精确指定Stream的大小

指定指定Stream的大小比模糊指定Stream的大小会稍微多少消耗一些性能。

2）模糊指定Stream的大小

127.0.0.1:6379> xadd stream-key maxlen ~ 1 * first first
"1636001034141-0"
127.0.0.1:6379> xadd stream-key maxlen ~ 1 * second second
"1636001044506-0"
127.0.0.1:6379> xadd stream-key maxlen ~ 1 * third third
"1636001057846-0"
127.0.0.1:6379> xinfo stream stream-key
 1) "length"
 2) (integer) 3
 3) "radix-tree-keys"
 4) (integer) 1
 5) "radix-tree-nodes"
 6) (integer) 2
 7) "last-generated-id"
 8) "1636001057846-0"
 9) "groups"
10) (integer) 0
11) "first-entry"
12) 1) "1636001034141-0"
    2) 1) "first"
       2) "first"
13) "last-entry"
14) 1) "1636001057846-0"
    2) 1) "third"
       2) "third"
127.0.0.1:6379>

模糊指定流的大小，可以看到指定的是1，实际上已经到了3.

XRANGE查看Stream中的消息

一、命令格式

xrange key start end [COUNT count]

二、准备数据

127.0.0.1:6379> multi
OK
127.0.0.1:6379(TX)> xadd stream-key * username zhangsan
QUEUED
127.0.0.1:6379(TX)> xadd stream-key * username lisi
QUEUED
127.0.0.1:6379(TX)> exec
1) "1636003481706-0"
2) "1636003481706-1"
   127.0.0.1:6379> xadd stream-key * username wangwu
   "1636003499055-0"
   127.0.0.1:6379>

使用redis的事务操作，获取到同一毫秒产生的多条数据，时间戳一样，序列号不一样。

三、举例

1、获取所有的数据(-和+的使用)

127.0.0.1:6379> xrange stream-key - +
1) 1) "1636003481706-0"
2) 1) "username"
2) "zhangsan"
2) 1) "1636003481706-1"
2) 1) "username"
2) "lisi"
3) 1) "1636003499055-0"
2) 1) "username"
2) "wangwu"
   127.0.0.1:6379>

-: 表示最小id的值
+:表示最大id的值

2、获取指定id范围内的数据，闭区间

127.0.0.1:6379> xrange stream-key 1636003481706-1 1636003499055-0
1) 1) "1636003481706-1"
2) 1) "username"
2) "lisi"
2) 1) "1636003499055-0"
2) 1) "username"
2) "wangwu"
   127.0.0.1:6379>

3、获取指定id范围内的数据，开区间

127.0.0.1:6379> xrange stream-key (1636003481706-0 (1636003499055-0
1) 1) "1636003481706-1"
2) 1) "username"
2) "lisi"
   127.0.0.1:6379>

(:表示开区间

4、获取某个毫秒后所有的数据

127.0.0.1:6379> xrange stream-key 1636003481706 +
1) 1) "1636003481706-0"
2) 1) "username"
2) "zhangsan"
2) 1) "1636003481706-1"
2) 1) "username"
2) "lisi"
3) 1) "1636003499055-0"
2) 1) "username"
2) "wangwu"
   127.0.0.1:6379>

直接写毫秒不写后面的序列号即可。

5、获取单条数据

127.0.0.1:6379> xrange stream-key 1636003499055-0 1636003499055-0
1) 1) "1636003499055-0"
2) 1) "username"
2) "wangwu"
   127.0.0.1:6379>

start和end的值写的一样即可获取单挑数据。

6、获取固定条数的数据

127.0.0.1:6379> xrange stream-key - + count 1
1) 1) "1636003481706-0"
2) 1) "username"
2) "zhangsan"
   127.0.0.1:6379>

使用count进行限制

XREVRANGE反向查看Stream中的消息

XREVRANGE key end start [COUNT count]

使用方式和XRANGE类似，略。

XDEL删除消息

一、命令格式

xdel key ID [ID ...]

二、准备数据

127.0.0.1:6379> xadd stream-key * username zhangsan
"1636004176924-0"
127.0.0.1:6379> xadd stream-key * username lisi
"1636004183638-0"
127.0.0.1:6379> xadd stream-key * username wangwu
"1636004189211-0"
127.0.0.1:6379>

三、举例

需求：往Stream中加入3条消息，然后删除第2条消息

127.0.0.1:6379> xdel stream-key 1636004183638-0
(integer) 1 # 返回的是删除记录的数量
127.0.0.1:6379> xrang stream -key - +
127.0.0.1:6379> xrange stream-key - +
1) 1) "1636004176924-0"
2) 1) "username"
2) "zhangsan"
2) 1) "1636004189211-0"
2) 1) "username"
2) "wangwu"
   127.0.0.1:6379>

注意：

需要注意的是，我们从Stream中删除一个消息，这个消息并不是被真正的删除了，而是被标记为删除，这个时候这个消息还是占据着内容空间的。如果所有Stream中所有的消息都被标记删除，这个时候才会回收内存空间。但是这个Stream并不会被删除。

XLEN查看Stream中元素的长度

一、命令格式

xlen key

二、举例

查看Stream中元素的长度

127.0.0.1:6379> xadd stream-key * username zhangsan
"1636004690578-0"
127.0.0.1:6379> xlen stream-key
(integer) 1
127.0.0.1:6379> xlen not-exists-stream-key
(integer) 0
127.0.0.1:6379>

注意：

如果xlen后方的key不存在则返回0，否则返回元素的个数。

XTRIM对Stream中的元素进行修剪

一、命令格式

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

二、准备数据

127.0.0.1:6379>  xadd stream-key * username zhangsan
"1636009745401-0"
127.0.0.1:6379> multi
OK
127.0.0.1:6379(TX)> xadd stream-key * username lisi
QUEUED
127.0.0.1:6379(TX)> xadd stream-key * username wangwu
QUEUED
127.0.0.1:6379(TX)> exec
1) "1636009763955-0"
2) "1636009763955-1"
   127.0.0.1:6379> xadd stream-key * username zhaoliu
   "1636009769625-0"
   127.0.0.1:6379>

三、举例

1、maxlen精确限制

127.0.0.1:6379> xtrim stream-key maxlen 2 # 保留最后的2个消息
(integer) 2
127.0.0.1:6379> xrange stream-key - + # 可以看到之前加入的2个消息被删除了
1) 1) "1636009763955-1"
2) 1) "username"
2) "wangwu"
2) 1) "1636009769625-0"
2) 1) "username"
2) "zhaoliu"
   127.0.0.1:6379>

上方的意思是，保留stream-key这个Stream中最后的2个消息。

2、minid模糊限制

minid 是删除比这个id小的数据，本地测试的时候没有测试出来，略。

XREAD独立消费消息

XREAD只是读取消息，读取完之后并不会删除消息。使用XREAD读取消息，是完全独立与消费者组的，多个客户端可以同时读取消息。

一、命令格式

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

二、准备数据

127.0.0.1:6379> xadd stream-key * username zhangsan
"1636011801365-0"
127.0.0.1:6379> xadd stream-key * username lisi
"1636011806261-0"
127.0.0.1:6379> xadd stream-key * username wangwu
"1636011810905-0"
127.0.0.1:6379>

三、举例

1、获取用户名是wangwu的数据

127.0.0.1:6379> xread streams stream-key 1636011806261-0 # 此处写的是lisi的id，即读取到的数据需要是 > 1636011806261-0
1) 1) "stream-key"
2) 1) 1) "1636011810905-0"
2) 1) "username"
2) "wangwu"

2、获取2条数据

127.0.0.1:6379> xread count 2 streams stream-key 0-0
1) 1) "stream-key"
2) 1) 1) "1636011801365-0"
2) 1) "username"
2) "zhangsan"
2) 1) "1636011806261-0"
2) 1) "username"
2) "lisi"
   127.0.0.1:6379>

count限制单次读取最后的消息，因为当前读取可能没有这么多。

3、非阻塞读取Stream对尾的数据

即读取队列尾的下一个消息，在非阻塞模式下始终是nil

127.0.0.1:6379> xread streams stream-key $
(nil)

4、阻塞读取Stream对尾的数据

注意：

$表示读取队列最新进来的一个消息，不是Stream的最后一个消息。是xread block执行后，再次使用xadd添加消息后，xread block才会返回。
block 0表示永久阻塞，当消息到来时，才接触阻塞。block 1000表示阻塞1000ms，如果1000ms还没有消息到来，则返回nil
xread进行顺序消费当使用xread进行顺序消息时，需要记住返回的消息id，同时下次调用xread时，需要将上次返回的消息id传递进去。
xread读取消息，完全无视消费组，此时Stream就可以理解为一个普通的list。

消费者组相关操作

一、消费者组命令

二、准备数据

1、创建Stream的名称是 stream-key

2、创建2个消息，aa和bb

127.0.0.1:6379> xadd stream-key * aa aa
"1636362619125-0"
127.0.0.1:6379> xadd stream-key * bb bb
"1636362623191-0"

三、创建消费者组

1、创建一个从头开始消费的消费者组

xgroup create stream-key(Stream 名) g1(消费者组名) 0-0(表示从头开始消费)

2、创建一个从Stream最新的一个消息消费的消费者组

xgroup create stream-key g2 $

$表示从最后一个元素消费，不包括Stream中的最后一个元素，即消费最新的消息。

四、创建一个从某个消息之后消费的消费者组

xgroup create stream-key g3 1636362619125-0  #1636362619125-0 这个是上方aa消息的id的值

1636362619125-0某个消息的具体的ID,这个g3消费者组中的消息都是大于>这个id的消息。

1、从消费者中读取消息

127.0.0.1:6379> xreadgroup group g1(消费组名) c1(消费者名，自动创建) count 3(读取3条) streams stream-key(Stream 名) >(从该消费者组中还未分配给另外的消费者的消息开始读取)
1) 1) "stream-key"
2) 1) 1) "1636362619125-0"
2) 1) "aa"
2) "aa"
2) 1) "1636362623191-0"
2) 1) "bb"
2) "bb"
   127.0.0.1:6379> xreadgroup group g2 c1 count 3 streams stream-key >
   (nil) # 返回 nil 是因为 g2消费组是从最新的一条信息开始读取(创建消费者组时使用了$)，需要在另外的窗口执行`xadd`命令，才可以再次读取到消息
   127.0.0.1:6379> xreadgroup group g3 c1 count 3 streams stream-key >  #只读取到一条消息是因为，在创建消费者组时，指定了aa消息的id，bb消息的id大于aa,所以读取出来了。
1) 1) "stream-key"
2) 1) 1) "1636362623191-0"
2) 1) "bb"
2) "bb"
   127.0.0.1:6379>

4、读取消费者的pending消息

127.0.0.1:6379> xgroup create stream-key g4 0-0
OK
127.0.0.1:6379> xinfo consumers stream-key g1
1) 1) "name"
2) "c1"
3) "pending"
4) (integer) 2
5) "idle"
6) (integer) 88792
   127.0.0.1:6379> xinfo consumers stream-key g4
   (empty array)
   127.0.0.1:6379> xreadgroup group g1 c1 count 1 streams stream-key 1636362619125-0
1) 1) "stream-key"
2) 1) 1) "1636362623191-0"
2) 1) "bb"
2) "bb"
   127.0.0.1:6379> xreadgroup group g4 c1 count 1 block 0 streams stream-key 1636362619125-0
1) 1) "stream-key"
2) (empty array)
   127.0.0.1:6379>

5、转移消费者的消息

127.0.0.1:6379> xpending stream-key g1 - + 10 c1
1) 1) "1636362619125-0"
2) "c1"
3) (integer) 2686183
4) (integer) 1
2) 1) "1636362623191-0"
2) "c1"
3) (integer) 102274
4) (integer) 7
   127.0.0.1:6379> xpending stream-key g1 - + 10 c2
   (empty array)
   127.0.0.1:6379> xclaim stream-key g1 c2 102274 1636362623191-0
1) 1) "1636362623191-0"
2) 1) "bb"
2) "bb"
   127.0.0.1:6379> xpending stream-key g1 - + 10 c2
1) 1) "1636362623191-0"
2) "c2"
3) (integer) 17616
4) (integer) 8
   127.0.0.1:6379>

也可以通过xautoclaim来实现。

一些监控命令

1、查看消费组中消费者的pending消息

127.0.0.1:6379> xpending stream-key g1 - + 10 c2
1) 1) "1636362623191-0"
2) "c2"
3) (integer) 1247680
4) (integer) 8
   127.0.0.1:6379>

2、查看消费组中的消费者信息

127.0.0.1:6379> xinfo consumers stream-key g1
1) 1) "name"
2) "c1"
3) "pending"
4) (integer) 1
5) "idle"
6) (integer) 1474864
2) 1) "name"
2) "c2"
3) "pending"
4) (integer) 1
5) "idle"
6) (integer) 1290069
   127.0.0.1:6379>

3、查看消费组信息

127.0.0.1:6379> xinfo groups stream-key
1) 1) "name"
2) "g1"
3) "consumers"
4) (integer) 2
5) "pending"
6) (integer) 2
7) "last-delivered-id"
8) "1636362623191-0"
2) 1) "name"
2) "g2"
3) "consumers"
   ......

4、查看Stream信息

127.0.0.1:6379> xinfo stream stream-key
1) "length"
2) (integer) 2
3) "radix-tree-keys"
4) (integer) 1
5) "radix-tree-nodes"
6) (integer) 2
7) "last-generated-id"
8) "1636362623191-0"
9) "groups"
10) (integer) 4
11) "first-entry"
12) 1) "1636362619125-0"
2) 1) "aa"
2) "aa"
13) "last-entry"
14) 1) "1636362623191-0"
2) 1) "bb"
2) "bb"
   127.0.0.1:6379>

引用资料

https://www.jb51.net/article/228778.htm

Redis入门 - 数据类型：Stream类型详解

# Redis入门 - 数据类型：Stream类型详解

# 背景

# Stream类型特点

# Stream结构

# Stream命令

# XADD 往Stream末尾添加消息

# 一、命令格式

# 二、举例

# XRANGE查看Stream中的消息

# 一、命令格式

# 二、准备数据

# 三、举例

# XREVRANGE反向查看Stream中的消息

# XDEL删除消息

# 一、命令格式

# 二、准备数据

# 三、举例

# XLEN查看Stream中元素的长度

# 一、命令格式

# 二、举例

# XTRIM对Stream中的元素进行修剪

# 一、命令格式

# 二、准备数据

# 三、举例

# XREAD独立消费消息

# 一、命令格式

# 二、准备数据

# 三、举例

# 消费者组相关操作

# 一、消费者组命令

# 二、准备数据

# 三、创建消费者组

# 四、创建一个从某个消息之后消费的消费者组

# 一些监控命令

# 引用资料

Redis入门 - 数据类型：Stream类型详解

背景

Stream类型特点

Stream结构

Stream命令

XADD 往Stream末尾添加消息

一、命令格式

二、举例

XRANGE查看Stream中的消息

一、命令格式

二、准备数据

三、举例

XREVRANGE反向查看Stream中的消息

XDEL删除消息

一、命令格式

二、准备数据

三、举例

XLEN查看Stream中元素的长度

一、命令格式

二、举例

XTRIM对Stream中的元素进行修剪

一、命令格式

二、准备数据

三、举例

XREAD独立消费消息

一、命令格式

二、准备数据

三、举例

消费者组相关操作

一、消费者组命令

二、准备数据

三、创建消费者组

四、创建一个从某个消息之后消费的消费者组

一些监控命令

引用资料