扫码关注官方微信

扫码下载APP
返回顶部
首页 > 资讯 > 精选 >Redis中Stream类型怎么用
  • 204

0
分享到

Redis中Stream类型怎么用

2023-06-25 14:06:18 204人浏览 薄情痞子
摘要

这篇文章主要介绍Redis中Stream类型怎么用，文中介绍的非常详细，具有一定的参考价值，感兴趣的小伙伴们一定要看完！一、背景最近在看Redis这方面的知识，发现在redis5中产生了一种新的数据类型Stream，它和kafka的设计有些

这篇文章主要介绍Redis中Stream类型怎么用，文中介绍的非常详细，具有一定的参考价值，感兴趣的小伙伴们一定要看完！

一、背景

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

二、redis中Stream类型的特点

  • 是可持久化的，可以保证数据不丢失。

  • 支持消息的多播、分组消费。

  • 支持消息的有序性。

三、Stream的结构

Redis中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 ...]

Redis中Stream类型怎么用

2、举例

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

向流中增加一条数据,

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

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

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>

手动指定ID的值

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

设置一个固定大小的Stream1、精确指定Stream的大小

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

Redis中Stream类型怎么用

模糊指定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) 011) "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]

Redis中Stream类型怎么用

2、准备数据
127.0.0.1:6379> multioK127.0.0.1:6379(TX)> xadd stream-key * username zhangsanQUEUED127.0.0.1:6379(TX)> xadd stream-key * username lisiQUEUED127.0.0.1:6379(TX)> exec1) "1636003481706-0"2) "1636003481706-1"127.0.0.1:6379> xadd stream-key * username wangwu"1636003499055-0"127.0.0.1:6379>

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

3、举例

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

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的值

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

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

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

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

(:表示开区间

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

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>

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

获取单条数据

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

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

获取固定条数的数据

127.0.0.1:6379> xrange stream-key - + count 11) 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、命令格式
xlen key
2、举例

查看Stream中元素的长度

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

注意：

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

6、XTRIM对Stream中的元素进行修剪

1、命令格式
xtrim key MAXLEN|MINID [=|~] threshold [LIMIT count]
2、准备数据
127.0.0.1:6379>  xadd stream-key * username zhangsan"1636009745401-0"127.0.0.1:6379> multiOK127.0.0.1:6379(TX)> xadd stream-key * username lisiQUEUED127.0.0.1:6379(TX)> xadd stream-key * username wangwuQUEUED127.0.0.1:6379(TX)> exec1) "1636009763955-0"2) "1636009763955-1"127.0.0.1:6379> xadd stream-key * username zhaoliu"1636009769625-0"127.0.0.1:6379>
3、举例

maxlen精确限制

127.0.0.1:6379> xtrim stream-key maxlen 2 # 保留最后的2个消息(integer) 2127.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个消息。

minid模糊限制

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

7、XREAD独立消费消息

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

1、命令格式
xread [COUNT count] [BLOCK milliseconds] STREAMS key [key ...] ID [ID ...]

Redis中Stream类型怎么用

2、准备数据
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>
3、举例

获取用户名是wangwu的数据

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

获取2条数据

127.0.0.1:6379> xread count 2 streams stream-key 0-01) 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限制单次读取最后的消息，因为当前读取可能没有这么多。

非阻塞读取Stream对尾的数据

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

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

阻塞读取Stream对尾的数据

Redis中Stream类型怎么用

注意：

  • $表示读取队列最新进来的一个消息，不是Stream的最后一个消息。是xread block执行后，再次使用xadd添加消息后，xread block才会返回。

  • block 0表示永久阻塞，当消息到来时，才接触阻塞。block 1000表示阻塞1000ms，如果1000ms还没有消息到来，则返回nil

  • xread进行顺序消费 当使用xread进行顺序消息时，需要记住返回的消息id，同时下次调用xread时，需要将上次返回的消息id传递进去。

  • xread读取消息，完全无视消费组，此时Stream就可以理解为一个普通的list。

8、消费者组相关操作

1、消费者组命令

Redis中Stream类型怎么用

2、准备数据

创建Stream的名称是 stream-key

创建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"
3、创建消费者组

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

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

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

xgroup create stream-key g2 $

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

4、创建一个从某个消息之后消费的消费者组
xgroup create stream-key g3 1636362619125-0  #1636362619125-0 这个是上方aa消息的id的值

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

从消费者中读取消息

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>

读取消费者的pending消息

127.0.0.1:6379> xgroup create stream-key g4 0-0OK127.0.0.1:6379> xinfo consumers stream-key g11) 1) "name"   2) "c1"   3) "pending"   4) (integer) 2   5) "idle"   6) (integer) 88792127.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-01) 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-01) 1) "stream-key"   2) (empty array)127.0.0.1:6379>

Redis中Stream类型怎么用

转移消费者的消息

127.0.0.1:6379> xpending stream-key g1 - + 10 c11) 1) "1636362619125-0"   2) "c1"   3) (integer) 2686183   4) (integer) 12) 1) "1636362623191-0"   2) "c1"   3) (integer) 102274   4) (integer) 7127.0.0.1:6379> xpending stream-key g1 - + 10 c2(empty array)127.0.0.1:6379> xclaim stream-key g1 c2 102274 1636362623191-01) 1) "1636362623191-0"   2) 1) "bb"      2) "bb"127.0.0.1:6379> xpending stream-key g1 - + 10 c21) 1) "1636362623191-0"   2) "c2"   3) (integer) 17616   4) (integer) 8127.0.0.1:6379>

Redis中Stream类型怎么用

也可以通过xautoclaim来实现。

6、一些监控命令

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

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

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

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

查看消费组信息

127.0.0.1:6379> xinfo groups stream-key1) 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"......

查看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) 411) "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>

以上是“Redis中Stream类型怎么用”这篇文章的所有内容，感谢各位的阅读！希望分享的内容对大家有帮助，更多相关知识，欢迎关注编程网精选频道！

--结束END--

本文标题: Redis中Stream类型怎么用

本文链接: https://lsjlt.com/news/305258.html(转载时请注明来源链接)

有问题或投稿请发送至: 邮箱/279061341@qq.com    QQ/279061341

猜你喜欢
  • Redis中Stream类型怎么用
    这篇文章主要介绍Redis中Stream类型怎么用，文中介绍的非常详细，具有一定的参考价值，感兴趣的小伙伴们一定要看完！一、背景最近在看redis这方面的知识，发现在redis5中产生了一种新的数据类型Stream，它和kafka的设计有些...
    99+
    2023-06-25
  • Redis Stream类型的使用详解
    目录一、背景二、redis中Stream类型的特点三、Stream的结构四、Stream的命令1、XADD 往Stream末尾添加消息1、命令格式2、举例2、XRANGE查看Stre...
    99+
    2024-04-02
  • Redis特殊数据类型之stream怎么应用
    这篇文章主要介绍“Redis特殊数据类型之stream怎么应用”，在日常操作中，相信很多人在Redis特殊数据类型之stream怎么应用问题上存在疑惑，小编查阅了各式资料，整理出简单好用的操作方法，希望对大...
    99+
    2024-04-02
  • Redis入门 - 数据类型：Stream详解
    Redis5.0 中还增加了一个数据类型Stream，它借鉴了Kafka的设计，是一个新的强大的支持多播的可持久化的消息队列。@pdai Redis入门 - 数据类型：Stream详解 为什么会设计Stream Stream详解 ...
    99+
    2017-07-13
    Redis入门 - 数据类型：Stream详解
  • Redis中怎么使用hash类型
    Redis中怎么使用hash类型，很多新手对此不是很清楚，为了帮助大家解决这个难题，下面小编将为大家详细讲解，有这方面需求的人可以来学习下，希望你能有所收获。 hash类型特别适...
    99+
    2024-04-02
  • redis中的hash数据类型怎么用
    这篇文章主要介绍redis中的hash数据类型怎么用，文中介绍的非常详细，具有一定的参考价值，感兴趣的小伙伴们一定要看完！1. hash类型数据概述我们先来看这个例子在上一节我们一起了解了 string 存...
    99+
    2024-04-02
  • node中的stream有哪些类型
    本篇内容介绍了“node中的stream有哪些类型”的有关知识，在实际案例的操作过程中，不少人都会遇到这样的困境，接下来就让小编带领大家学习一下如何处理这些情况吧！希望大家仔细阅读，能够学有所成！ ...
    99+
    2024-04-02
  • redis中Hash类型有什么用
    这篇文章将为大家详细讲解有关redis中Hash类型有什么用，小编觉得挺实用的，因此分享给大家做个参考，希望大家阅读完这篇文章后可以有所收获。Hash类型是String类型的field和value映射表，或...
    99+
    2024-04-02
  • Redis中的5种数据类型怎么应用
    本文小编为大家详细介绍“Redis中的5种数据类型怎么应用”，内容详细，步骤清晰，细节处理妥当，希望这篇“Redis中的5种数据类型怎么应用”文章能帮助大家解决疑惑，下面跟着小编的思路慢慢深入，一起来学习新...
    99+
    2024-04-02
  • Redis中Hash类型的使用
    目录Hash与String比较Hash类型的常见命令（显而易见，在String命令前加入H就是Hash的命令） 1.HSET key field  value：添加或修改hash类型key的field...
    99+
    2023-06-09
    Redis Hash类型 Redis Hash
  • Redis中怎么实现字符串类型
    Redis中怎么实现字符串类型，相信很多没有经验的人对此束手无策，为此本文总结了问题出现的原因和解决方法，通过这篇文章希望你能解决这个问题。Redis是一个基于键-值存储的数据库。Redis中使用字符串作为...
    99+
    2024-04-02
  • Redis数据类型string和Hash怎么用
    这篇文章给大家分享的是有关Redis数据类型string和Hash怎么用的内容。小编觉得挺实用的，因此分享给大家做个参考，一起跟随小编过来看看吧。String类型命令操作String类型是在Redis中最常用的数据类型，它也是很多程序员在日...
    99+
    2023-06-29
  • Redis中如何使用set类型
    本篇文章给大家分享的是有关Redis中如何使用set类型，小编觉得挺实用的，因此分享给大家学习，希望大家阅读完这篇文章后可以有所收获，话不多说，跟着小编一起来看看吧。 set是s...
    99+
    2024-04-02
  • Redis中的Zset是什么类型
    Zset是Redis中的有序集合（Sorted Set）类型，它是通过将元素与分数（score）关联在一起来实现的，元素在有序集合中...
    99+
    2024-04-09
    Redis
  • Node.js中Stream怎么用
    这篇文章主要为大家展示了“Node.js中Stream怎么用”，内容简而易懂，条理清晰，希望能够帮助大家解决疑惑，下面让小编带领大家一起研究并学习一下“Node.js中Stream怎么用”这篇文章吧。一、什...
    99+
    2024-04-02
  • redis怎么更改数据类型
    在Redis中，可以使用以下命令更改数据类型：1. `SET key value`：将键 key 的值设置为 value，如果键 k...
    99+
    2023-08-30
    redis
  • redis中hash数据类型
    remoteSelf:1>hset website google "www.google.com" "1" remoteSelf:1>hget website "ERR wrong number of arguments for ...
    99+
    2021-09-09
    redis中hash数据类型
  • Redis特殊数据类型Geospatial怎么使用
    这篇文章主要介绍“Redis特殊数据类型Geospatial怎么使用”的相关知识，小编通过实际案例向大家展示操作过程，操作方法简单快捷，实用性强，希望这篇“Redis特殊数据类型Geospatial怎么使用”文章能帮助大家解决问题。Redi...
    99+
    2023-06-30
  • java中Stream转成List类型如下写代码？
    在Java中，将Stream转换为List类型可以使用Collectors.toList()方法。以下是示例代码： javaCopy code import java.util.List; import java.util.stream....
    99+
    2023-09-17
    java list python
  • 怎么删除redis的set类型key
    要删除Redis的set类型key，可以使用以下命令：1. 使用DEL命令删除key：   ```   DEL ...
    99+
    2023-08-30
    redis
软考高级职称资格查询
热门wiki
近期文章
推荐阅读
热门问答
热门标签
编程网，编程工程师的家园，是目前国内优秀的开源技术社区之一，形成了由开源软件库、代码分享、资讯、协作翻译、讨论区和博客等几大频道内容，为IT开发者提供了一个发现、使用、并交流开源技术的平台。

  • 官方手机版

  • 微信公众号

  • 商务合作