MatrixCube 揭秘 102——300 行实现的完整分布式存储系统 MatrixKV

上篇文章详细的介绍了 MatrixCube 的功能与架构，MatrixCube 是 MatrixOne 数据库获得分布式能力的重要组件。

今天我们将通过一个简单的分布式存储 demo 实验来完整的体验下 MatrixCube 的功能。

MatrixKV 项目介绍

这个 demo 项目叫做 MatrixKV，在 Github 的仓库地址是：https://github.com/matrixorigin/matrixkv

MatrixKV 是一个简单的分布式强一致 KV 存储系统，采用 Pebble 作为底层的存储引擎，MatrixCube 作为分布式组件，以及自定义了最简单的读写请求接口。用户可以非常简单的在像任意一个节点发起读写数据的请求，也可以从任意一个节点读到需要的数据。

如果对 TiDB 架构比较熟悉的同学可以把 MatrixKV 等同于一个 TiKV+PD，而 MatrixKV 其中使用的 RocksDB 换成了 Pebble。

本次实验以 Docker 模拟一个小型 MatrixKV 集群的形式，来进一步说明 MatrixCube 的功能与运作机制。

第一步：环境准备

工具准备

我们这个实验需要用到 docker 与 docker-compose 工具，因此需要安装好 docker 与 docker-compose。一般来说可以直接安装 Docker-desktop，里面自带了 docker 引擎，CLI 工具及 Compose 插件。

官方提供了各种操作系统的完整安装包：https://www.docker.com/products/docker-desktop/

安装好之后可以通过以下命令查看是否安装完成，如果顺利完成安装的话会显示相应版本。

docker -vdocker-compose -v

Docker 本身是跨平台的，因此本次实验对操作系统没有要求，不过推荐 macOS12 + 或者 CentOS8+（因为完整验证过）。本次教程是在 macOS12 的环境中允许的。本次试验由于只有单机的一块硬盘，Prophet 对各个节点进行负载再平衡 Rebalance 的功能无法使用，因此在本次试验中会出现节点负载和数据量并不均衡的情况，而在完整的多机系统中可以更好体验这个功能。

Clone 代码

将 MatrixKV 代码 Clone 到本地。

git clone https://github.com/matrixorigin/matrixkv

第二步：MatrixKV 集群配置

在上一篇文章中，我们提到过 MatrixCube 基于 Raft 构建分布式共识协议，因此需要至少三个节点来作为最小部署规模，而最初的三个节点都属于调度用的 Prophet 节点。我们这次实验准备的这个小型集群有四个节点，其中三个为 Prophet 节点， 一个为数据节点。我们以 docker 进行容器包装的形式来在单机上进行模拟。

Prophet 节点设置

我们可以看到在 /cfg 文件夹中有 node0-node3 的配置文件，其中 Node0-Node2 均为 Prophet 节点，Node3 为数据节点。Prophet 节点的配置以 Node0 举例如下如下：

#raft-group的RPC通信地址，节点之间通过这个地址来发送raft message和snapshot。addr-raft = "node0:8081" 
#对客户端开放的地址，客户通过这个端口和cube来交互自定义的业务请求。addr-client = "node0:8082"
#Cube的数据存放目录，每个节点会根据这个目录所在的磁盘统计存储的使用情况，上报给调度节点。dir-data = "/tmp/matrixkv"
[raft]#Cube会对Raft的写请求做batch，会把多个写请求合并到一个Raft-Log，只做一次Raft的Proposal，这个配置指定一个Proposal的大小，这个 #配置取决于应用的具体情况max-entry-bytes = "200MB"
[replication]# 1. 一个raft-group的副本最大的down time，当一个副本的down time超过这个值，调度节点就会认为这个副本用久的故障了，# 然后会在集群中选择一个合适的节点来重新创建这个副本。如果后面这个节点又重新启动了，那么调度节点会通知这个副本# 销毁自己。# 2. 这里的默认设置一般是30分钟，这个时间我们认为是设备一般出现故障可以在30分钟内完成故障处理恢复，如果超过这个时间说明已经无法  # 恢复。在这里我们为了做实验的方便，设置成15秒。max-peer-down-time = "15s"
[prophet]#该Prophet调度节点的名称name = "node0"#该Prophet调度节点对外的RPC地址rpc-addr = "node0:8083"#指定该节点为Prophet节点prophet-node = true
[prophet.schedule]# Cube集群中的所有节点都会定期发送心跳到调度的Leader节点，当一个节点超过一定的时间都没有发送心跳，# 那么调度节点会把这个节点的状态修改为Down，并且会把这个节点上，所有的Shard在集群其他节点来重建，# 当这个节点恢复后，这个节点上的所有Shard都会收到销毁的调度消息。# 这里也是为了实验方便设置成10秒，默认也是30分钟。max-container-down-time = "10s"
#Prophet中内嵌一个ETCD作为存储元数据的组件[prophet.embed-etcd]#Cube的Prophet调度节点会先后启动, 假设我们有node0, node1, node2三个调度节点, 第一个启动的是node0节点, 那么node0节点就会#组成一个只有一个副本的etcd, 对于node0而言, `join`参数不需要填写, 后面的node1, node1启动的时候, `join`设置为node1#的Etcd的Peer addressjoin = ""#内嵌Etcd的client addressclient-urls = "http://0.0.0.0:8084"#内嵌Etcd的advertise client address, 不填写, 默认和`client-urls`一致advertise-client-urls = "http://node0:8084"#内嵌Etcd的peer addresspeer-urls = "http://0.0.0.0:8085"#内嵌Etcd的advertise peer address, 不填写, 默认和`peer-urls`一致advertise-peer-urls = "http://node0:8085"
[prophet.replication]#每个Shard最多多少个副本，当Cube的调度节点周期性巡检的时候，发现Shard的副本个数和这个值不匹配的时候，会执行创建副本或者删除副本的调#度操作。max-replicas = 3

Node1 与 Node2 的配置除了需要在 ETCD 配置部分中 join 前面的节点，其他的几乎与 Node0 没有差别。

数据节点设置

而 Node3 作为数据节点，则配置相对比较简单，除了 prophet-node 设置成 false 以外，其他没有需要额外配置的部分。

addr-raft = "node3:8081"addr-client = "node3:8082"dir-data = "/tmp/matrixkv"
[raft]max-entry-bytes = "200MB"
[prophet]name = "node3"rpc-addr = "node3:8083"prophet-node = falseexternal-etcd = [    "http://node0:8084",    "http://node1:8084",    "http://node2:8084",]

Docker-Compose 设置

Docker-compose 将根据 docker-compose.yml 中的配置来进行容器启动，其中我们需要将每个节点的数据目录改成自己指定的目录。我们以 Node0 为例。

node0:    image: matrixkv    ports:      - "8080:8080"    volumes:      - ./cfg/node0.toml:/etc/cube.toml      # /data/node0需要修改成用户指定的某个本地目录      - /data/node0:/data/matrixkv    command:       - --addr=node0:8080      - --cfg=/etc/cube.toml      # shard will split after 1024 bytes      - --shard-capacity=1024

第三步：集群启动

配置好这些选项后，在 MatrixKV 代码库中，我们已经写好了构建镜像的 dockerfile 及启动构建流程的 Makefile。

我们直接在 MatrixKV 的路径下运行 make docker 命令，它会将 MatrixKV 整体打包成镜像。

#如果是MAC X86架构平台或者Linux，可以直接运行以下命令(make docker)#如果是MAC的ARM版本，则需要将Makefile中的docker build -t matrixkv -f Dockerfile .改成docker buildx build --platform linux/amd64 -t matrixkv -f Dockerfile .
make docker

另外注意国内用户如果可能碰到 go 源站速度极慢无法下载依赖库的情况，可以在 Dockerfile 中增加 go 的中国源站设置：

RUN go env -w GOPROXY=https://goproxy.cn,direct

然后通过 docker-compose up 命令将 MatrixKV 的镜像分别根据不同的节点配置启动四份，从而形成我们的 Node0 到 Node3 的四节点集群。

docker-compose up

在 docker desktop 中我们应该就可以看到我们的 4 个 MatrixKV 的节点都以镜像的形式启动了。

在看到如下日志中出现各个节点启动监听 8080 端口的时候，就代表集群已经启动完成。

同时可以看到在我们指定的数据目录中已经开始生成了很多存储数据的文件夹以及一些初始文件。

关闭集群的话可以在启动的命令行中停止进程即可，或者也可以在 Docker desktop 中以图形化界面方式停止任意节点。

第四步：读写请求接口与路由

在启动好集群之后，我们就可以对集群进行读写数据的请求。MatrixKV 包装了几个非常简单的数据读写接口：

• 数据写入 SET：

  curl -X POST -H 'Content-Type: application/json' -d '{"key":"k1","value":"v1"}' http://127.0.0.1:8080/set
  

• 数据读取 GET：

  curl http://127.0.0.1:8080/get?key=k1
  

• 数据删除 DELETE

  curl -X POST -H 'Content-Type: application/json' -d '{"key":"k1"}' http://127.0.0.1:8080/delete
  

上一篇文章中介绍了 MatrixCube 中的 Shard Proxy，这个组件可以使得我们可以从集群的任意一个节点发起请求，不管是写入，读取还是删除的请求，Shard Proxy 都会自动将请求路由到相应的处理节点上。

比如我们可以在 node0 上写入数据，而在 node0 到 node3 上都可以进行读取，是完全一样的。

//向node0发起写入请求curl -X POST  -H 'Content-Type: application/json' -d '{"key":"k1","value":"v1"}' http://127.0.0.1:8080/set//从node0-node3进行读取curl http://127.0.0.1:8080/get?key=k1curl http://127.0.0.1:8081/get?key=k1curl http://127.0.0.1:8082/get?key=k1curl http://127.0.0.1:8083/get?key=k1

这里如果实验的系统配置及写入读取数据规模更大一些的话，大家也可以验证一些更极端的场景，比如有多个客户端在快速的读取各个节点的数据，而每次写入的数据在客户端读到的时候都可以保证是最新的以及一致的，通过这种方式可以验证 MatrixCube 的强一致性，保证任何时刻从任何节点读到的数据都是最新的以及一致的。

第五步：数据分片查询与分裂

MatrixCube 会在写入的数据量达到一定级别的时候产生 Shard 分裂，在 MatrixKV 中，我们将 Shard 的大小设置成了 1024Byte。因此写入数据超过这个尺寸的数据会产生分裂。MatrixKV 提供了一个简单的查询当前集群或者当前节点中有多少个 Shard 的接口。

#当前集群中的Shard情况curl http://127.0.0.1:8080/shards#当前节点中的Shard情况curl http://127.0.0.1:8080/shards?local=true

我们启动集群后可以看到初始状态下集群只有 3 个 Shard，id 分别为 4, 6, 8, 而他们实际存储的节点在 node0，node2 与 node3 中。

而在我们通过以下命令写入一个超过 1024Byte 的数据之后，我们可以看到 node0，node2 与 node3 中的 Shard 全部进行了分裂，每个原来的 Shard 都形成了两个新的 Shard，初始状态下的 3 个 Shard 变成了 11，12，13，15，16，17 六个 Shard。

#test.json是测试数据，数据内容需要严格按照Key，Value格式规范，比如{"key":"item0","value":"XXXXXXXXXXX"}curl -X POST  -H 'Content-Type: application/json' -d@test.json http://127.0.0.1:8083/set

同时我们仍然可以在任意一个节点访问到我们写入的数据。

第六步：节点变化与副本生成

接下来我们再来看下 MatrixCube 的高可用保障的功能。我们可以通过 Docker desktop 来手动关停单个容器，以此来模拟真实环境中的机器故障情况。

在第五步中我们输入一个较大数据之后系统整体存在 6 个 Shard，每个 Shard 有 3 个 Replica。我们现在将 node3 手动关掉。

尝试再访问 node3 的命令均以失败告终。

但是从其他节点发起读请求，数据仍然都可以读取，这也就是分布式系统对整体高可用性的体现。

按照前面我们的设置，store3 的心跳 10 秒内没有发到 Prophet，Prophet 会认为这个 Store 已经下线，而通过查看目前副本情况发现，所有的 Shard 都只有两个 Replica，为了满足 3 副本的要求，Prophet 会开始自动去寻找空闲节点，将 Shard 复制到上面，在我们这里也就是 node1，那么我们再来看下每个节点 Shard 的情况。

可以看到 node1 中以前是没有 Shard 的，现在也与 node0 和 node2 一样都有 6 个 shard。这也就是 Prophet 自动的副本生成功能，始终保证系统中有三份副本来保证高可用性。

除了副本生成以外，如果出现问题的是某一个 Shard 的 Raft Group Leader，那么这个 Shard 的 Raft Group 会重新发起选举，然后先选举新的 Leader，再由 Leader 发起请求进行新的副本生成。这个大家可以自行试验，并且通过日志的信息进行验证。

MatrixKV 代码扫描

通过整个实验我们已经完整体验了在 MatrixCube 帮助下将一个单机的 KV 存储引擎 Pebble 变成了一个分布式的 KV 存储。而其中需要 MatrixKV 本身实现的代码是非常简单的。总的来说就只有 4 个 go 文件，不到 300 行代码就可以完成 MatrixKV 的全部搭建。

1. /cmd/matrixkv.go: 整体程序启动的入口，进行最基本的初始化并启动服务。

2. /pkg/config/config.go: 定义了一个 MatrixKV 整体配置的数据结构。

3. /pkg/metadata/metadata.go：定义了用户与 MatrixKV 读写交互请求的数据结构。

4. /pkg/server/server.go：这是 MatrixKV 的最主体功能，其中主要做了三件事：

• 定义 MatrixKV server 的数据结构。

• 定义 Set/Get/Delete 等相关请求的 Executor 具体实现。

• 调用 Pebble 库作为单机存储引擎，实现 MatrixCube 指定的 DataStorage 接口，将 MatrixCube 的 Config 项设置到相应方法上。

福利时间

请加小助手微信 WeChat ID：MatrixOrigin001

1. 发送您的 MatrixKV 初体验完整录屏，即可获得限量 MatrixOrigin T 恤一件。

2. 发送您的 MatrixKV 初体验完整录屏并发布在 CSDN，即可获得价值 200 元的京东卡 + 限量 MatrixOrigin T 恤一件。

总结

作为 MatrixCube 系列的第二篇，我们通过基于 MatrixCube 和 Pebble 所实现的一个自定义分布式存储系统 MatrixKV 的实验，更进一步的展示了 MatrixCube 的运作机制，同时也展示了 300 行代码即可以迅速的搭建一个完整强一致的分布式存储系统。下一期我们将带来 MatrixCube 更加深入的代码精讲，敬请期待。

MatrixOne 社区

欢迎添加 MO 小助手微信 → ID：MatrixOrigin001，加入 MatrixOne 社群参与讨论！

官网：matrixorigin.cn

源码：github.com/matrixorigin/matrixone

Slack：matrixoneworkspace.slack.com

知乎 | CSDN | 墨天轮 | OSCHINA | SegmentFault：MatrixOrigin