CRDT 算法

发表于 更新于 分类于

背景

CRDT (Conflict-free Replicated Data Types) 直译的话即 冲突避免可复制数据类型

在研究分布式系统时，尤其是要实现最终一致性分布式系统的过程中，一个最基本的问题就是，应该采用什么样的数据结构保证最终一致性，目前关于这个问题有一个讨论较为详尽的论文

CRDT 简介

在分布式系统中，CRDT 是指一种能够无需合作就可以在网络中多个主机中并行地复制的一种数据结构，并且总能够解决可能的不一致性。

CRDT 的类型

有两种 CRDT 都可以实现数据的最终一致性

  • 基于操作的 CRDT
  • 基于状态的 CRDT

这两种 CRDT 在数学上等价的，可以相互转换。
基于操作的 CRDT 需要消息中间件的提供额外的支持，对操作命名并保证通信过程中不会丢失或者交递信息时保证消息唯一。
基于状态的消息中间件则需要消息通信时保证全部的状态都完好地交递。

基于操作的 CRDT

基于操作的 CRDT 又被称作 commutative replicated data types 或者 CmRDTs.
CmRDT 在传播时只包含数据更新操作信息。
举个栗子，一个整数在执行了(+10), (-20) 操作后，CmRDT 广播时则只包含关于这个整数进行了(+10), (-20) 操作。其他的 DC 收到这个 CmRDT 后会在本地对相应的数据执行 CmRDT 中包含的操作。
CmRDT 所能携带的操作必须是可交换的，通信模块必须保证所有 CmRDT 包都能被正确交递，但是顺序无需保证。

基于状态的 CRDTs

基于状态的 CRDTs 全称为 convergent replicated data types 或者 CvRDTs.
CvRDTs 会将本地全部的数据状态传播给其他 DC, 这些状态在接受到后会被一个函数做 merge 处理，所以这些状态必须是可交换的，关联的，幂等的。
merge 函数会为在 CvRDT 之间提供一个 join 操作，将接收到 CvRDT 与本地数据合并。

CRDT 的几种实现

目前 CRDT 有以下几种常见的实现

  • G-Counter(Grow-Only Set)
  • PN-Counter(Positive-Negative Counter)
  • G-Set(Grow-only Set)
  • 2P-Set(Two-Phase Set)
  • LWW-Element-Set(Last Write Wins element Set)
  • OR-Set(Observed-Removed Set)
  • Sequence CRDTs

G-Counter(Grow-only Counter)

这是一个只增不减的计数器，对于 N 个节点，每个节点上维护一个长度为 N 的向量
该向量表示节点 m 上的计数，当需要增加这个计数器时，只需要任意选择一个节点操作，操作会将对应节点的计数器 . 当要统计整个集群的计数器总和时，只需要对向量 V 中的所有元素求和即可。

计算伪代码如下

1
2
3
4
5
6
7
8
9
10
11
12
payload integer[n] p
    initial [0, 0, ..., 0]
update increment()
    let g = myId()
    P[g] := P[g] + 1
query value(): integer v
    let v  = sum(p)
compare (X, Y): boolean b
    let b = i if Y.p[i] >= X.p[i]
merge (X, Y): payload Z
    for i = 0, 1, 2, ..., n - 1:
        let Z.p[i] = max(X.p[i], Y.p[i])

PN-Counter(Positive-Negative Counter)

G-Counter 有一个限制，即计数器只能增加，不能减少。不过可以使用两个计数器来实现一个既能增加也能减少计数器(PN-Counter). 简单来说，就是用一个 G-Counter 来记录所有累加结果，另一个 G-Counter 来记录累减结果，查询当前结果时，只需要计算两个 G-Counter 的差即可

1
2
3
4
5
6
7
8
9
10
11
12
13
14
payload integer[n] P, integer[n] N
    initial [0, 0, 0, ..., 0], [0, 0, 0, ..., 0]
update increment()
    let g = myID()
    P(g) := P(g) + 1
update decrement()
    let g = myID()
    N(g) := N(g) + 1
query value(): integer v
    let v = sum(P) - sum(N)
merge (X, Y): payload Z
    for i = 0, 1, ..., n - 1:
        Z.p[i] = max(X.p[i], Y.p[i])
        Z.n[i] = max(X.n[i], Y.n[i])

LWW-Element-Set(last-Write-Wins-Element-Set)

LWW-Element-Set 包含一个 add set, remove set, 在每个元素上都有一个时间戳.
每个元素都在添加到 LWW-Element-Set 时都会添加到 add-set 中，并且附带一个时间戳。将元素从 LWW-Element-Set 中移除也会添加到 remove set 中一行，并且附带一个时间戳。

当一个元素在 add-set 中时，他就会认为是 LWW-Element-Set 中的成员，或者在 remove-set 中时，就不是 LWW-Element-Set 的成员。但是如果两者都在，且 remove-set 中的时间戳小于 add-set 中的时间戳，也认为在数据库中，相反则不在。

合并两个 LWW-Element-Set 需要求两个数据库的 add-set 和 remove-set 中的并集，并且只保留时间戳最新的即可，如果时间戳相等，就需要引入 LWW-Element-Set 中的 bias.