跳到主要内容

基于streams的消息队列方案

前言​

Redis 从5.0版本开始提供了Streams数据类型。Streams是Redis专门为消息队列设计的数据类型，它提供了强大的消息传递功能，支持持久化、消费组、消息确认等高级特性，使其成为构建可靠消息队列的理想选择。与传统的Redis List或Pub/Sub实现相比，Streams在功能完整性、可靠性和可扩展性方面都有显著提升。

Streams架构与核心概念​

基本结构​

Streams是一个新的数据类型，每个Stream由多个消息组成，每个消息都有一个唯一的ID和一系列字段-值对。消息ID是一个64位的时间戳-序列号组合，格式为timestamp-sequence，确保了消息的全局有序性和唯一性。

消费组（Consumer Groups）​

消费组是Streams的核心特性之一，允许多个消费者共同消费一个Stream中的消息。每个消费组都有自己的消费位置，实现了消息的负载均衡和并行处理。消费者组机制支持以下特性：

  • 消息分发：消息被分配给组内的不同消费者
  • 消费状态跟踪：记录每个消费者已读取但未确认的消息
  • 重试机制：未确认的消息可以被重新分配给其他消费者

消费者（Consumers）​

消费者是消费组的成员，负责从Stream中读取和处理消息。消费者可以独立工作，也可以在消费组中协同工作。每个消费者都有自己的名称，用于标识其身份。

消息ID与偏移量（Offsets）​

  • 消息ID：全局唯一标识，格式为timestamp-sequence
  • 偏移量：消费组内记录每个消费者最后读取的消息位置
  • 特殊ID：
    • *：表示自动生成ID
    • 0：表示Stream的开始
    • $：表示Stream的末尾

相关操作指令​

消息生产​

  • XADD：向Stream中插入消息，保证有序，可以自动生成一个全局唯一ID

    # 基本语法：XADD key [ID] field value [field value ...]
    # * 表示自动生成ID
    XADD mystream * name Alice age 30
    # output: 1625097600000-0

  • XTRIM：限制Stream长度，防止无限增长

    # 保留最新的100条消息
    XTRIM mystream MAXLEN 100

  • XDEL：删除指定ID的消息

    XDEL mystream 1625097600000-0

消息消费​

  • XREAD：从Stream中读取数据，可以根据ID读取数据

    # 从指定ID开始读取
    XREAD STREAMS mystream 1625097600000-0
    # 从最新消息开始读取
    XREAD STREAMS mystream $
    # 阻塞式读取，等待新消息
    XREAD BLOCK 1000 STREAMS mystream $

  • XREADGROUP：按消费组形式读取数据

    # 基本语法：XREADGROUP GROUP group consumer STREAMS key ID
    # 从名为mygroup的消费组中读取，消费者名为consumer1
    XREADGROUP GROUP mygroup consumer1 STREAMS mystream >
    # > 表示从消费者未处理的消息开始

  • XGROUP CREATE：创建消费组

    # 创建消费组，从Stream的开始位置消费
    XGROUP CREATE mystream mygroup 0
    # 从Stream的末尾位置消费
    XGROUP CREATE mystream mygroup $

  • XGROUP SETID：修改消费组的最后读取ID

    XGROUP SETID mystream mygroup 1625097600000-0

  • XGROUP DELCONSUMER：删除消费者

    XGROUP DELCONSUMER mystream mygroup consumer1

消息确认与状态管理​

  • XPENDING：查询每个消费者组内所有消费者已读取但未确认的消息

    # 基本语法：XPENDING key group [minID maxID count] [consumer]
    XPENDING mystream mygroup
    # 查询特定消费者的待处理消息
    XPENDING mystream mygroup - + 10 consumer1

  • XACK：向消息队列确认消息处理已完成

    XACK mystream mygroup 1625097600000-0

  • XCLAIM：重新声明消息所有权，用于处理崩溃或超时的消费者未确认的消息

    # 基本语法：XCLAIM key group consumer min-idle-time ID [ID ...]
    XCLAIM mystream mygroup newconsumer 60 1625097600000-0

  • XAUTOCLAIM：自动声明未确认的消息

    # 基本语法：XAUTOCLAIM key group consumer start-id min-idle-time COUNT|JUSTID
    XAUTOCLAIM mystream mygroup newconsumer 1625097600000-0 60 100

Stream信息查询​

  • XINFO：获取Stream或消费组的元信息

    # 查询Stream信息
    XINFO STREAM mystream
    # 查询消费组信息
    XINFO GROUPS mystream
    # 查询消费者信息
    XINFO CONSUMERS mystream mygroup

  • XLEN：获取Stream中的消息数量

    XLEN mystream

完整示例​

创建消费组与消息生产​

# 创建Stream
XADD user_events * user_id 1001 event login timestamp 1625097600000
XADD user_events * user_id 1002 event logout timestamp 1625097601000
XADD user_events * user_id 1001 event purchase timestamp 1625097602000

# 创建消费组
XGROUP CREATE user_events event_consumers 0

消费者消费消息​

# 消费者1消费消息
XREADGROUP GROUP event_consumers consumer1 STREAMS user_events >
# 输出: 1) 1) "user_events"
#       2) 1) 1) "1625097600000-0"
#             2) 1) "user_id"
#                2) "1001"
#                3) "event"
#                4) "login"
#                5) "timestamp"
#                6) "1625097600000"

# 消费者2消费消息
XREADGROUP GROUP event_consumers consumer2 STREAMS user_events >
# 输出: 1) 1) "user_events"
#       2) 1) 1) "1625097601000-0"
#             2) 1) "user_id"
#                2) "1002"
#                3) "event"
#                4) "logout"
#                5) "timestamp"
#                6) "1625097601000"

消息确认与处理​

# 消费者1确认处理完成第一条消息
XACK user_events event_consumers 1625097600000-0

# 查询待处理消息
XPENDING user_events event_consumers
# 输出: 1) 1) "1625097601000-0"
#          2) "1625097601000-0"
#          3) "1625097601000"
#          4) "1"
#       2) 1) "1625097602000-0"
#          2) "1625097602000-0"
#          3) "1625097602000"
#          4) "1"

Streams与其他消息队列方案的对比​

与Redis List的比较​

特性Redis ListRedis Streams
消息顺序简单FIFO，无法保证多消费者顺序全局有序，保证消息顺序
消费模式单消费者或阻塞式消费者支持多消费组，并行消费
消息确认无确认机制完整的确认机制
消息持久化支持RDB/AOF内置持久化
消息重试无自动重试支持消息重试和重新分配
消息状态无状态跟踪完整的消费状态跟踪

与Redis Pub/Sub的比较​

特性Redis Pub/SubRedis Streams
消息传递发布-订阅模式，无状态基于队列，有状态
消息持久化无持久化，连接断开丢失支持持久化
消费者数量无限制受限于系统资源
消息确认无确认机制完整的确认机制
消息重试无重试机制支持消息重试
消息过滤基于频道过滤支持基于ID和范围过滤

Streams的优势与适用场景​

优势​

  1. 高可靠性：消息持久化与确认机制确保消息不丢失
  2. 可扩展性：消费组支持水平扩展，提高消息处理能力
  3. 有序性：保证消息的全局有序性
  4. 状态跟踪：完整的消息状态跟踪，便于监控和调试
  5. 容错性：消息重试和重新分配机制提高系统容错能力
  6. 持久化：内置持久化机制，确保数据安全

适用场景​

  1. 需要可靠消息传递的场景：如订单处理、支付系统等关键业务
  2. 需要消息顺序保证的场景：如日志处理、事件溯源等
  3. 需要多消费者并行处理的场景：如任务队列、数据处理管道等
  4. 需要消息状态跟踪的场景：如长时间运行的任务、异步处理流程等
  5. 需要消息重试机制的场景：如需要处理失败重试的业务逻辑

最佳实践​

  1. 合理设置Stream长度：使用XTRIM限制Stream长度，防止无限增长
  2. 配置合适的持久化策略：根据业务需求选择RDB、AOF或混合持久化
  3. 合理设置消费者数量：根据消息处理能力设置消费者数量，避免资源浪费
  4. 及时确认消息：确保消费者正确处理消息后发送XACK，避免消息重复处理
  5. 监控待处理消息：定期使用XPENDING监控待处理消息数量，及时发现异常
  6. 合理设置消息超时：根据业务特点设置合理的消息处理超时时间
  7. 使用消费者组实现负载均衡：通过多个消费者并行处理消息提高吞吐量
  8. 处理异常情况：实现消息重试和死信队列机制，处理异常情况

总结​

Redis Streams作为Redis 5.0引入的新特性，为Redis提供了强大的消息队列功能。通过消费组、消息确认、状态跟踪等机制，Streams实现了可靠的消息传递，适用于构建高可用、可扩展的消息系统。与传统的List或Pub/Sub实现相比，Streams在功能完整性和可靠性方面有显著提升，是构建现代分布式系统中消息队列的理想选择。

参考​