经验首页 前端设计 程序设计 Java相关 移动开发 数据库/运维 软件/图像 大数据/云计算 其他经验
当前位置:技术经验 » 数据库/运维 » Redis » 查看文章
Redis入门 - 数据类型:Stream详解
来源:cnblogs  作者:pdai  时间:2021/4/19 8:51:28  对本文有异议

Redis5.0 中还增加了一个数据类型Stream,它借鉴了Kafka的设计,是一个新的强大的支持多播的可持久化的消息队列。@pdai

为什么会设计Stream

Redis5.0 中还增加了一个数据结构Stream,从字面上看是流类型,但其实从功能上看,应该是Redis对消息队列(MQ,Message Queue)的完善实现。

用过Redis做消息队列的都了解,基于Reids的消息队列实现有很多种,例如:

  • PUB/SUB,订阅/发布模式
    • 但是发布订阅模式是无法持久化的,如果出现网络断开、Redis 宕机等,消息就会被丢弃;
  • 基于List LPUSH+BRPOP 或者 基于Sorted-Set的实现
    • 支持了持久化,但是不支持多播,分组消费等

为什么上面的结构无法满足广泛的MQ场景? 这里便引出一个核心的问题:如果我们期望设计一种数据结构来实现消息队列,最重要的就是要理解设计一个消息队列需要考虑什么?初步的我们很容易想到

  • 消息的生产
  • 消息的消费
    • 单播和多播(多对多)
    • 阻塞和非阻塞读取
  • 消息有序性
  • 消息的持久化

其它还要考虑啥嗯?借助美团技术团队的一篇文章,消息队列设计精要 中的图

我们不妨看看Redis考虑了哪些设计

  • 消息ID的序列化生成
  • 消息遍历
  • 消息的阻塞和非阻塞读取
  • 消息的分组消费
  • 未完成消息的处理
  • 消息队列监控
  • ...

这也是我们需要理解Stream的点,但是结合上面的图,我们也应该理解Redis Stream也是一种超轻量MQ并没有完全实现消息队列所有设计要点,这决定着它适用的场景。

Stream详解

经过梳理总结,我认为从以下几个大的方面去理解Stream是比较合适的,总结如下:@pdai

  • Stream的结构设计
  • 生产和消费
    • 基本的增删查改
    • 单一消费者的消费
    • 消费组的消费
  • 监控状态

Stream的结构

每个 Stream 都有唯一的名称,它就是 Redis 的 key,在我们首次使用 xadd 指令追加消息时自动创建。

上图解析:

  • 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,这是一个很核心的数据结构,它用来确保客户端至少消费了消息一次,而不会在网络传输的中途丢失了没处理。

此外我们还需要理解两点:

  • 消息ID: 消息ID的形式是timestampInMillis-sequence,例如1527846880572-5,它表示当前的消息在毫米时间戳1527846880572时产生,并且是该毫秒内产生的第5条消息。消息ID可以由服务器自动生成,也可以由客户端自己指定,但是形式必须是整数-整数,而且必须是后面加入的消息的ID要大于前面的消息ID。
  • 消息内容: 消息内容就是键值对,形如hash结构的键值对,这没什么特别之处。

增删改查

消息队列相关命令:

  • XADD - 添加消息到末尾
  • XTRIM - 对流进行修剪,限制长度
  • XDEL - 删除消息
  • XLEN - 获取流包含的元素数量,即消息长度
  • XRANGE - 获取消息列表,会自动过滤已经删除的消息
  • XREVRANGE - 反向获取消息列表,ID 从大到小
  • XREAD - 以阻塞或非阻塞方式获取消息列表
  1. # *号表示服务器自动生成ID,后面顺序跟着一堆key/value
  2. 127.0.0.1:6379> xadd codehole * name laoqian age 30 # 名字叫laoqian,年龄30岁
  3. 1527849609889-0 # 生成的消息ID
  4. 127.0.0.1:6379> xadd codehole * name xiaoyu age 29
  5. 1527849629172-0
  6. 127.0.0.1:6379> xadd codehole * name xiaoqian age 1
  7. 1527849637634-0
  8. 127.0.0.1:6379> xlen codehole
  9. (integer) 3
  10. 127.0.0.1:6379> xrange codehole - + # -表示最小值, +表示最大值
  11. 127.0.0.1:6379> xrange codehole - +
  12. 1) 1) 1527849609889-0
  13. 1) 1) "name"
  14. 1) "laoqian"
  15. 2) "age"
  16. 3) "30"
  17. 2) 1) 1527849629172-0
  18. 1) 1) "name"
  19. 1) "xiaoyu"
  20. 2) "age"
  21. 3) "29"
  22. 3) 1) 1527849637634-0
  23. 1) 1) "name"
  24. 1) "xiaoqian"
  25. 2) "age"
  26. 3) "1"
  27. 127.0.0.1:6379> xrange codehole 1527849629172-0 + # 指定最小消息ID的列表
  28. 1) 1) 1527849629172-0
  29. 2) 1) "name"
  30. 2) "xiaoyu"
  31. 3) "age"
  32. 4) "29"
  33. 2) 1) 1527849637634-0
  34. 2) 1) "name"
  35. 2) "xiaoqian"
  36. 3) "age"
  37. 4) "1"
  38. 127.0.0.1:6379> xrange codehole - 1527849629172-0 # 指定最大消息ID的列表
  39. 1) 1) 1527849609889-0
  40. 2) 1) "name"
  41. 2) "laoqian"
  42. 3) "age"
  43. 4) "30"
  44. 2) 1) 1527849629172-0
  45. 2) 1) "name"
  46. 2) "xiaoyu"
  47. 3) "age"
  48. 4) "29"
  49. 127.0.0.1:6379> xdel codehole 1527849609889-0
  50. (integer) 1
  51. 127.0.0.1:6379> xlen codehole # 长度不受影响
  52. (integer) 3
  53. 127.0.0.1:6379> xrange codehole - + # 被删除的消息没了
  54. 1) 1) 1527849629172-0
  55. 2) 1) "name"
  56. 2) "xiaoyu"
  57. 3) "age"
  58. 4) "29"
  59. 2) 1) 1527849637634-0
  60. 2) 1) "name"
  61. 2) "xiaoqian"
  62. 3) "age"
  63. 4) "1"
  64. 127.0.0.1:6379> del codehole # 删除整个Stream
  65. (integer) 1

独立消费

我们可以在不定义消费组的情况下进行Stream消息的独立消费,当Stream没有新消息时,甚至可以阻塞等待。Redis设计了一个单独的消费指令xread,可以将Stream当成普通的消息队列(list)来使用。使用xread时,我们可以完全忽略消费组(Consumer Group)的存在,就好比Stream就是一个普通的列表(list)。

  1. # 从Stream头部读取两条消息
  2. 127.0.0.1:6379> xread count 2 streams codehole 0-0
  3. 1) 1) "codehole"
  4. 2) 1) 1) 1527851486781-0
  5. 2) 1) "name"
  6. 2) "laoqian"
  7. 3) "age"
  8. 4) "30"
  9. 2) 1) 1527851493405-0
  10. 2) 1) "name"
  11. 2) "yurui"
  12. 3) "age"
  13. 4) "29"
  14. # 从Stream尾部读取一条消息,毫无疑问,这里不会返回任何消息
  15. 127.0.0.1:6379> xread count 1 streams codehole $
  16. (nil)
  17. # 从尾部阻塞等待新消息到来,下面的指令会堵住,直到新消息到来
  18. 127.0.0.1:6379> xread block 0 count 1 streams codehole $
  19. # 我们从新打开一个窗口,在这个窗口往Stream里塞消息
  20. 127.0.0.1:6379> xadd codehole * name youming age 60
  21. 1527852774092-0
  22. # 再切换到前面的窗口,我们可以看到阻塞解除了,返回了新的消息内容
  23. # 而且还显示了一个等待时间,这里我们等待了93s
  24. 127.0.0.1:6379> xread block 0 count 1 streams codehole $
  25. 1) 1) "codehole"
  26. 2) 1) 1) 1527852774092-0
  27. 2) 1) "name"
  28. 2) "youming"
  29. 3) "age"
  30. 4) "60"
  31. (93.11s)

客户端如果想要使用xread进行顺序消费,一定要记住当前消费到哪里了,也就是返回的消息ID。下次继续调用xread时,将上次返回的最后一个消息ID作为参数传递进去,就可以继续消费后续的消息。

block 0表示永远阻塞,直到消息到来,block 1000表示阻塞1s,如果1s内没有任何消息到来,就返回nil

  1. 127.0.0.1:6379> xread block 1000 count 1 streams codehole $
  2. (nil)
  3. (1.07s)

消费组消费

  • 消费组消费图

  • 相关命令:

    • XGROUP CREATE - 创建消费者组
    • XREADGROUP GROUP - 读取消费者组中的消息
    • XACK - 将消息标记为"已处理"
    • XGROUP SETID - 为消费者组设置新的最后递送消息ID
    • XGROUP DELCONSUMER - 删除消费者
    • XGROUP DESTROY - 删除消费者组
    • XPENDING - 显示待处理消息的相关信息
    • XCLAIM - 转移消息的归属权
    • XINFO - 查看流和消费者组的相关信息;
    • XINFO GROUPS - 打印消费者组的信息;
    • XINFO STREAM - 打印流信息
  • 创建消费组

Stream通过xgroup create指令创建消费组(Consumer Group),需要传递起始消息ID参数用来初始化last_delivered_id变量。

  1. 127.0.0.1:6379> xgroup create codehole cg1 0-0 # 表示从头开始消费
  2. OK
  3. # $表示从尾部开始消费,只接受新消息,当前Stream消息会全部忽略
  4. 127.0.0.1:6379> xgroup create codehole cg2 $
  5. OK
  6. 127.0.0.1:6379> xinfo stream codehole # 获取Stream信息
  7. 1) length
  8. 2) (integer) 3 # 共3个消息
  9. 3) radix-tree-keys
  10. 4) (integer) 1
  11. 5) radix-tree-nodes
  12. 6) (integer) 2
  13. 7) groups
  14. 8) (integer) 2 # 两个消费组
  15. 9) first-entry # 第一个消息
  16. 10) 1) 1527851486781-0
  17. 2) 1) "name"
  18. 2) "laoqian"
  19. 3) "age"
  20. 4) "30"
  21. 11) last-entry # 最后一个消息
  22. 12) 1) 1527851498956-0
  23. 2) 1) "name"
  24. 2) "xiaoqian"
  25. 3) "age"
  26. 4) "1"
  27. 127.0.0.1:6379> xinfo groups codehole # 获取Stream的消费组信息
  28. 1) 1) name
  29. 2) "cg1"
  30. 3) consumers
  31. 4) (integer) 0 # 该消费组还没有消费者
  32. 5) pending
  33. 6) (integer) 0 # 该消费组没有正在处理的消息
  34. 2) 1) name
  35. 2) "cg2"
  36. 3) consumers # 该消费组还没有消费者
  37. 4) (integer) 0
  38. 5) pending
  39. 6) (integer) 0 # 该消费组没有正在处理的消息
  • 消费组消费

Stream提供了xreadgroup指令可以进行消费组的组内消费,需要提供消费组名称、消费者名称和起始消息ID。它同xread一样,也可以阻塞等待新消息。读到新消息后,对应的消息ID就会进入消费者的PEL(正在处理的消息)结构里,客户端处理完毕后使用xack指令通知服务器,本条消息已经处理完毕,该消息ID就会从PEL中移除。

  1. # >号表示从当前消费组的last_delivered_id后面开始读
  2. # 每当消费者读取一条消息,last_delivered_id变量就会前进
  3. 127.0.0.1:6379> xreadgroup GROUP cg1 c1 count 1 streams codehole >
  4. 1) 1) "codehole"
  5. 2) 1) 1) 1527851486781-0
  6. 2) 1) "name"
  7. 2) "laoqian"
  8. 3) "age"
  9. 4) "30"
  10. 127.0.0.1:6379> xreadgroup GROUP cg1 c1 count 1 streams codehole >
  11. 1) 1) "codehole"
  12. 2) 1) 1) 1527851493405-0
  13. 2) 1) "name"
  14. 2) "yurui"
  15. 3) "age"
  16. 4) "29"
  17. 127.0.0.1:6379> xreadgroup GROUP cg1 c1 count 2 streams codehole >
  18. 1) 1) "codehole"
  19. 2) 1) 1) 1527851498956-0
  20. 2) 1) "name"
  21. 2) "xiaoqian"
  22. 3) "age"
  23. 4) "1"
  24. 2) 1) 1527852774092-0
  25. 2) 1) "name"
  26. 2) "youming"
  27. 3) "age"
  28. 4) "60"
  29. # 再继续读取,就没有新消息了
  30. 127.0.0.1:6379> xreadgroup GROUP cg1 c1 count 1 streams codehole >
  31. (nil)
  32. # 那就阻塞等待吧
  33. 127.0.0.1:6379> xreadgroup GROUP cg1 c1 block 0 count 1 streams codehole >
  34. # 开启另一个窗口,往里塞消息
  35. 127.0.0.1:6379> xadd codehole * name lanying age 61
  36. 1527854062442-0
  37. # 回到前一个窗口,发现阻塞解除,收到新消息了
  38. 127.0.0.1:6379> xreadgroup GROUP cg1 c1 block 0 count 1 streams codehole >
  39. 1) 1) "codehole"
  40. 2) 1) 1) 1527854062442-0
  41. 2) 1) "name"
  42. 2) "lanying"
  43. 3) "age"
  44. 4) "61"
  45. (36.54s)
  46. 127.0.0.1:6379> xinfo groups codehole # 观察消费组信息
  47. 1) 1) name
  48. 2) "cg1"
  49. 3) consumers
  50. 4) (integer) 1 # 一个消费者
  51. 5) pending
  52. 6) (integer) 5 # 共5条正在处理的信息还有没有ack
  53. 2) 1) name
  54. 2) "cg2"
  55. 3) consumers
  56. 4) (integer) 0 # 消费组cg2没有任何变化,因为前面我们一直在操纵cg1
  57. 5) pending
  58. 6) (integer) 0
  59. # 如果同一个消费组有多个消费者,我们可以通过xinfo consumers指令观察每个消费者的状态
  60. 127.0.0.1:6379> xinfo consumers codehole cg1 # 目前还有1个消费者
  61. 1) 1) name
  62. 2) "c1"
  63. 3) pending
  64. 4) (integer) 5 # 共5条待处理消息
  65. 5) idle
  66. 6) (integer) 418715 # 空闲了多长时间ms没有读取消息了
  67. # 接下来我们ack一条消息
  68. 127.0.0.1:6379> xack codehole cg1 1527851486781-0
  69. (integer) 1
  70. 127.0.0.1:6379> xinfo consumers codehole cg1
  71. 1) 1) name
  72. 2) "c1"
  73. 3) pending
  74. 4) (integer) 4 # 变成了5条
  75. 5) idle
  76. 6) (integer) 668504
  77. # 下面ack所有消息
  78. 127.0.0.1:6379> xack codehole cg1 1527851493405-0 1527851498956-0 1527852774092-0 1527854062442-0
  79. (integer) 4
  80. 127.0.0.1:6379> xinfo consumers codehole cg1
  81. 1) 1) name
  82. 2) "c1"
  83. 3) pending
  84. 4) (integer) 0 # pel空了
  85. 5) idle
  86. 6) (integer) 745505

信息监控

Stream提供了XINFO来实现对服务器信息的监控,可以查询:

  • 查看队列信息
  1. 127.0.0.1:6379> Xinfo stream mq
  2. 1) "length"
  3. 2) (integer) 7
  4. 3) "radix-tree-keys"
  5. 4) (integer) 1
  6. 5) "radix-tree-nodes"
  7. 6) (integer) 2
  8. 7) "groups"
  9. 8) (integer) 1
  10. 9) "last-generated-id"
  11. 10) "1553585533795-9"
  12. 11) "first-entry"
  13. 12) 1) "1553585533795-3"
  14. 2) 1) "msg"
  15. 2) "4"
  16. 13) "last-entry"
  17. 14) 1) "1553585533795-9"
  18. 2) 1) "msg"
  19. 2) "10"
  • 消费组信息
  1. 127.0.0.1:6379> Xinfo groups mq
  2. 1) 1) "name"
  3. 2) "mqGroup"
  4. 3) "consumers"
  5. 4) (integer) 3
  6. 5) "pending"
  7. 6) (integer) 3
  8. 7) "last-delivered-id"
  9. 8) "1553585533795-4"
  • 消费者组成员信息
  1. 127.0.0.1:6379> XINFO CONSUMERS mq mqGroup
  2. 1) 1) "name"
  3. 2) "consumerA"
  4. 3) "pending"
  5. 4) (integer) 1
  6. 5) "idle"
  7. 6) (integer) 18949894
  8. 2) 1) "name"
  9. 2) "consumerB"
  10. 3) "pending"
  11. 4) (integer) 1
  12. 5) "idle"
  13. 6) (integer) 3092719
  14. 3) 1) "name"
  15. 2) "consumerC"
  16. 3) "pending"
  17. 4) (integer) 1
  18. 5) "idle"
  19. 6) (integer) 23683256

至此,消息队列的操作说明大体结束!

更深入理解

我们结合MQ中常见问题,看Redis是如何解决的,来进一步理解Redis。

Stream用在什么样场景

可用作时通信等,大数据分析,异地数据备份等

客户端可以平滑扩展,提高处理能力

消息ID的设计是否考虑了时间回拨的问题?

分布式算法 - ID算法设计中, 一个常见的问题就是时间回拨问题,那么Redis的消息ID设计中是否考虑到这个问题呢?

XADD生成的1553439850328-0,就是Redis生成的消息ID,由两部分组成:时间戳-序号。时间戳是毫秒级单位,是生成消息的Redis服务器时间,它是个64位整型(int64)。序号是在这个毫秒时间点内的消息序号,它也是个64位整型。

可以通过multi批处理,来验证序号的递增:

  1. 127.0.0.1:6379> MULTI
  2. OK
  3. 127.0.0.1:6379> XADD memberMessage * msg one
  4. QUEUED
  5. 127.0.0.1:6379> XADD memberMessage * msg two
  6. QUEUED
  7. 127.0.0.1:6379> XADD memberMessage * msg three
  8. QUEUED
  9. 127.0.0.1:6379> XADD memberMessage * msg four
  10. QUEUED
  11. 127.0.0.1:6379> XADD memberMessage * msg five
  12. QUEUED
  13. 127.0.0.1:6379> EXEC
  14. 1) "1553441006884-0"
  15. 2) "1553441006884-1"
  16. 3) "1553441006884-2"
  17. 4) "1553441006884-3"
  18. 5) "1553441006884-4"

由于一个redis命令的执行很快,所以可以看到在同一时间戳内,是通过序号递增来表示消息的。

为了保证消息是有序的,因此Redis生成的ID是单调递增有序的。由于ID中包含时间戳部分,为了避免服务器时间错误而带来的问题(例如服务器时间延后了),Redis的每个Stream类型数据都维护一个latest_generated_id属性,用于记录最后一个消息的ID。若发现当前时间戳退后(小于latest_generated_id所记录的),则采用时间戳不变而序号递增的方案来作为新消息ID(这也是序号为什么使用int64的原因,保证有足够多的的序号),从而保证ID的单调递增性质。

强烈建议使用Redis的方案生成消息ID,因为这种时间戳+序号的单调递增的ID方案,几乎可以满足你全部的需求。但同时,记住ID是支持自定义的,别忘了!

消费者崩溃带来的会不会消息丢失问题?

为了解决组内消息读取但处理期间消费者崩溃带来的消息丢失问题,STREAM 设计了 Pending 列表,用于记录读取但并未处理完毕的消息。命令XPENDIING 用来获消费组或消费内消费者的未处理完毕的消息。演示如下:

  1. 127.0.0.1:6379> XPENDING mq mqGroup # mpGroup的Pending情况
  2. 1) (integer) 5 # 5个已读取但未处理的消息
  3. 2) "1553585533795-0" # 起始ID
  4. 3) "1553585533795-4" # 结束ID
  5. 4) 1) 1) "consumerA" # 消费者A有3个
  6. 2) "3"
  7. 2) 1) "consumerB" # 消费者B有1个
  8. 2) "1"
  9. 3) 1) "consumerC" # 消费者C有1个
  10. 2) "1"
  11. 127.0.0.1:6379> XPENDING mq mqGroup - + 10 # 使用 start end count 选项可以获取详细信息
  12. 1) 1) "1553585533795-0" # 消息ID
  13. 2) "consumerA" # 消费者
  14. 3) (integer) 1654355 # 从读取到现在经历了1654355ms,IDLE
  15. 4) (integer) 5 # 消息被读取了5次,delivery counter
  16. 2) 1) "1553585533795-1"
  17. 2) "consumerA"
  18. 3) (integer) 1654355
  19. 4) (integer) 4
  20. # 共5个,余下3个省略 ...
  21. 127.0.0.1:6379> XPENDING mq mqGroup - + 10 consumerA # 在加上消费者参数,获取具体某个消费者的Pending列表
  22. 1) 1) "1553585533795-0"
  23. 2) "consumerA"
  24. 3) (integer) 1641083
  25. 4) (integer) 5
  26. # 共3个,余下2个省略 ...

每个Pending的消息有4个属性:

  • 消息ID
  • 所属消费者
  • IDLE,已读取时长
  • delivery counter,消息被读取次数

上面的结果我们可以看到,我们之前读取的消息,都被记录在Pending列表中,说明全部读到的消息都没有处理,仅仅是读取了。那如何表示消费者处理完毕了消息呢?使用命令 XACK 完成告知消息处理完成,演示如下:

  1. 127.0.0.1:6379> XACK mq mqGroup 1553585533795-0 # 通知消息处理结束,用消息ID标识
  2. (integer) 1
  3. 127.0.0.1:6379> XPENDING mq mqGroup # 再次查看Pending列表
  4. 1) (integer) 4 # 已读取但未处理的消息已经变为4个
  5. 2) "1553585533795-1"
  6. 3) "1553585533795-4"
  7. 4) 1) 1) "consumerA" # 消费者A,还有2个消息处理
  8. 2) "2"
  9. 2) 1) "consumerB"
  10. 2) "1"
  11. 3) 1) "consumerC"
  12. 2) "1"
  13. 127.0.0.1:6379>

有了这样一个Pending机制,就意味着在某个消费者读取消息但未处理后,消息是不会丢失的。等待消费者再次上线后,可以读取该Pending列表,就可以继续处理该消息了,保证消息的有序和不丢失。

消费者彻底宕机后如何转移给其它消费者处理?

还有一个问题,就是若某个消费者宕机之后,没有办法再上线了,那么就需要将该消费者Pending的消息,转义给其他的消费者处理,就是消息转移。

消息转移的操作时将某个消息转移到自己的Pending列表中。使用语法XCLAIM来实现,需要设置组、转移的目标消费者和消息ID,同时需要提供IDLE(已被读取时长),只有超过这个时长,才能被转移。演示如下:

  1. # 当前属于消费者A的消息1553585533795-1,已经15907,787ms未处理了
  2. 127.0.0.1:6379> XPENDING mq mqGroup - + 10
  3. 1) 1) "1553585533795-1"
  4. 2) "consumerA"
  5. 3) (integer) 15907787
  6. 4) (integer) 4
  7. # 转移超过3600s的消息1553585533795-1到消费者B的Pending列表
  8. 127.0.0.1:6379> XCLAIM mq mqGroup consumerB 3600000 1553585533795-1
  9. 1) 1) "1553585533795-1"
  10. 2) 1) "msg"
  11. 2) "2"
  12. # 消息1553585533795-1已经转移到消费者B的Pending中。
  13. 127.0.0.1:6379> XPENDING mq mqGroup - + 10
  14. 1) 1) "1553585533795-1"
  15. 2) "consumerB"
  16. 3) (integer) 84404 # 注意IDLE,被重置了
  17. 4) (integer) 5 # 注意,读取次数也累加了1次

以上代码,完成了一次消息转移。转移除了要指定ID外,还需要指定IDLE,保证是长时间未处理的才被转移。被转移的消息的IDLE会被重置,用以保证不会被重复转移,以为可能会出现将过期的消息同时转移给多个消费者的并发操作,设置了IDLE,则可以避免后面的转移不会成功,因为IDLE不满足条件。例如下面的连续两条转移,第二条不会成功。

  1. 127.0.0.1:6379> XCLAIM mq mqGroup consumerB 3600000 1553585533795-1
  2. 127.0.0.1:6379> XCLAIM mq mqGroup consumerC 3600000 1553585533795-1

这就是消息转移。至此我们使用了一个Pending消息的ID,所属消费者和IDLE的属性,还有一个属性就是消息被读取次数,delivery counter,该属性的作用由于统计消息被读取的次数,包括被转移也算。这个属性主要用在判定是否为错误数据上。

坏消息问题,Dead Letter,死信问题

正如上面所说,如果某个消息,不能被消费者处理,也就是不能被XACK,这是要长时间处于Pending列表中,即使被反复的转移给各个消费者也是如此。此时该消息的delivery counter就会累加(上一节的例子可以看到),当累加到某个我们预设的临界值时,我们就认为是坏消息(也叫死信,DeadLetter,无法投递的消息),由于有了判定条件,我们将坏消息处理掉即可,删除即可。删除一个消息,使用XDEL语法,演示如下:

  1. # 删除队列中的消息
  2. 127.0.0.1:6379> XDEL mq 1553585533795-1
  3. (integer) 1
  4. # 查看队列中再无此消息
  5. 127.0.0.1:6379> XRANGE mq - +
  6. 1) 1) "1553585533795-0"
  7. 2) 1) "msg"
  8. 2) "1"
  9. 2) 1) "1553585533795-2"
  10. 2) 1) "msg"
  11. 2) "3"

注意本例中,并没有删除Pending中的消息因此你查看Pending,消息还会在。可以执行XACK标识其处理完毕!

参考文章

本文主要梳理总结自:

知识体系

知识体系

相关文章

首先,我们通过学习Redis的概念基础,了解它适用的场景。

  • Redis入门 - Redis概念和基础
    • Redis是一种支持key-value等多种数据结构的存储系统。可用于缓存,事件发布或订阅,高速队列等场景。支持网络,提供字符串,哈希,列表,队列,集合结构直接存取,基于内存,可持久化。

其次,这些适用场景都是基于Redis支持的数据类型的,所以我们需要学习它支持的数据类型;同时在redis优化中还需要对底层数据结构了解,所以也需要了解一些底层数据结构的设计和实现。

再者,需要学习Redis支持的核心功能,包括持久化,消息,事务,高可用;高可用方面包括,主从,哨兵等;高可拓展方面,比如 分片机制等。

  • Redis进阶 - 持久化:RDB和AOF机制详解
    • 为了防止数据丢失以及服务重启时能够恢复数据,Redis支持数据的持久化,主要分为两种方式,分别是RDB和AOF; 当然实际场景下还会使用这两种的混合模式。
  • Redis进阶 - 消息传递:发布订阅模式详解
    • Redis 发布订阅(pub/sub)是一种消息通信模式:发送者(pub)发送消息,订阅者(sub)接收消息。
  • Redis进阶 - 事件:Redis事件机制详解
    • Redis 采用事件驱动机制来处理大量的网络IO。它并没有使用 libevent 或者 libev 这样的成熟开源方案,而是自己实现一个非常简洁的事件驱动库 ae_event。
  • Redis进阶 - 事务:Redis事务详解
    • Redis 事务的本质是一组命令的集合。事务支持一次执行多个命令,一个事务中所有命令都会被序列化。在事务执行过程,会按照顺序串行化执行队列中的命令,其他客户端提交的命令请求不会插入到事务执行命令序列中。
  • Redis进阶 - 高可用:主从复制详解
    • 我们知道要避免单点故障,即保证高可用,便需要冗余(副本)方式提供集群服务。而Redis 提供了主从库模式,以保证数据副本的一致,主从库之间采用的是读写分离的方式。本文主要阐述Redis的主从复制。
  • Redis进阶 - 高可用:哨兵机制(Redis Sentinel)详解
    • 在上文主从复制的基础上,如果注节点出现故障该怎么办呢? 在 Redis 主从集群中,哨兵机制是实现主从库自动切换的关键机制,它有效地解决了主从复制模式下故障转移的问题。
  • Redis进阶 - 高可拓展:分片技术(Redis Cluster)详解
    • 前面两篇文章,主从复制和哨兵机制保障了高可用,就读写分离而言虽然slave节点来扩展主从的读并发能力,但是写能力和存储能力是无法进行扩展的,就只能是master节点能够承载的上限。如果面对海量数据那么必然需要构建master(主节点分片)之间的集群,同时必然需要吸收高可用(主从复制和哨兵机制)能力,即每个master分片节点还需要有slave节点,这是分布式系统中典型的纵向扩展(集群的分片技术)的体现;所以在Redis 3.0版本中对应的设计就是Redis Cluster。

最后,就是具体的实践以及实践中遇到的问题和解决方法了:在不同版本中有不同特性,所以还需要了解版本;以及性能优化,大厂实践等。

  • Redis进阶 - 缓存问题:一致性, 穿击, 穿透, 雪崩, 污染等
    • Redis最常用的一个场景就是作为缓存,本文主要探讨作为缓存,在实践中可能会有哪些问题?比如一致性, 穿击, 穿透, 雪崩, 污染等
  • Redis进阶 - 版本特性: Redis4.0、5.0、6.0特性整理
    • 在学习Redis知识体系时,我们难免会需要查看版本实现之间的差异,本文主要整理Redis较为新的版本的特性。
  • Redis进阶 - 运维监控:Redis的监控详解
    • Redis实战中包含开发,集群 和 运维,Redis用的好不好,如何让它更好,这是运维要做的;本文主要在 Redis自身状态及命令可视化监控工具,以及Redis监控体系等方面帮助你构建对redis运维/监控体系的认知,它是性能优化的前提。
  • Redis进阶 - 性能调优:Redis性能调优详解
    • Redis 的性能问题,涉及到的知识点非常广,几乎涵盖了 CPU、内存、网络、甚至磁盘的方方面面;同时还需要对上文中一些基础或底层有详细的了解。针对Redis的性能调优,这里整理分享一篇水滴与银弹(公众号)的文章,这篇文章可以帮助你构筑Redis性能调优的知识体系。
  • Redis大厂经验 - 微博:万亿级日访问量下,Redis在微博的9年优化历程
    • 再分享一篇微博使用redis的经验的文章,因为Redis在微博内部分布在各个应用场景,比如像现在春晚必争的“红包飞”活动,还有像粉丝数、用户数、阅读数、转评赞、评论盖楼、广告推荐、负反馈、音乐榜单等等都有用到Redis;我们可以通过大厂使用redis的经验来强化对redis使用上的认知。

学习资料

本篇文章由一文多发平台ArtiPub自动发布

原文链接:http://www.cnblogs.com/pengdai/p/14664214.html

 友情链接:直通硅谷  点职佳  北美留学生论坛

本站QQ群:前端 618073944 | Java 606181507 | Python 626812652 | C/C++ 612253063 | 微信 634508462 | 苹果 692586424 | C#/.net 182808419 | PHP 305140648 | 运维 608723728

W3xue 的所有内容仅供测试,对任何法律问题及风险不承担任何责任。通过使用本站内容随之而来的风险与本站无关。
关于我们  |  意见建议  |  捐助我们  |  报错有奖  |  广告合作、友情链接(目前9元/月)请联系QQ:27243702 沸活量
皖ICP备17017327号-2 皖公网安备34020702000426号