目录
- 1 基本特性
- 2 主要操作命令
- 2.1 XADD key ID field value [field value ...]
- 2.2 XREAD [COUNT count] [block milliseconds] STREAMS key [key ...] ID [ID ...]
- 2.3 XRANGE key start end [COUNT count]
- 2.4 XREVRANGE key end start [COUNT count]
- 2.5 XGROUP CREATE key groupname id-or-$ [MKSTREAM]
- 2.6 XREADGROUP GROUP group consumer [COUNT count] [BLOCK milliseconds] [NOACK] STREAMS key [key ...] ID [ID ...]
- 2.7 XACK key group ID [ID ...]
- 2.8 XPENDING key group [start end count] [IDLE idle]
- 2.9 XCLAIM key group consumer min-idle-time ID [ID ...] [IDLE idle] [TIME time] [RETries count] [FORCE]
- 2.10 XINFO
- 2.11 XDEL key ID [ID ...]
- 2.12 XTRIM key MAXLEN [~] len
- 3 使用场景
Redis Streams 是 Redis 5.0 引入的一种新的数据类型,它提供了一种强大的日志结构化数据存储方式。Streams 类型非常适合用于构建消息队列、事件日志以及其他需要持久化和高效处理时间序列数据的应用场景。
1 基本特性
- 持久性:与传统的发布/订阅不同,Streams 中的消息是持久化的,即使客户端断开连接后重新连接,仍然可以访问到之前的消息。
- 多消费者支持:支持多个消费者组(consumer groups),每个组可以独立地消费流中的消息。消费者组允许不同的消费者处理相同的消息,但每个消息在一个组内只能被一个消费者处理一次。
- 消息 ID 和范围查询:每条消息都有一个唯一的 ID,由时间戳和序列号组成。可以通过指定消息 ID 范围来获取特定时间段内的消息。
- 阻塞读取:支持阻塞读取(
XREAD
和XREADGROUP
命令的BLOCK
选项),使得客户端可以在没有新消息时等待一段时间。 - 自动删除:可以设置最大长度(
MAXLEN
选项)来限制流的大小,超过长度的消息会自动被删除。 - 灵活的消息格式:每条消息可以包含多个字段-值对,类似于哈希表,这使得消息可以携带丰富的信息。
2 主要操作命令
2.1 XADD key ID field value [field value ...]
向指定的流中添加一条新消息,ID 可以是 *(表示自动生成)或指定的时间戳和序列号。
127.0.0.1:6379> xadd mystream * sensor_id 123 temmperature 22.5
"1729306027171-0"
返回的结构可以分为两部分:
- 时间戳:
1729306027171 (
表示条目被添加的时间,单位是毫秒。你可以将这个时间戳转换为可读的日期和时间格式。)
- 序列号:
0 (
表示在同一毫秒内这是第一个条目。如果在同一毫秒内添加了多个条目,序列号将会递增,例如1729306027171-1
、1729306027171-2
等。)
2.2 XREAD [COUNT count] [BLOCK milliseconds] STREAMS key [key ...] ID [ID ...]
- 从一个或多个 Stream 中读取数据。
COUNT
指定返回的最大条目数。BLOCK
指定在没有新消息时阻塞的时间(毫秒)。STREAMS
指定要读取的 Stream 和起始 ID。
COUNT
指定返回的最大条目数。BLOCK
指定在没有新消息时阻塞的时间(毫秒)。STREAMS
指定要读取的 Stream 和起始 ID。
127.0.0.1:6379> xread count 2 streams mystream 0-0
1) 1) "mystream"
2) 1) 1) "1729306027171-0"
2) 1) "sensor_id"
2) "123"
3) "temmperature"
4) "22.5"
2.3 XRANGE key start end [COUNT count]
- 返回指定 ID 范围内的条目。
start
和end
是 ID,可以使用-
表示最小 ID,+
表示最大 ID。
127.0.0.1:6379> xrange mystream - +
1) 1) "1729306027171-0"
2) 1) "sensor_id"
2) "123"
3) "temmperature"
4) "22.5"
2.4 XREVRANGE key end start [COUNT count]
返回指定 ID 范围内的条目,但按逆序排列。
127.0.0.1:6379> xadd mystream * sensor_id 234 temmperature 23.5
"1729329067777-0"
127.0.0.1:6379> xadd mystream * sensor_id 345
"1729329079135-0"
127.0.0.1:6379> xrevrange mystream + - count 2
1) 1) "1729329079135-0"
2) 1) "sensor_id"
2) "345"
2) 1) "1729329067777-0"
2) 1) "sensor_id"
2) "234"
3) "temmperature"
4) "23.5"
2.5 XGROUP CREATE key groupname id-or-$ [MKSTREAM]
- 创建一个新的消费者组。
id-or-$
是起始位置,可以是具体的 ID 或$
表示只消费新的条目。MKSTREAM
如果 Stream 不存在则创建它。
127.0.0.1:6379> xgroup create mystream mygroup 0
OK
2.6 XREADGROUP GROUP group consumer [COUNT count] [BLOCK milliseconds] [NOACK] STREAMS key [key ...] ID [ID ...]
- 从消费者组中读取数据。
- GROUP:指定消费者组的名称。
- consumer:指定消费者的名称。
- COUNT count:可选参数,指定一次最多读取的消息数量。
- BLOCK milliseconds:可选参数,如果当前没有可用的消息,命令将阻塞指定的时间(以毫秒为单位),等待新消息的到来。
- NOACK: 表示不确认消息,通常用于快速消费。
- STREAMS:指定要读取的流及其对应的 ID。
- ID 通常是一个特殊值
>
,表示只读取新的消息;也可以是具体的 ID,表示从该 ID 开始读取。
127.0.0.1:6379> xreadgroup group mygroup consumer1 count 2 streams mystream >
1) 1) "mystream"
2) 1) 1) "1729306027171-0"
2) 1) "sensor_id"
2) "123"
3) "temmperature"
4) "22.5"
2) 1) "1729329067777-0"
2) 1) "sensor_id"
2) "234"
3) "temmperature"
4) "23.5"
2.7 XACK key group ID [ID ...]
确认已处理的消息。XACK
命令用于确认消费者组中的消息已经被成功处理。当你使用 XACK
命令时,Redis 会将指定的消息从“待处理”状态转换为“已确认”状态,并从消费者的待处理列表中移除。
127.0.0.1:6379> xack mystream mygroup 1729329067777-0
(integer) 1
127.0.0.1:6379> xack mystream mygroup 1729329079135-0
(integer) 0
当 XACK 命令成功确认一条消息时,返回值为 1,表示该消息已经被确认并且从待处理列表中移除。例如,如果消息 1729329067777-0 是由 consumer1 处理的,并且现在调用 XACK 确认它,那么这条消息将不再出现在 consumer1 的待处理列表中。
2.8 XPENDING key group [start end count] [IDLE idle]
查看待处理的消息。
127.0.0.1:6379> xpending mystream mygroup
1) (integer) 1
2) "1729306027171-0"
3) "1729306027171-0"
4) 1) 1) "consumer1"
2) "1"
127.0.0.1:6379> xack mystream mygroup 1729306027171-0
(integer) 1
127.0.0.1:6379> xpending mystream mygroup
1) (integer) 0
2) (nil)
3) (nil)
4) (nil)
2.9 XCLAIM key group consumer min-idle-time ID [ID ...] [IDLE idle] [TIME time] [RETries count] [FORCE]
用于将一个或多个消息从一个消费者转移到另一个消费者。这个命令通常用于处理消息超时或重新分配消息的情况。XCLAIM
允许你手动将消息从一个消费者的待处理列表移动到另一个消费者的待处理列表。
127.0.0.1:6379> xreadgroup group mygroup consumer1 count 2 streams mystream >
1) 1) "mystream"
2) 1) 1) "1729329079135-0"
2) 1) "sensor_id"
2) "345"
127.0.0.1:6379> xclaim mystream mygroup consumer2 10000 1729329079135-0
1) 1) "1729329079135-0"
2) 1) "sensor_id"
2) "345"
- mystream:流的名称。mygroup:消费者组的名称。
- consumer2:目标消费者的名称,即消息将被转移给这个消费者。
- 10000:消息的空闲时间(以毫秒为单位)。只有那些空闲时间超过这个值的消息才会被转移。
- 1729329079135-0:要转移的消息 ID。
2.10 XINFO
获取 Stream 或消费者组的信息。
127.0.0.1:6379> xinfo stream mystream
1) "length"
2) (integer) 3
3) "radix-tree-keys"
4) (integer) 1
5) "radix-tree-nodes"
6) (integer) 2
7) "groups"
8) (integer) 1
9) "last-generated-id"
10) "1729329079135-0"
11) "first-entry"
12) 1) "1729306027171-0"
2) 1) "sensor_id"
2) "123"
3) "temmperature"
4) "22.5"
13) "last-entry"
14) 1) "1729329079135-0"
2) 1) "sensor_id"
2) "345"
127.0.0.1:6379> xinfo groups mystream
1) 1) "name"
2) "mygroup"
3) "consumers"
4) (integer) 2
5) "pending"
6) (integer) 1
7) "last-delivered-id"
8) "1729329079135-0"
127.0.0.1:6379> xinfo consumers mystream mygroup
1) 1) "name"
2) "consumer1"
3) "pending"
4) (integer) 0
5) "idle"
6) (integer) 255317
2) 1) "name"
2) "consumer2"
3) "pending"
4) (integer) 1
5) "idle"
6) (integer) 191940
XINFO STREAM mystream
length
:
流中的消息总数:3 条。
radix-tree-keys
:
用于存储流数据的 radix tree 中的键的数量:1 个。
radix-tree-nodes
:
用于存储流数据的 radix tree 中的节点数量:2 个。
groups
:
与该流关联的消费者组数量:1 个。
last-generated-id
:
流中最后生成的消息 ID:1729329079135-0
。
first-entry
:
流中的第一条消息: 消息 ID: 1729306027171-0
消息内容: sensor_id
: 123
temmperature
: 22.5
last-entry
:
流中的最后一条消息: 消息 ID: 1729329079135-0
消息内容: sensor_id
: 345
XINFO GROUPS mystream
name
:
消费者组的名称:mygroup
。
consumers
:
该组中的消费者数量:2 个。
pending
:
该组中待处理的消息数量:1 条。
last-delivered-id
:
该组中最后一个被交付的消息 ID:1729329079135-0
。
XINFO CONSUMERS mystream mygroup
第一个消费者:
第一个消费者:
name
:consumer1
pending
: 待处理的消息数量:0 条idle
: 空闲时间(以毫秒为单位):255,317 毫秒(约 4 分钟 15 秒)
第二个消费者:
name
:consumer2
pending
: 待处理的消息数量:1 条idle
: 空闲时间(以毫秒为单位):191,940 毫秒(约 3 分钟 12 秒)
2.11 XDEL key ID [ID ...]
从 Stream 中删除一个或多个条目。
127.0.0.1:6379> xdel mystream 1729306027171-0
(integer) 1
127.0.0.1:6379> xrange mystrea - +
(empty list or set)
127.0.0.1:6379> xrange mystream - +
1) 1) "1729329067777-0"
2) 1) "sensor_id"
2) "234"
3) "temmperature"
4) "23.5"
2) 1) "1729329079135-0"
2) 1) "sensor_id"
2) "345"
2.12 XTRIM key MAXLEN [~] len
修剪 Stream,保留最多 len 个条目,~ 表示近似长度。
127.0.0.1:6379> xrange mystream - +
1) 1) "1729329067777-0"
2) 1) "sensor_id"
2) "234"
3) "temmperature"
4) "23.5"
2) 1) "1729329079135-0"
2) 1) "sensor_id"
2) "345"
127.0.0.1:6379> xtrim mystream maxlen 1
(integer) 1
127.0.0.1:6379> xrange mystream - +
1) 1) "1729329079135-0"
2) 1) "sensor_id"
2) "345"
3 使用场景
- 日志记录:可以用来存储系统的日志信息,方便后续分析和处理。
- 事件流:处理实时事件,如传感器数据、用户行为等。
- 消息队列:实现可靠的消息传递系统,支持多个消费者组。
- 任务队列:管理后台任务,确保任务被正确处理。
更多命令请参考:Commands | Docs
到此这篇关于Redis 数据类型Streams的文章就介绍到这了,更多相关Redis 数据类型Streams内容请搜索编程网(www.lsjlt.com)以前的文章或继续浏览下面的相关文章希望大家以后多多支持编程客栈(www.lsjlt.com)!
免责声明:
① 本站未注明“稿件来源”的信息均来自网络整理。其文字、图片和音视频稿件的所属权归原作者所有。本站收集整理出于非商业性的教育和科研之目的,并不意味着本站赞同其观点或证实其内容的真实性。仅作为临时的测试数据,供内部测试之用。本站并未授权任何人以任何方式主动获取本站任何信息。
② 本站未注明“稿件来源”的临时测试数据将在测试完成后最终做删除处理。有问题或投稿请发送至: 邮箱/279061341@qq.com QQ/279061341
软考中级精品资料免费领
- 历年真题答案解析
- 备考技巧名师总结
- 高频考点精准押题
- 资料下载
- 历年真题
193.9 KB下载数265
191.63 KB下载数245
143.91 KB下载数1148
183.71 KB下载数642
644.84 KB下载数2756
相关文章
发现更多好内容- Java 性能调优的有效手段有哪些?(Java性能调优有何有效手段)
- 探讨Uncomtrade数据库的安全防护措施
- Java 如何获取某个接口的实现类?(Java怎么获取某个接口的实现类)
- Java中如何正确使用 collections.shuffle 方法?(Java中collections.shuffle怎么使用)
- Java Swing 中常用的布局有哪些?(java swing常用布局有哪些)
- 如何通过 Java Reflection 获取泛型信息?(java reflection如何获取泛型信息)
- 如何自定义 Java 泛型通配符?(java泛型通配符怎么自定义)
- Java Spring 注解与 XML 配置的差异究竟有哪些?(java spring注解与XML配置的区别是什么)
- Java 动态线程池对性能究竟有哪些影响呢?(Java动态线程池对性能的影响)
- 在 Java 中,Guava 究竟有哪些作用呢?(java中guava的作用是什么)