您现在的位置是：网站首页> 编程资料编程资料
Redis Stream类型的使用详解_Redis_
2023-05-27
613人已围观
简介 Redis Stream类型的使用详解_Redis_
一、背景
最近在看redis这方面的知识，发现在redis5中产生了一种新的数据类型Stream，它和kafka的设计有些类似，可以当作一个简单的消息队列来使用。
二、redis中Stream类型的特点
- 是可持久化的，可以保证数据不丢失。
- 支持消息的多播、分组消费。
- 支持消息的有序性。
三、Stream的结构
解释：
消费者组： Consumer Group,即使用 XGROUP CREATE 命令创建的，一个消费者组中可以存在多个消费者，这些消费者之间是竞争关系。
- 同一条消息，只能被这个消费者组中的某个消费者获取。
- 多个消费者之间是相互独立的，互不干扰。
消费者： Consumer 消费消息。
last_delivered_id： 这个id保证了在同一个消费者组中，一个消息只能被一个消费者获取。每当消费者组的某个消费者读取到了这个消息后，这个last_delivered_id的值会往后移动一位，保证消费者不会读取到重复的消息。
pending_ids：记录了消费者读取到的消息id列表，但是这些消息可能还没有处理，如果认为某个消息处理，需要调用ack命令。这样就确保了某个消息一定会被执行一次。
消息内容：是一个键值对的格式。
Stream 中 消息的 ID： 默认情况下，ID使用 * ，redis可以自动生成一个，格式为 时间戳-序列号，也可以自己指定，一般使用默认生成的即可，且后生成的id号要比之前生成的大。
四、Stream的命令
1、XADD 往Stream末尾添加消息
1、命令格式
xadd key [NOMKSTREAM] [MAXLEN|MINID [=|~] threshold [LIMIT count]] *|ID field value [field value ...]
2、举例
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、设置一个固定大小的Stream1、精确指定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.
2、XRANGE查看Stream中的消息
1、命令格式
xrange key start end [COUNT count]
2、准备数据
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的事务操作，获取到同一毫秒产生的多条数据，时间戳一样，序列号不一样
3、举例
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进行限制
3、XREVRANGE反向查看Stream中的消息
XREVRANGE key end start [COUNT count]
使用方式和XRANGE类似，略。
4、XDEL删除消息
1、命令格式
xdel key ID [ID ...]
2、准备数据
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>
3、举例
需求：往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并不会被删除。
5、XLEN查看Stream中元素的长度
1、命令格式
本站声明：
1、本站所有资源均来源于互联网，不保证100%完整、不提供任何技术支持；
2、本站所发布的文章以及附件仅限用于学习和研究目的;不得将用于商业或者非法用途；否则由此产生的法律后果，本站概不负责！
