Redis stream 方案

[luky116]

原文档链接：https://t8dj523v5p.feishu.cn/wiki/YRmJwKI5oiPYtLkiQfcczmd2ndd
by 周开颜

这个 stream 功能对于 redis 来说是相对比较独立的，和其它模块耦合性非常低，同理 pika 应该也一样

1. 实现需求上的疑问

2. 实现原则

是做一个允许消息丢失的mq（比如 redis），还是一个百分百可靠的mq？

• 性能优先：数据结构放在内存，采用定期刷盘的模式，缺点是崩溃后恢复可能会导致消息丢失
• 可靠性优先：用wal日志，或者直接naive一点，用pika自带的存储做消息存储和索引，可以保证崩溃发生后，能完整恢复出整条 Stream，缺点是每条（组）消息至少多了一个刷盘延迟。

1. 需要实现的指令

具体要做到什么程度，是要做到跟 redis 一样的功能以及兼容其所有指令么？（redis stream 的实现代码，除了数据结构和测试代码，本身有 4000 多行，工作量初步看起来其实有点大，如果真的有这个需求，我可以当作一个长期的 issue 去做）

参考 redis 的指令，总共有 21 条指令（https://redis.io/commands/?group=stream）

其中核心指令有以下几条

XADD // 发布消息，创建stream
XDELL // 删除消息
XREAD // 独立消费：
XGROUP CREATE // 创建消费组
XGROUP CREATECONSUMER // 组内创建消费者
XREADGROUP GROUP // 组内消费
XACK // 确认一条消息被消费

其中比较关键的概念是消费组，每个组可以互相独立地消费消息，但组内的成员对消息是竞争消费关系。

指令具体使用以及详细解析：

• 官方文档：https://redis.io/docs/data-types/streams-tutorial/
• 博客解析：https://www.zhihu.com/question/279540635/answer/409746087?utm_id=0

1. redis stream 实现原理

2. 宏观结构

代码：t_stream.c

宏观上来看所谓消息队列其实就是一个 List，链头通过插入的方式发布消息，链尾则以向前遍历的方式进行消费。

通常来说，消息队列除了队列的性质，还需要支持对消息的快速查找，所以一般的实现方式都是一个类似于跳表的形式加上索引结构。前者用于保持队列的性质，后者用于低时间复杂度的查找。

• 索引：redis 中使用的是 radix tree，其实还可以使用 hash table，红黑树，AVL树等结构。
• 队列：redis 中使用的是 listpack 结构，队列，列表，链表类的形式都可以

1. 数据结构：radix tree

代码：rax.c

redis stream 中，针对每一条消息，都必须设置唯一且递增的消息 ID。

这种一段时间内的连续 ID，前缀都有一些高度重复性，所以用这类前缀树可以有效节约空间使用率。

除了用于存储消息，同 hash_map 的使用方式类似，它还被用来存储以下关系。

• Consumer 与 Consumer Group 之间的关联关系
• Stream 与 Consumer Group 之间的关联关系
• 每个 Consumer Group 都有一个 Pending Entries List（PEL）来追踪它尚未消费的消息。
• ...

// 源码中创建消费组时，使用了多个 radix tree
streamCG *streamCreateCG(stream *s, char *name, size_t namelen, streamID *id, long long entries_read) {
    if (s->cgroups == NULL) s->cgroups = raxNew(); // Group => Stream  映射

    streamCG *cg = zmalloc(sizeof(*cg));
    cg->pel = raxNew(); // Pending Entries List
    cg->consumers = raxNew(); // Consumer => Consumer Group 映射
    cg->last_id = *id;
    cg->entries_read = entries_read;
    raxInsert(s->cgroups,(unsigned char*)name,namelen,cg,NULL);
    return cg;
}

1. 数据结构：listpack

代码：listpack.c

作为ziplist的替代版。listpack在6以后已经作为t_hash的基础底层结构。

*A lists of strings serialization format：*一个字符串列表的序列化格式，也就是将一个字符串列表进行序列化存储。

简单的理解 listpack 就是一款专门为节省内存空间，通过特定的编码方式将数据进行编码和解码的数据结构，这种结构天生就是为节省空间而存在的。

总的来看，每个 stream 都用对应一个 radix tree，value 则存储了一个指向 listpack 的指针，同一个 listpack 可能存储多个消息的 value。

新的消息条目会被添加到Listpack中。如果添加新条目后的Listpack大小超过了设定的限制，那么会从Radix Tree中分裂出一个新的Listpack来存储超出的数据。

1. 多播支持

实际上就是对消费组的支持，主要需要注意以下细节：

• 每个消费组通过组名称唯一标识，每个消费组都可以消费该消息队列的全部消息，多个消费组之间相互独立。
• 每个消费组可以有多个消费者，消费者通过名称唯一标识，消费者之间的关系是竞争关系，也就是说一个消息只能由该组的一个成员消费。
• 组内成员消费消息后需要确认，每个消息组都有一个待确认消息队列（pending entry list, pel），用以维护该消费组已经消费但没有确认的消息。
• 消费组中的每个成员也有一个待确认消息队列，维护着该消费者已经消费尚未确认的消息。

持久化方案

理论上需要将 stream 和其用到的数据结构对 RDB 和 AOF 做支持。

RDB 支持:

src/rdb.c

// 持久化
ssize_t rdbSaveStreamPEL(rio *rdb, rax *pel, int nacks)
size_t rdbSaveStreamConsumers(rio *rdb, streamCG *cg)

// 恢复
robj *rdbLoadObject(int rdbtype, rio *rdb, sds key, int dbid, int *error)

参考博客

• https://juejin.cn/post/7094646063784525832
• https://zhuanlan.zhihu.com/p/435724042
• https://blog.csdn.net/weixin_40000301/article/details/112521383

1. pika 具体实现方式（实现文档初稿，进行中）

2. 命令注册

所有命令在以下文件夹中与cmd_table绑定（以 LPUSH 为例）

pika/src/pika_command.cc

  std::unique_ptr<Cmd> lpushptr = std::make_unique<LPushCmd>(kCmdNameLPush, -3, kCmdFlagsWrite | kCmdFlagsSingleSlot | kCmdFlagsList);
  cmd_table->insert(std::pair<std::string, std::unique_ptr<Cmd>>(kCmdNameLPush, std::move(lpushptr)));

其次在下方文件声明对应的命令，继承于 Cmd 类

pika/include/pika_list.h

class LPushCmd : public Cmd {
    // ...
}

最终通过 Do 函数，传递到命令的具体实现位置

pika/src/storage/src/redis_lists.cc

Status RedisLists::LPush(const Slice& key, const std::vector<std::string>& values, uint64_t* ret) {
    // ...
}

（原来 wiki上有现成的 https://www.cnblogs.com/sigma0-/p/12831546.html）

1. 实现位置

Stream 类算是一个存储抽象，应该是在 storage 中实现？

pika/src/storage/src/

1. 基础数据结构

2. Pika 的存储结构

疑问：代码里面并没有看到 blackwidow 相关的概念。倒是大致看到了以下的结构：

从存储的实现上来看，最底层是一个叫做Redis的类，在Redis的基础上，分别派生出了string，set，lists，hash，zset几个存储结构，它们的实现都是基于 RocksDB 的实例：

rocksdb::DB* db_;

在一个 storage 抽象中会同时拥有上述五种存储的实例（这个也和文档上的 blackwindow 概念相似）

class Storage {
private:
    std::unique_ptr<RedisStrings> strings_db_;
    std::unique_ptr<RedisHashes> hashes_db_;
    std::unique_ptr<RedisSets> sets_db_;
    std::unique_ptr<RedisZSets> zsets_db_;
    std::unique_ptr<RedisLists> lists_db_;
 }

命令在解析和执行时，有个关键的步骤用于获取 storage 实例。后续命令的实现，会从 slot 获取具体的 storage，并执行相应的操作。

void Cmd::ProcessSingleSlotCmd() {
  std::shared_ptr<Slot> slot;
  // slot 的成员之一便是 storage
  slot = g_pika_server->GetSlotByDBName(db_name_);
  // ...
}

1. Stream 的结构

stream 初步来看有三种实现方式，前两种实现方式较为现实

1. 直接用 blackwidow 实现的结构，比如 List 和 Hash

   1. 缺点：看起来貌似有点 heavy - TODO：List 实现代码略看

2. 类似 blackwidow 的实现，直接在 RocksDB 上封装一层实现？

   1. 缺点：至少需要两种抽象，索引和列表，不知道能不能在同一个 RocksDB 实例上实现，工作量大于 1

3. 抠 rax-tree 和 listpack，或者手写数据结构。

   1. 缺点：貌似和 pika 用 RocksDB 简化存储的方案不太符合，而且工程量大了很多，需要单独考虑持久化等操作（自己写的轮子大概率会效率也会很拉）

4. 实现细节

5. 命令解析

应该有大部分现成的解析代码，只需要正确链接到指令对应的实现函数就行

细节 TODO

1. 主要功能

我理解的主要功能包含：

• 创建 Stream：
• 添加消息：
• 独立读取消息：只需要保存一些元数据就行，比如当前读到的消息ID，读取的方向。
• 添加消费组：其实就是添加了一个 Group => Stream的映射，需要用一个可持久化的 hash 结构来实现，redis 中使用的是 ratix tree， pika 中可能可以用 RocksDB 来存储。除此之外，还需要存储一些相关的元数据。
• 组内添加消费者：同理，就是添加了一个 Consumer => Group的映射，以及一些元数据。
• PEL：同理

1. 持久化

如果用 RocskDB 实现，应该不用直面持久化的问题，但是暂时还不知道 pika 恢复的流程，stream 相关的元数据也要额外考虑。

细节 TODO

1. 兼容主从复制

注意到每条命令都有一个如下的函数，应该是用于生成 binlog 的函数，应该需要自己实现？

既然 Binlog 有了，按道理主从复制的逻辑不会和 Stream 的具体实现耦合度太高。

 std::string ToBinlog(uint32_t exec_time, uint32_t term_id, uint64_t logic_id, uint32_t filenum,
                               uint64_t offset) override;

• 掌握 pika client 和 sever 通讯原理，用于发布和消费，指令执行。 - 不需要，按道理只需要实现命令本身就可以做到了
• 掌握 pika 指令注册方式，需要添加新的指令并解析。
• 敲定实现需求
• 敲定消息队列需要用到的数据结构，是和 redis 的一致（比如 ratix），还是 pika 内部有合适的现成数据结构？

[jcy1001]

需求原始是想从项目从redis迁移到pika， 原来是用了redis的mq机制，不需要严格的mq服务器那种的，希望pika有这块的实现能迁移过来就很ok了