Stream 实现方案

[KKorpse]

From issue #1584
Draft PR #1673
方案原文链接：https://t8dj523v5p.feishu.cn/wiki/YRmJwKI5oiPYtLkiQfcczmd2ndd

当前实现进度/待办

实现

• 实现方案设计
• 方案 review，批准
• Stream 基本元数据
• Stream 其它元数据：cgroup，consumer，pel
• treeID 生成器
• Stream data 存储实现
• XADD
• XRANGE
• XREAD：除了 Block 功能
• XGROUP CREATE
• XGROUP CREATECONSUMER
• XREADGROUP：除了 Block 功能
• XACK
• DEL 适配：删除一个 stream
• XTRIM
• XDEL
• XLEN
• XINFO
• XSETID
• XREVRANGE
• XGROUP SETID
• XGROUP DESTROY
• XGROUP DELCONSUMER
• XPENDING
• XGROUP HELP
• XCLAIM
• XAUTOCLAIM

测试

• XADD
• XDEL
• XTRIM
• XRANGE
• XREVRANGE
• XREAD（无 block）
• XLEN
• XINFO
• DEL：Stream 的删除

pika 实现方案

阶段一：实现基本的数据格式

可以参照 blackwidow 的实现，但是底层不用 rocksdb，而是选取现有的五种数据结构实现：set，zset，list，hash，string。但抽象其实还是和这五类数据结构一样的，将需要存储的数据分为两个部分：

• 元数据(meta_key, meta_value)
• 普通数据(data_key, data_value)

我们需要先捋清楚数据存储的格式以及存储的需求，再选取合适的数据结构。

先看结论：

• 可以全部采用 pika hash 来实现

1. stream data（message）

存储格式：类似于顺序存储的键值对，键是 ID，值是消息内容

Key：id (uint64 * 2)
Value：message(string)

存储需求：表现上是一个队列，但整体功能来看更像是一个有序 kv 结构，最核心的功能有两个：

1. 增加消息：在消息队列末尾添加一条消息，id 递增但不严格连续
2. 读取消息：stream中基本的读取方式是“返回特定 id 之后的 n 条消息”，类似于一个 range_scan 操作，先查找到第一个 id 大于特定值的消息，再往后读取 n 条。这个特定的 id 不一定是存在的，只要存在 id 比它大的消息即可。

结构选取：

为什么其它四种类型不可用？

• list：list 可以看作一个数组，它的 index 不能满足 id 不连续的特性。最重要的是，List 不能提供按 id 进行查找的功能，且它的 PKScanRange 返回的只有 key。
• zset：如果将 id 作为 score，倒是是可以满足 id 不连续但有序的特性，也可以用 ZRevrangebyscore 来实现读取操作。核心问题是：score 是固定的 64bit，而 stream id 是固定的128 bit。
• String：显然不行
• set：set 设定上是不保证有序的，况且连有序的 zset 都不可用。

那为什么可以用 hash 实现？

消息的存储方式：

• hash filed 用于存储 id，value 用于存储 message

• hash 的 key 直接使用 stream 的 key（使用时需要避免 hash 和 stream 的key冲突！）

增加消息：直接用HSet()实现，只要保证 id 递增，就可以当作是往队列末尾增加消息（其实由于它们 key 值连续，最后在rocksdb中同一层的实际存储按道理也是按序的？）。

读取消息：用 PKHScanRange() ，只要把 pattern 设置为"*"，就可以当作范围扫描，从而实现 “返回特定 id 之后的 n 条消息” 这一需求，

1. stream meta（stream, consumer_group, consumer, pel）

存储格式：由于 Stream 中有消费组的存在，导致它的依赖关系是树形的，一层套一层。比如一个 Stream 里面有多个 cgroup ，每个 cgroup 里面又有多个 consumer 。其中 consumer 和 cgroup 都有自己的元数据。

这让 Stream 的元数据不能简单地设置成下面这种固定长度的形式：

拿 stream 的元数据举例：

rax *cgroups;           /* 一棵字典树，Key 是 group_name，Value 是 group 结构的指针 */
uint64_t length;        /* Current number of elements inside this stream. */
streamID last_id;       /* Zero if there are yet no items. */
streamID first_id;      /* The first non-tombstone entry, zero if empty. */
streamID max_deleted_entry_id;  /* The maximal ID that was deleted. */
uint64_t entries_added; /* All time count of elements added. */

其中 cgroups 是指向一棵字典树的指针，树内存储的是 group_name : group_meta_data 这样的键值对。

但 pika 的基本数据结构中，没有字典树的结构，更不能用指针指向这些树。

现在看来，我们可以将上述的问题拆分成下面两个问题：

1. 怎么利用现有的五种数据结构抽象出树形结构（替代 ratix-tree）
2. 怎么用类似于指针的形式，可以指向抽象出来的树（替代指针）

只要解决这两个问题，元数据的形式直接和 redis 一致就可以了。

对上述两个问题的解决方案：

使用一个固定长度的值比如 4Bytes 作为一棵树的标识: uint32_t tree_id。

每创建一颗新的树，只需要将 tree_id加一，作为这棵树的标识即可。

而每棵树使用 hash 来抽象：

hash 的 key："tree"(string) + tree_id(uint32_t)

hash 的 filed：树的键

hash 的 value：树的值

这样一来，只要我拿到了某棵树的 tree_id，就可以用它获取到对应的 hash，也就拿到了对应的元数据。

注意：为了同普通的 hash 结构做区分，这些树的 key 都加上了 "STREE" 字符串做为前缀。为了确保用户不在使用 hash 时修改到这些元数据，他们不能使用 "tree" 开头的 key 做为 hash 结构的 key 了。

最后，在上述方案的基础上，使用 hash 设计出所有的元数据结构，元数据的内容参照 redis stream。

• Stream 元数据，统一放在一个 hash 中，该hash 的 key 设定为 "STREAM"

Filed: 
    Stream Key 
Value:
    uint64_t stream_size
    uint64_t entries_added
    treeID cgroups // 消费组kv树的前缀
    streamID last_id
    streamID first_id
    streamID max_deleted_entry_id

• Consumer Group 元数据，每个 stream 需要一颗该元数据树，这些树对应的 hash 的 key 设定为 "tree" + tree_id。

Filed: 
    group_name 
Value:
    streamID last_id
    long long entries_read
    treeID pel //  pel kv树的前缀
    treeID consumers // 消费者 kv树的前缀

• Consumer 元数据，每个 Consumer Group 需要一颗该元数据树，这些树对应的 hash 的 key 设定为 "tree" + tree_id。

Filed: 
    consumer_name
Value:
    mstime_t seen_time
    mstime_t active_time
    treeID pel //  pel kv树的前缀

• pel：Pending Entries List 元数据，每个 Consumer Group 和 Consumer 需要一颗该元数据树，这些树对应的 hash 的 key 设定为 "tree" + tree_id。

其实 pel 用 List 实现更自然，可以考虑用 List。

Key: 
    tree_id(4B) + message_id(streamID)
Value:
    mstime_t delivery_time
    uint64_t delivery_count
    treeID consum

具体需要为上述元数据抽象出如下 class：

每个元数据对应两种个类，分别用于新建元数据和读取解析现有的元数据

class StreamMetaValue
class ParsedStreamMetaValue

class CGroupMetaValue
class ParsedCGroupMetaValue

class ConsumerMetaValue
class ParsedConsumerMetaValue

class PelMetaValue
class ParsedPelMetaValue

除了上述数据类型设计，还需要考虑 TreeID 的生成

TreeID 的生成其实就是返回一个不断递增的 uint32_t 整形，只需要持久化该整形即可，可以强行用 hash 实现，在 hash 里面只存一个字段就行。
所以 TreeID 是存储在用于存储 StreamMeta 的 hash 中，该 hash 的 key 统一为 “STREAM”

Key："stream_tree_id" // 同样的，用户不能使用这个字符串做为其它 string 的 key
Value：last_tree_id (uint32_t) // 需要序列化

redis stream 实现原理（附录）

对在 pika 上怎么实现好像没有非常大的参考价值，可以简单了解下

宏观结构

代码：t_stream.c

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

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

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

数据结构：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;
}

数据结构：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来存储超出的数据。

多播支持

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

• 每个消费组通过组名称唯一标识，每个消费组都可以消费该消息队列的全部消息，多个消费组之间相互独立。
• 每个消费组可以有多个消费者，消费者通过名称唯一标识，消费者之间的关系是竞争关系，也就是说一个消息只能由该组的一个成员消费。
• 组内成员消费消息后需要确认，每个消息组都有一个待确认消息队列（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