Redis 数据结构简介
Redis 基础文章非常多，关于基础数据结构类型，我推荐你先看下官方网站内容，然后再看下面的小结。文章源自菜鸟学院-https://www.cainiaoxueyuan.com/sjk/45498.html
首先对 redis 来说，所有的 key（键）都是字符串。我们在谈基础数据结构时，讨论的是存储值的数据类型，主要包括常见的5种数据类型，分别是：String、List、Set、Zset、Hash。
文章源自菜鸟学院-https://www.cainiaoxueyuan.com/sjk/45498.html
5 种基础数据类型
内容其实比较简单，我觉得理解的重点在于这个结构怎么用，能够用来做什么？所以我在梳理时，围绕图例，命令，执行和场景来阐述。文章源自菜鸟学院-https://www.cainiaoxueyuan.com/sjk/45498.html
String 字符串
String是redis中最基本的数据类型，一个key对应一个value。文章源自菜鸟学院-https://www.cainiaoxueyuan.com/sjk/45498.html
String类型是二进制安全的，意思是 redis 的 string 可以包含任何数据。如数字，字符串，jpg图片或者序列化的对象。文章源自菜鸟学院-https://www.cainiaoxueyuan.com/sjk/45498.html
图例
下图是一个String类型的实例，其中键为hello，值为world文章源自菜鸟学院-https://www.cainiaoxueyuan.com/sjk/45498.html
命令使用
命令执行
127.0.0.1:6379> set hello world
OK
127.0.0.1:6379> get hello
"world"
127.0.0.1:6379> del hello
(integer) 1
127.0.0.1:6379> get hello
(nil)
127.0.0.1:6379> set counter 2
OK
127.0.0.1:6379> get counter
"2"
127.0.0.1:6379> incr counter
(integer) 3
127.0.0.1:6379> get counter
"3"
127.0.0.1:6379> incrby counter 100
(integer) 103
127.0.0.1:6379> get counter
"103"
127.0.0.1:6379> decr counter
(integer) 102
127.0.0.1:6379> get counter
"102"
实战场景
缓存：经典使用场景，把常用信息，字符串，图片或者视频等信息放到redis中，redis 作为缓存层，mysql做持久化层，降低mysql的读写压力。 计数器：redis是单线程模型，一个命令执行完才会执行下一个，同时数据可以一步落地到其他的数据源。 session：常见方案spring session + redis实现session共享，
List列表
Redis 中的 List 其实就是链表（Redis 用双端链表实现 List ）。文章源自菜鸟学院-https://www.cainiaoxueyuan.com/sjk/45498.html
使用 List 结构，我们可以轻松地实现最新消息排队功能（比如新浪微博的TimeLine）。List 的另一个应用就是消息队列，可以利用List的 PUSH 操作，将任务存放在 List 中，然后工作线程再用 POP 操作将任务取出进行执行。文章源自菜鸟学院-https://www.cainiaoxueyuan.com/sjk/45498.html
图例
命令使用
使用列表的技巧 lpush+lpop=Stack(栈) lpush+rpop=Queue（队列） lpush+ltrim=Capped Collection（有限集合） lpush+brpop=Message Queue（消息队列）
命令执行
127.0.0.1:6379> lpush mylist 1 2 ll ls mem
(integer) 5
127.0.0.1:6379> lrange mylist 0 -1
1) "mem"
2) "ls"
3) "ll"
4) "2"
5) "1"
127.0.0.1:6379> lindex mylist -1
"1"
127.0.0.1:6379> lindex mylist 10 # index不在 mylist 的区间范围内
(nil)
实战场景
微博TimeLine: 有人发布微博，用lpush加入时间轴，展示新的列表信息。 消息队列
Set集合
Redis 的 Set 是 String 类型的无序集合。集合成员是唯一的，这就意味着集合中不能出现重复的数据。文章源自菜鸟学院-https://www.cainiaoxueyuan.com/sjk/45498.html
Redis 中集合是通过哈希表实现的，所以添加，删除，查找的复杂度都是 O(1)。文章源自菜鸟学院-https://www.cainiaoxueyuan.com/sjk/45498.html
图例
命令使用
命令执行
127.0.0.1:6379> sadd myset hao hao1 xiaohao hao
(integer) 3
127.0.0.1:6379> smembers myset
1) "xiaohao"
2) "hao1"
3) "hao"
127.0.0.1:6379> sismember myset hao
(integer) 1
实战场景
标签（tag）,给用户添加标签，或者用户给消息添加标签，这样有同一标签或者类似标签的可以给推荐关注的事或者关注的人。 点赞，或点踩，收藏等，可以放到set中实现
Hash散列
Redis hash 是一个 string 类型的 field（字段） 和 value（值） 的映射表，hash 特别适合用于存储对象。文章源自菜鸟学院-https://www.cainiaoxueyuan.com/sjk/45498.html
图例
命令使用
命令执行
127.0.0.1:6379> hset user name1 hao
(integer) 1
127.0.0.1:6379> hset user email1 hao@163.com
(integer) 1
127.0.0.1:6379> hgetall user
1) "name1"
2) "hao"
3) "email1"
4) "hao@163.com"
127.0.0.1:6379> hget user user
(nil)
127.0.0.1:6379> hget user name1
"hao"
127.0.0.1:6379> hset user name2 xiaohao
(integer) 1
127.0.0.1:6379> hset user email2 xiaohao@163.com
(integer) 1
127.0.0.1:6379> hgetall user
1) "name1"
2) "hao"
3) "email1"
4) "hao@163.com"
5) "name2"
6) "xiaohao"
7) "email2"
8) "xiaohao@163.com"
实战场景
缓存：能直观，相比string更节省空间，的维护缓存信息，如用户信息，视频信息等。
Zset有序集合
Redis 有序集合和集合一样也是 string 类型元素的集合,且不允许重复的成员。不同的是每个元素都会关联一个 double 类型的分数。redis 正是通过分数来为集合中的成员进行从小到大的排序。文章源自菜鸟学院-https://www.cainiaoxueyuan.com/sjk/45498.html
有序集合的成员是唯一的, 但分数(score)却可以重复。有序集合是通过两种数据结构实现：文章源自菜鸟学院-https://www.cainiaoxueyuan.com/sjk/45498.html
压缩列表(ziplist): ziplist是为了提高存储效率而设计的一种特殊编码的双向链表。它可以存储字符串或者整数，存储整数时是采用整数的二进制而不是字符串形式存储。它能在O(1)的时间复杂度下完成list两端的push和pop操作。但是因为每次操作都需要重新分配ziplist的内存，所以实际复杂度和ziplist的内存使用量相关 跳跃表（zSkiplist): 跳跃表的性能可以保证在查找，删除，添加等操作的时候在对数期望时间内完成，这个性能是可以和平衡树来相比较的，而且在实现方面比平衡树要优雅，这是采用跳跃表的主要原因。跳跃表的复杂度是O(log(n))。
图例
命令使用
命令执行
127.0.0.1:6379> zadd myscoreset 100 hao 90 xiaohao
(integer) 2
127.0.0.1:6379> ZRANGE myscoreset 0 -1
1) "xiaohao"
2) "hao"
127.0.0.1:6379> ZSCORE myscoreset hao
"100"
实战场景
排行榜：有序集合经典使用场景。例如小说视频等网站需要对用户上传的小说视频做排行榜，榜单可以按照用户关注数，更新时间，字数等打分，做排行。
3 种特殊类型
Redis 除了上文中 5 种基础数据类型，还有 3 种特殊的数据类型，分别是 HyperLogLogs（基数统计）， Bitmaps (位图) 和 geospatial （地理位置）。文章源自菜鸟学院-https://www.cainiaoxueyuan.com/sjk/45498.html
HyperLogLogs（基数统计）
Redis 2.8.9 版本更新了 Hyperloglog 数据结构！文章源自菜鸟学院-https://www.cainiaoxueyuan.com/sjk/45498.html
什么是基数？
举个例子，A = {1, 2, 3, 4, 5}， B = {3, 5, 6, 7, 9}；那么基数（不重复的元素）= 1, 2, 4, 6, 7, 9；（允许容错，即可以接受一定误差）文章源自菜鸟学院-https://www.cainiaoxueyuan.com/sjk/45498.html
HyperLogLogs 基数统计用来解决什么问题？
这个结构可以非常省内存的去统计各种计数，比如注册 IP 数、每日访问 IP 数、页面实时UV、在线用户数，共同好友数等。文章源自菜鸟学院-https://www.cainiaoxueyuan.com/sjk/45498.html
它的优势体现在哪？
一个大型的网站，每天 IP 比如有 100 万，粗算一个 IP 消耗 15 字节，那么 100 万个 IP 就是 15M。而 HyperLogLog 在 Redis 中每个键占用的内容都是 12K，理论存储近似接近 2^64 个值，不管存储的内容是什么，它一个基于基数估算的算法，只能比较准确的估算出基数，可以使用少量固定的内存去存储并识别集合中的唯一元素。而且这个估算的基数并不一定准确，是一个带有 0.81% 标准错误的近似值（对于可以接受一定容错的业务场景，比如IP数统计，UV 等，是可以忽略不计的）。文章源自菜鸟学院-https://www.cainiaoxueyuan.com/sjk/45498.html
相关命令使用
127.0.0.1:6379> pfadd key1 a b c d e f g h i # 创建第一组元素
(integer) 1
127.0.0.1:6379> pfcount key1 # 统计元素的基数数量
(integer) 9
127.0.0.1:6379> pfadd key2 c j k l m e g a # 创建第二组元素
(integer) 1
127.0.0.1:6379> pfcount key2
(integer) 8
127.0.0.1:6379> pfmerge key3 key1 key2 # 合并两组：key1 key2 -> key3 并集
OK
127.0.0.1:6379> pfcount key3
(integer) 13
Bitmap （位存储）
Bitmap 即位图数据结构，都是操作二进制位来进行记录，只有0 和 1 两个状态。文章源自菜鸟学院-https://www.cainiaoxueyuan.com/sjk/45498.html
用来解决什么问题？
比如：统计用户信息，活跃，不活跃！登录，未登录！打卡，不打卡！两个状态的，都可以使用 Bitmaps！文章源自菜鸟学院-https://www.cainiaoxueyuan.com/sjk/45498.html
如果存储一年的打卡状态需要多少内存呢？365 天 = 365 bit 1字节 = 8bit 46 个字节左右！文章源自菜鸟学院-https://www.cainiaoxueyuan.com/sjk/45498.html
相关命令使用
使用bitmap 来记录 周一到周日的打卡！周一：1 周二：0 周三：0 周四：1 ......文章源自菜鸟学院-https://www.cainiaoxueyuan.com/sjk/45498.html
127.0.0.1:6379> setbit sign 0 1
(integer) 0
127.0.0.1:6379> setbit sign 1 1
(integer) 0
127.0.0.1:6379> setbit sign 2 0
(integer) 0
127.0.0.1:6379> setbit sign 3 1
(integer) 0
127.0.0.1:6379> setbit sign 4 0
(integer) 0
127.0.0.1:6379> setbit sign 5 0
(integer) 0
127.0.0.1:6379> setbit sign 6 1
(integer) 0
查看某一天是否有打卡！文章源自菜鸟学院-https://www.cainiaoxueyuan.com/sjk/45498.html
127.0.0.1:6379> getbit sign 3
(integer) 1
127.0.0.1:6379> getbit sign 5
(integer) 0
统计操作，统计 打卡的天数！文章源自菜鸟学院-https://www.cainiaoxueyuan.com/sjk/45498.html
127.0.0.1:6379> bitcount sign # 统计这周的打卡记录，就可以看到是否有全勤！
(integer) 3
geospatial (地理位置)
Redis 的 Geo 在 Redis 3.2 版本就推出了! 这个功能可以推算地理位置的信息: 两地之间的距离, 方圆几里的人文章源自菜鸟学院-https://www.cainiaoxueyuan.com/sjk/45498.html
geoadd 添加地理位置
127.0.0.1:6379> geoadd china:city 118.76 32.04 manjing 112.55 37.86 taiyuan 123.43 41.80 shenyang
(integer) 3
127.0.0.1:6379> geoadd china:city 144.05 22.52 shengzhen 120.16 30.24 hangzhou 108.96 34.26 xian
(integer) 3
规则
两级无法直接添加，我们一般会下载城市数据(这个网址可以查询 GEO：http://www.jsons.cn/lngcode)！文章源自菜鸟学院-https://www.cainiaoxueyuan.com/sjk/45498.html
有效的经度从-180度到180度。 有效的纬度从-85.05112878度到85.05112878度。
# 当坐标位置超出上述指定范围时，该命令将会返回一个错误。
127.0.0.1:6379> geoadd china:city 39.90 116.40 beijin
(error) ERR invalid longitude,latitude pair 39.900000,116.400000
geopos 获取指定的成员的经度和纬度
127.0.0.1:6379> geopos china:city taiyuan manjing
1) 1) "112.54999905824661255"
1) "37.86000073876942196"
2) 1) "118.75999957323074341"
1) "32.03999960287850968"
获得当前定位, 一定是一个坐标值!文章源自菜鸟学院-https://www.cainiaoxueyuan.com/sjk/45498.html
geodist
如果不存在, 返回空。文章源自菜鸟学院-https://www.cainiaoxueyuan.com/sjk/45498.html
单位如下：文章源自菜鸟学院-https://www.cainiaoxueyuan.com/sjk/45498.html
m km mi 英里 ft 英尺
127.0.0.1:6379> geodist china:city taiyuan shenyang m
"1026439.1070"
127.0.0.1:6379> geodist china:city taiyuan shenyang km
"1026.4391"
georadius
附近的人 ==> 获得所有附近的人的地址, 定位, 通过半径来查询。文章源自菜鸟学院-https://www.cainiaoxueyuan.com/sjk/45498.html
获得指定数量的人文章源自菜鸟学院-https://www.cainiaoxueyuan.com/sjk/45498.html
127.0.0.1:6379> georadius china:city 110 30 1000 km 以 100,30 这个坐标为中心, 寻找半径为1000km的城市
1) "xian"
2) "hangzhou"
3) "manjing"
4) "taiyuan"
127.0.0.1:6379> georadius china:city 110 30 500 km
1) "xian"
127.0.0.1:6379> georadius china:city 110 30 500 km withdist
1) 1) "xian"
2) "483.8340"
127.0.0.1:6379> georadius china:city 110 30 1000 km withcoord withdist count 2
1) 1) "xian"
2) "483.8340"
3) 1) "108.96000176668167114"
2) "34.25999964418929977"
2) 1) "manjing"
2) "864.9816"
3) 1) "118.75999957323074341"
2) "32.03999960287850968"
参数：key 经度 纬度 半径 单位 [显示结果的经度和纬度] [显示结果的距离] [显示的结果的数量]文章源自菜鸟学院-https://www.cainiaoxueyuan.com/sjk/45498.html
georadiusbymember
显示与指定成员一定半径范围内的其他成员文章源自菜鸟学院-https://www.cainiaoxueyuan.com/sjk/45498.html
127.0.0.1:6379> georadiusbymember china:city taiyuan 1000 km
1) "manjing"
2) "taiyuan"
3) "xian"
127.0.0.1:6379> georadiusbymember china:city taiyuan 1000 km withcoord withdist count 2
1) 1) "taiyuan"
2) "0.0000"
3) 1) "112.54999905824661255"
2) "37.86000073876942196"
2) 1) "xian"
2) "514.2264"
3) 1) "108.96000176668167114"
2) "34.25999964418929977"
参数与 georadius 一样文章源自菜鸟学院-https://www.cainiaoxueyuan.com/sjk/45498.html
geohash(较少使用)
该命令返回11个字符的hash字符串文章源自菜鸟学院-https://www.cainiaoxueyuan.com/sjk/45498.html
127.0.0.1:6379> geohash china:city taiyuan shenyang
1) "ww8p3hhqmp0"
2) "wxrvb9qyxk0"
将二维的经纬度转换为一维的字符串, 如果两个字符串越接近, 则距离越近文章源自菜鸟学院-https://www.cainiaoxueyuan.com/sjk/45498.html
底层
geo 底层的实现原理实际上就是 Zset, 我们可以通过 Zse t命令来操作 geo。文章源自菜鸟学院-https://www.cainiaoxueyuan.com/sjk/45498.html
127.0.0.1:6379> type china:city
zset
查看全部元素 删除指定的元素文章源自菜鸟学院-https://www.cainiaoxueyuan.com/sjk/45498.html
127.0.0.1:6379> zrange china:city 0 -1 withscores
1) "xian"
2) "4040115445396757"
3) "hangzhou"
4) "4054133997236782"
5) "manjing"
6) "4066006694128997"
7) "taiyuan"
8) "4068216047500484"
9) "shenyang"
1) "4072519231994779"
2) "shengzhen"
3) "4154606886655324"
127.0.0.1:6379> zrem china:city manjing
(integer) 1
127.0.0.1:6379> zrange china:city 0 -1
1) "xian"
2) "hangzhou"
3) "taiyuan"
4) "shenyang"
5) "shengzhen"
Stream 类型
为什么会设计Stream
Redis5.0 中还增加了一个数据结构 Stream，从字面上看是流类型，但其实从功能上看，应该是 Redis 对消息队列（MQ，Message Queue）的完善实现。文章源自菜鸟学院-https://www.cainiaoxueyuan.com/sjk/45498.html
用过 Redis 做消息队列的都了解，基于 Reids 的消息队列实现有很多种，例如：文章源自菜鸟学院-https://www.cainiaoxueyuan.com/sjk/45498.html
PUB/SUB，订阅/发布模式 但是发布订阅模式是无法持久化的，如果出现网络断开、Redis 宕机等，消息就会被丢弃；
基于List LPUSH+BRPOP 或者 基于Sorted-Set的实现 支持了持久化，但是不支持多播，分组消费等
为什么上面的结构无法满足广泛的MQ场景？ 这里便引出一个核心的问题：如果我们期望设计一种数据结构来实现消息队列，最重要的就是要理解设计一个消息队列需要考虑什么？初步的我们很容易想到文章源自菜鸟学院-https://www.cainiaoxueyuan.com/sjk/45498.html
消息的生产 消息的消费 单播和多播（多对多） 阻塞和非阻塞读取
消息有序性 消息的持久化
其它还要考虑啥嗯？借助美团技术团队的一篇文章，消息队列设计精要中的图文章源自菜鸟学院-https://www.cainiaoxueyuan.com/sjk/45498.html
我们不妨看看Redis考虑了哪些设计？文章源自菜鸟学院-https://www.cainiaoxueyuan.com/sjk/45498.html
消息ID的序列化生成 消息遍历 消息的阻塞和非阻塞读取 消息的分组消费 未完成消息的处理 消息队列监控 ...
这也是我们需要理解Stream的点，但是结合上面的图，我们也应该理解Redis Stream也是一种超轻量MQ并没有完全实现消息队列所有设计要点，这决定着它适用的场景。更多关于 Redis 学习的文章，请参阅：NoSQL 数据库系列之 Redis ，本系列持续更新中。文章源自菜鸟学院-https://www.cainiaoxueyuan.com/sjk/45498.html
Stream详解
经过梳理总结，我认为从以下几个大的方面去理解Stream是比较合适的，总结如下：文章源自菜鸟学院-https://www.cainiaoxueyuan.com/sjk/45498.html
Stream的结构设计 生产和消费 基本的增删查改 单一消费者的消费 消费组的消费
监控状态
Stream的结构
每个 Stream 都有唯一的名称，它就是 Redis 的 key，在我们首次使用 xadd 指令追加消息时自动创建。文章源自菜鸟学院-https://www.cainiaoxueyuan.com/sjk/45498.html
上图解析：文章源自菜鸟学院-https://www.cainiaoxueyuan.com/sjk/45498.html
Consumer Group：消费组，使用 XGROUP CREATE 命令创建，一个消费组有多个消费者(Consumer), 这些消费者之间是竞争关系。last_delivered_id：游标，每个消费组会有个游标 last_delivered_id，任意一个消费者读取了消息都会使游标 last_delivered_id 往前移动。pending_ids：消费者(Consumer)的状态变量，作用是维护消费者的未确认的 id。 pending_ids 记录了当前已经被客户端读取的消息，但是还没有ack(Acknowledge character：确认字符）。如果客户端没有ack，这个变量里面的消息ID会越来越多，一旦某个消息被ack，它就开始减少。这个pending_ids变量在Redis官方被称之为PEL，也就是Pending Entries List，这是一个很核心的数据结构，它用来确保客户端至少消费了消息一次，而不会在网络传输的中途丢失了没处理。
此外我们还需要理解两点：文章源自菜鸟学院-https://www.cainiaoxueyuan.com/sjk/45498.html
消息ID: 消息ID的形式是timestampInMillis-sequence，例如1527846880572-5，它表示当前的消息在毫米时间戳1527846880572时产生，并且是该毫秒内产生的第5条消息。消息ID可以由服务器自动生成，也可以由客户端自己指定，但是形式必须是整数-整数，而且必须是后面加入的消息的ID要大于前面的消息ID。消息内容: 消息内容就是键值对，形如hash结构的键值对，这没什么特别之处。
增删改查
消息队列相关命令：文章源自菜鸟学院-https://www.cainiaoxueyuan.com/sjk/45498.html
XADD #添加消息到末尾
XTRIM # 对流进行修剪，限制长度
XDEL #删除消息
XLEN #获取流包含的元素数量，即消息长度
XRANGE #获取消息列表，会自动过滤已经删除的消息
XREVRANGE #反向获取消息列表，ID 从大到小
XREAD #以阻塞或非阻塞方式获取消息列表
# *号表示服务器自动生成ID，后面顺序跟着一堆key/value
127.0.0.1:6379> xadd codehole * name laoqian age 30 # 名字叫laoqian，年龄30岁
1527849609889-0 # 生成的消息ID
127.0.0.1:6379> xadd codehole * name xiaoyu age 29
1527849629172-0
127.0.0.1:6379> xadd codehole * name xiaoqian age 1
1527849637634-0
127.0.0.1:6379> xlen codehole
(integer) 3
127.0.0.1:6379> xrange codehole - + # -表示最小值, +表示最大值
127.0.0.1:6379> xrange codehole - +
1) 1) 1527849609889-0
1) 1) "name"
1) "laoqian"
2) "age"
3) "30"
2) 1) 1527849629172-0
1) 1) "name"
1) "xiaoyu"
2) "age"
3) "29"
3) 1) 1527849637634-0
1) 1) "name"
1) "xiaoqian"
2) "age"
3) "1"
127.0.0.1:6379> xrange codehole 1527849629172-0 + # 指定最小消息ID的列表
1) 1) 1527849629172-0
2) 1) "name"
2) "xiaoyu"
3) "age"
4) "29"
2) 1) 1527849637634-0
2) 1) "name"
2) "xiaoqian"
3) "age"
4) "1"
127.0.0.1:6379> xrange codehole - 1527849629172-0 # 指定最大消息ID的列表
1) 1) 1527849609889-0
2) 1) "name"
2) "laoqian"
3) "age"
4) "30"
2) 1) 1527849629172-0
2) 1) "name"
2) "xiaoyu"
3) "age"
4) "29"
127.0.0.1:6379> xdel codehole 1527849609889-0
(integer) 1
127.0.0.1:6379> xlen codehole # 长度不受影响
(integer) 3
127.0.0.1:6379> xrange codehole - + # 被删除的消息没了
1) 1) 1527849629172-0
2) 1) "name"
2) "xiaoyu"
3) "age"
4) "29"
2) 1) 1527849637634-0
2) 1) "name"
2) "xiaoqian"
3) "age"
4) "1"
127.0.0.1:6379> del codehole # 删除整个Stream
(integer) 1
独立消费
我们可以在不定义消费组的情况下进行Stream消息的独立消费，当Stream没有新消息时，甚至可以阻塞等待。Redis设计了一个单独的消费指令xread，可以将Stream当成普通的消息队列(list)来使用。使用xread时，我们可以完全忽略消费组(Consumer Group)的存在，就好比Stream就是一个普通的列表(list)。文章源自菜鸟学院-https://www.cainiaoxueyuan.com/sjk/45498.html
# 从Stream头部读取两条消息
127.0.0.1:6379> xread count 2 streams codehole 0-0
1) 1) "codehole"
2) 1) 1) 1527851486781-0
2) 1) "name"
2) "laoqian"
3) "age"
4) "30"
2) 1) 1527851493405-0
2) 1) "name"
2) "yurui"
3) "age"
4) "29"
# 从Stream尾部读取一条消息，毫无疑问，这里不会返回任何消息
127.0.0.1:6379> xread count 1 streams codehole $
(nil)
# 从尾部阻塞等待新消息到来，下面的指令会堵住，直到新消息到来
127.0.0.1:6379> xread block 0 count 1 streams codehole $
# 我们从新打开一个窗口，在这个窗口往Stream里塞消息
127.0.0.1:6379> xadd codehole * name youming age 60
1527852774092-0
# 再切换到前面的窗口，我们可以看到阻塞解除了，返回了新的消息内容
# 而且还显示了一个等待时间，这里我们等待了93s
127.0.0.1:6379> xread block 0 count 1 streams codehole $
1) 1) "codehole"
2) 1) 1) 1527852774092-0
2) 1) "name"
2) "youming"
3) "age"
4) "60"
(93.11s)
客户端如果想要使用xread进行顺序消费，一定要记住当前消费到哪里了，也就是返回的消息ID。下次继续调用xread时，将上次返回的最后一个消息ID作为参数传递进去，就可以继续消费后续的消息。文章源自菜鸟学院-https://www.cainiaoxueyuan.com/sjk/45498.html
block 0表示永远阻塞，直到消息到来，block 1000表示阻塞1s，如果1s内没有任何消息到来，就返回nil文章源自菜鸟学院-https://www.cainiaoxueyuan.com/sjk/45498.html
127.0.0.1:6379> xread block 1000 count 1 streams codehole $
(nil)
(1.07s)
消费组消费
消费组消费图文章源自菜鸟学院-https://www.cainiaoxueyuan.com/sjk/45498.html
相关命令：
XGROUP CREATE #创建消费者组
XREADGROUP GROUP #读取消费者组中的消息
XACK - #将消息标记为"已处理"
XGROUP SETID #为消费者组设置新的最后递送消息ID
XGROUP DELCONSUMER #删除消费者
XGROUP DESTROY #删除消费者组
XPENDING #显示待处理消息的相关信息
XCLAIM #转移消息的归属权
XINFO #查看流和消费者组的相关信息；
XINFO GROUPS #打印消费者组的信息；
XINFO STREAM #打印流信息
创建消费组文章源自菜鸟学院-https://www.cainiaoxueyuan.com/sjk/45498.html
Stream通过xgroup create指令创建消费组(Consumer Group)，需要传递起始消息ID参数用来初始化last_delivered_id变量。文章源自菜鸟学院-https://www.cainiaoxueyuan.com/sjk/45498.html
127.0.0.1:6379> xgroup create codehole cg1 0-0 # 表示从头开始消费
OK
# $表示从尾部开始消费，只接受新消息，当前Stream消息会全部忽略
127.0.0.1:6379> xgroup create codehole cg2 $
OK
127.0.0.1:6379> xinfo stream codehole # 获取Stream信息
1) length
2) (integer) 3 # 共3个消息
3) radix-tree-keys
4) (integer) 1
5) radix-tree-nodes
6) (integer) 2
7) groups
8) (integer) 2 # 两个消费组
9) first-entry # 第一个消息
10) 1) 1527851486781-0
2) 1) "name"
2) "laoqian"
3) "age"
4) "30"
11) last-entry # 最后一个消息
12) 1) 1527851498956-0
2) 1) "name"
2) "xiaoqian"
3) "age"
4) "1"
127.0.0.1:6379> xinfo groups codehole # 获取Stream的消费组信息
1) 1) name
2) "cg1"
3) consumers
4) (integer) 0 # 该消费组还没有消费者
5) pending
6) (integer) 0 # 该消费组没有正在处理的消息
2) 1) name
2) "cg2"
3) consumers # 该消费组还没有消费者
4) (integer) 0
5) pending
6) (integer) 0 # 该消费组没有正在处理的消息
消费组消费文章源自菜鸟学院-https://www.cainiaoxueyuan.com/sjk/45498.html
Stream提供了xreadgroup指令可以进行消费组的组内消费，需要提供消费组名称、消费者名称和起始消息ID。它同xread一样，也可以阻塞等待新消息。读到新消息后，对应的消息ID就会进入消费者的PEL(正在处理的消息)结构里，客户端处理完毕后使用xack指令通知服务器，本条消息已经处理完毕，该消息ID就会从PEL中移除。文章源自菜鸟学院-https://www.cainiaoxueyuan.com/sjk/45498.html
# >号表示从当前消费组的last_delivered_id后面开始读
# 每当消费者读取一条消息，last_delivered_id变量就会前进
127.0.0.1:6379> xreadgroup GROUP cg1 c1 count 1 streams codehole >
1) 1) "codehole"
2) 1) 1) 1527851486781-0
2) 1) "name"
2) "laoqian"
3) "age"
4) "30"
127.0.0.1:6379> xreadgroup GROUP cg1 c1 count 1 streams codehole >
1) 1) "codehole"
2) 1) 1) 1527851493405-0
2) 1) "name"
2) "yurui"
3) "age"
4) "29"
127.0.0.1:6379> xreadgroup GROUP cg1 c1 count 2 streams codehole >
1) 1) "codehole"
2) 1) 1) 1527851498956-0
2) 1) "name"
2) "xiaoqian"
3) "age"
4) "1"
2) 1) 1527852774092-0
2) 1) "name"
2) "youming"
3) "age"
4) "60"
# 再继续读取，就没有新消息了
127.0.0.1:6379> xreadgroup GROUP cg1 c1 count 1 streams codehole >
(nil)
# 那就阻塞等待吧
127.0.0.1:6379> xreadgroup GROUP cg1 c1 block 0 count 1 streams codehole >
# 开启另一个窗口，往里塞消息
127.0.0.1:6379> xadd codehole * name lanying age 61
1527854062442-0
# 回到前一个窗口，发现阻塞解除，收到新消息了
127.0.0.1:6379> xreadgroup GROUP cg1 c1 block 0 count 1 streams codehole >
1) 1) "codehole"
2) 1) 1) 1527854062442-0
2) 1) "name"
2) "lanying"
3) "age"
4) "61"
(36.54s)
127.0.0.1:6379> xinfo groups codehole # 观察消费组信息
1) 1) name
2) "cg1"
3) consumers
4) (integer) 1 # 一个消费者
5) pending
6) (integer) 5 # 共5条正在处理的信息还有没有ack
2) 1) name
2) "cg2"
3) consumers
4) (integer) 0 # 消费组cg2没有任何变化，因为前面我们一直在操纵cg1
5) pending
6) (integer) 0
# 如果同一个消费组有多个消费者，我们可以通过xinfo consumers指令观察每个消费者的状态
127.0.0.1:6379> xinfo consumers codehole cg1 # 目前还有1个消费者
1) 1) name
2) "c1"
3) pending
4) (integer) 5 # 共5条待处理消息
5) idle
6) (integer) 418715 # 空闲了多长时间ms没有读取消息了
# 接下来我们ack一条消息
127.0.0.1:6379> xack codehole cg1 1527851486781-0
(integer) 1
127.0.0.1:6379> xinfo consumers codehole cg1
1) 1) name
2) "c1"
3) pending
4) (integer) 4 # 变成了5条
5) idle
6) (integer) 668504
# 下面ack所有消息
127.0.0.1:6379> xack codehole cg1 1527851493405-0 1527851498956-0 1527852774092-0 1527854062442-0
(integer) 4
127.0.0.1:6379> xinfo consumers codehole cg1
1) 1) name
2) "c1"
3) pending
4) (integer) 0 # pel空了
5) idle
6) (integer) 745505
信息监控
Stream 提供了XINFO来实现对服务器信息的监控，可以查询：文章源自菜鸟学院-https://www.cainiaoxueyuan.com/sjk/45498.html
查看队列信息
127.0.0.1:6379> Xinfo stream mq
1) "length"
2) (integer) 7
3) "radix-tree-keys"
4) (integer) 1
5) "radix-tree-nodes"
6) (integer) 2
7) "groups"
8) (integer) 1
9) "last-generated-id"
10) "1553585533795-9"
11) "first-entry"
12) 1) "1553585533795-3"
2) 1) "msg"
2) "4"
13) "last-entry"
14) 1) "1553585533795-9"
2) 1) "msg"
2) "10"
消费组信息
127.0.0.1:6379> Xinfo groups mq
1) 1) "name"
2) "mqGroup"
3) "consumers"
4) (integer) 3
5) "pending"
6) (integer) 3
7) "last-delivered-id"
8) "1553585533795-4"
消费者组成员信息
127.0.0.1:6379> XINFO CONSUMERS mq mqGroup
1) 1) "name"
2) "consumerA"
3) "pending"
4) (integer) 1
5) "idle"
6) (integer) 18949894
2) 1) "name"
2) "consumerB"
3) "pending"
4) (integer) 1
5) "idle"
6) (integer) 3092719
3) 1) "name"
2) "consumerC"
3) "pending"
4) (integer) 1
5) "idle"
6) (integer) 23683256
至此，消息队列的操作说明大体结束！更多关于 Redis 学习的文章，请参阅：NoSQL 数据库系列之 Redis ，本系列持续更新中。文章源自菜鸟学院-https://www.cainiaoxueyuan.com/sjk/45498.html
Stream用在什么样场景
可用作时通信等，大数据分析，异地数据备份等文章源自菜鸟学院-https://www.cainiaoxueyuan.com/sjk/45498.html
客户端可以平滑扩展，提高处理能力文章源自菜鸟学院-https://www.cainiaoxueyuan.com/sjk/45498.html
消息ID的设计是否考虑了时间回拨的问题？
在分布式算法 - ID算法设计中, 一个常见的问题就是时间回拨问题，那么Redis的消息ID设计中是否考虑到这个问题呢？文章源自菜鸟学院-https://www.cainiaoxueyuan.com/sjk/45498.html
XADD生成的1553439850328-0，就是Redis生成的消息ID，由两部分组成:时间戳-序号。时间戳是毫秒级单位，是生成消息的Redis服务器时间，它是个64位整型（int64）。序号是在这个毫秒时间点内的消息序号，它也是个64位整型。文章源自菜鸟学院-https://www.cainiaoxueyuan.com/sjk/45498.html
可以通过multi批处理，来验证序号的递增：文章源自菜鸟学院-https://www.cainiaoxueyuan.com/sjk/45498.html
127.0.0.1:6379> MULTI
OK
127.0.0.1:6379> XADD memberMessage * msg one
QUEUED
127.0.0.1:6379> XADD memberMessage * msg two
QUEUED
127.0.0.1:6379> XADD memberMessage * msg three
QUEUED
127.0.0.1:6379> XADD memberMessage * msg four
QUEUED
127.0.0.1:6379> XADD memberMessage * msg five
QUEUED
127.0.0.1:6379> EXEC
1) "1553441006884-0"
2) "1553441006884-1"
3) "1553441006884-2"
4) "1553441006884-3"
5) "1553441006884-4"
由于一个redis命令的执行很快，所以可以看到在同一时间戳内，是通过序号递增来表示消息的。文章源自菜鸟学院-https://www.cainiaoxueyuan.com/sjk/45498.html
为了保证消息是有序的，因此 Redis 生成的 ID 是单调递增有序的。由于 ID 中包含时间戳部分，为了避免服务器时间错误而带来的问题（例如服务器时间延后了），Redis 的每个 Stream 类型数据都维护一个 latest_generated_id属性，用于记录最后一个消息的ID。若发现当前时间戳退后（小于latest_generated_id所记录的），则采用时间戳不变而序号递增的方案来作为新消息ID（这也是序号为什么使用int64的原因，保证有足够多的的序号），从而保证ID的单调递增性质。文章源自菜鸟学院-https://www.cainiaoxueyuan.com/sjk/45498.html
强烈建议使用Redis的方案生成消息ID，因为这种时间戳+序号的单调递增的ID方案，几乎可以满足你全部的需求。但同时，记住ID是支持自定义的，别忘了！文章源自菜鸟学院-https://www.cainiaoxueyuan.com/sjk/45498.html
消费者崩溃带来的会不会消息丢失问题?
为了解决组内消息读取但处理期间消费者崩溃带来的消息丢失问题，STREAM 设计了 Pending 列表，用于记录读取但并未处理完毕的消息。命令XPENDIING 用来获消费组或消费内消费者的未处理完毕的消息。演示如下：文章源自菜鸟学院-https://www.cainiaoxueyuan.com/sjk/45498.html
127.0.0.1:6379> XPENDING mq mqGroup # mpGroup的Pending情况
1) (integer) 5 # 5个已读取但未处理的消息
2) "1553585533795-0" # 起始ID
3) "1553585533795-4" # 结束ID
4) 1) 1) "consumerA" # 消费者A有3个
2) "3"
2) 1) "consumerB" # 消费者B有1个
2) "1"
3) 1) "consumerC" # 消费者C有1个
2) "1"
127.0.0.1:6379> XPENDING mq mqGroup - + 10 # 使用 start end count 选项可以获取详细信息
1) 1) "1553585533795-0" # 消息ID
2) "consumerA" # 消费者
3) (integer) 1654355 # 从读取到现在经历了1654355ms，IDLE
4) (integer) 5 # 消息被读取了5次，delivery counter
2) 1) "1553585533795-1"
2) "consumerA"
3) (integer) 1654355
4) (integer) 4
# 共5个，余下3个省略 ...
127.0.0.1:6379> XPENDING mq mqGroup - + 10 consumerA # 在加上消费者参数，获取具体某个消费者的Pending列表
1) 1) "1553585533795-0"
2) "consumerA"
3) (integer) 1641083
4) (integer) 5
# 共3个，余下2个省略 ...
每个Pending的消息有4个属性：文章源自菜鸟学院-https://www.cainiaoxueyuan.com/sjk/45498.html
消息ID 所属消费者 IDLE，已读取时长 delivery counter，消息被读取次数
上面的结果我们可以看到，我们之前读取的消息，都被记录在Pending列表中，说明全部读到的消息都没有处理，仅仅是读取了。那如何表示消费者处理完毕了消息呢？使用命令 XACK 完成告知消息处理完成，演示如下：文章源自菜鸟学院-https://www.cainiaoxueyuan.com/sjk/45498.html
127.0.0.1:6379> XACK mq mqGroup 1553585533795-0 # 通知消息处理结束，用消息ID标识
(integer) 1
127.0.0.1:6379> XPENDING mq mqGroup # 再次查看Pending列表
1) (integer) 4 # 已读取但未处理的消息已经变为4个
2) "1553585533795-1"
3) "1553585533795-4"
4) 1) 1) "consumerA" # 消费者A，还有2个消息处理
2) "2"
2) 1) "consumerB"
2) "1"
3) 1) "consumerC"
2) "1"
127.0.0.1:6379>
有了这样一个Pending机制，就意味着在某个消费者读取消息但未处理后，消息是不会丢失的。等待消费者再次上线后，可以读取该Pending列表，就可以继续处理该消息了，保证消息的有序和不丢失。文章源自菜鸟学院-https://www.cainiaoxueyuan.com/sjk/45498.html
消费者彻底宕机后如何转移给其它消费者处理？
还有一个问题，就是若某个消费者宕机之后，没有办法再上线了，那么就需要将该消费者Pending的消息，转义给其他的消费者处理，就是消息转移。文章源自菜鸟学院-https://www.cainiaoxueyuan.com/sjk/45498.html
消息转移的操作时将某个消息转移到自己的Pending列表中。使用语法XCLAIM来实现，需要设置组、转移的目标消费者和消息ID，同时需要提供IDLE（已被读取时长），只有超过这个时长，才能被转移。演示如下：文章源自菜鸟学院-https://www.cainiaoxueyuan.com/sjk/45498.html
# 当前属于消费者A的消息1553585533795-1，已经15907,787ms未处理了
127.0.0.1:6379> XPENDING mq mqGroup - + 10
1) 1) "1553585533795-1"
2) "consumerA"
3) (integer) 15907787
4) (integer) 4
# 转移超过3600s的消息1553585533795-1到消费者B的Pending列表
127.0.0.1:6379> XCLAIM mq mqGroup consumerB 3600000 1553585533795-1
1) 1) "1553585533795-1"
2) 1) "msg"
2) "2"
# 消息1553585533795-1已经转移到消费者B的Pending中。
127.0.0.1:6379> XPENDING mq mqGroup - + 10
1) 1) "1553585533795-1"
2) "consumerB"
3) (integer) 84404 # 注意IDLE，被重置了
4) (integer) 5 # 注意，读取次数也累加了1次
以上代码，完成了一次消息转移。转移除了要指定ID外，还需要指定IDLE，保证是长时间未处理的才被转移。被转移的消息的IDLE会被重置，用以保证不会被重复转移，以为可能会出现将过期的消息同时转移给多个消费者的并发操作，设置了IDLE，则可以避免后面的转移不会成功，因为IDLE不满足条件。例如下面的连续两条转移，第二条不会成功。文章源自菜鸟学院-https://www.cainiaoxueyuan.com/sjk/45498.html
127.0.0.1:6379> XCLAIM mq mqGroup consumerB 3600000 1553585533795-1
127.0.0.1:6379> XCLAIM mq mqGroup consumerC 3600000 1553585533795-1
这就是消息转移。至此我们使用了一个 Pending 消息的 ID，所属消费者和IDLE 的属性，还有一个属性就是消息被读取次数，delivery counter，该属性的作用由于统计消息被读取的次数，包括被转移也算。这个属性主要用在判定是否为错误数据上。文章源自菜鸟学院-https://www.cainiaoxueyuan.com/sjk/45498.html
坏消息问题，Dead Letter，死信问题
正如上面所说，如果某个消息，不能被消费者处理，也就是不能被XACK，这是要长时间处于Pending列表中，即使被反复的转移给各个消费者也是如此。此时该消息的delivery counter就会累加（上一节的例子可以看到），当累加到某个我们预设的临界值时，我们就认为是坏消息（也叫死信，DeadLetter，无法投递的消息），由于有了判定条件，我们将坏消息处理掉即可，删除即可。删除一个消息，使用XDEL语法，演示如下：文章源自菜鸟学院-https://www.cainiaoxueyuan.com/sjk/45498.html
# 删除队列中的消息
127.0.0.1:6379> XDEL mq 1553585533795-1
(integer) 1
# 查看队列中再无此消息
127.0.0.1:6379> XRANGE mq - +
1) 1) "1553585533795-0"
2) 1) "msg"
2) "1"
2) 1) "1553585533795-2"
2) 1) "msg"
2) "3"
注意本例中，并没有删除 Pending 中的消息因此你查看Pending，消息还会在。可以执行 XACK 标识其处理完毕！文章源自菜鸟学院-https://www.cainiaoxueyuan.com/sjk/45498.html
文章源自菜鸟学院-https://www.cainiaoxueyuan.com/sjk/45498.html参考文章： pdai.tech/md/db/nosql-redis/db-redis-data-type-stream.html pdai.tech/md/db/nosql-redis/db-redis-data-type-special.html