多主复制冲突实践模式 - 版本向量

点击上方小坤探游架构笔记可以订阅哦

前面我们谈及Versioned Value原理的时候提及到由于无法保证全局顺序性, 不论是Multi-Leader复制模型还是Leaderless复制模型都无法采用Versioned Value来解决复制冲突问题, 那么今天我们来聊聊如何在多主复制模型下另一种解决冲突的模式 -- Version Vector模式.

为什么需要Version Vector

首先我们先看下面一个基于双主复制模型的双数据运营中心部署架构, UserA 以及UserB修改同一份数据分别发起write key = x 以及 write key = y 分别落到对应的IDC1以及IDC2中心, 每个IDC都能够接收key的写入请求并为对应的key赋予对应的版本号, 如下:

这个时候我们会发现在复制过程中存在多个版本, 这个时候我们面临的一个问题就是需要检测存在多个版本的key是否存在并发冲突.

为什么这么说呢? 如果我们想维护对一个key操作事件前后的因果关系, 那么自然我们会考虑引入Lamport Clock来维护key的全序顺序性, 自然也就解决了冲突问题, 但前面我们也看到其存在的局限性, 但反之则不成立, 即key = x, ver = v1 以及 key = y, ver = v2 且 v2 > v1 的结果认为key = y 在业务层面上是相较于key = x 更新.

于是我们就需要有一种用于检测上述是否存在并发冲突的模式, 这个模式我们称之为Version Vector, 即版本向量.

Version Vector实现原理与局限性

那么什么是Version Vector呢? 关于Vector, 其实我们可以换一个词来说明, 那就是数组, 也就是说Version Vector其实就是一组由多个版本组成的数组, 那么是谁来维护版本呢? 当然是对应接收数据值写入的节点, 在多主复制模型或者无主复制模型下, 我们就可以得到类似下面的版本向量, 其中node1 ~ node4 均表示为可接收数据值写入的节点, 即:

换言之, 我们通过版本向量来维护整个集群不同节点的版本值, 本质上版本向量是一组计数器的集合，集群中的每个节点对应一个计数器, 比如现在有4个节点node1 ~ node4, 分别存储对应的版本如下:

vertors = [node1: 21, node2: 34, node3: 23, node: 32]

那么当node1节点接收到数据值写入的时候, 此时node1的版本ver = 21 会递增加1 得到ver = 22, 即:

vectors = [node1: 22, node2: 34, node3: 23, node: 32]

当任意两个节点进行通信时，它们会同步各自的版本向量标记（vector stamps），通过这种方式，节点能够检测到是否存在并发更新。

怎么检测是否存在并发冲突呢? 我们仍然以上述UserA以及UserB为例, 当前IDC1以及IDC2持有的向量版本均为:

// 当前IDC1.leader 以及 IDC2.leader 均持有相同的向量:
vectors = [IDC1.leader: 22, IDC2.leader: 34]

那么当UserA、UserB分别针对相同的key发起对应的写入请求write key = x 以及 write key = y 的时候, 此时对应的版本向量需要分别递增加1 , 即:

// IDC1.leader 接收write key = x, 递增版本号得到:
key = x, vectors = [IDC1.leader: 23, IDC2.leader: 34]
// IDC2.leader 接收write key = y, 递增版本号得到:
key = y, vectors = [IDC1.leader: 22, IDC2.leader: 35]

通过上述的版本向量我们可以识别到IDC1.leader持有key对应的向量[IDC1.leader: 23, IDC2.leader: 34] 与 IDC2.leader持有key 对应的向量[IDC1.leader: 22, IDC2.leader: 35] 存在冲突, 因为两个向量不具备可比性, 是属于并发关系.

那么向量版本是如何比较呢? 版本向量的比较方式是对比每个节点对应的版本号。若满足以下两个条件，则版本向量A被认为“高于”版本向量B：

• 版本向量A与B包含相同集群节点的版本号（即节点集合完全一致）；
• 版本向量A中每个节点的版本号均大于版本向量B中对应节点的版本号。

若出现以下两种情况之一，则版本向量A与B被认为是“并发”关系：

• 两个向量中均不存在“所有节点版本号均高于对方”的情况（例如A中部分节点版本号高于B，部分低于B）；
• 两个向量包含的集群节点版本号不完全相同（即一方存在另一方没有的节点及其版本号）。

比如现在有四个可接收请求写入的节点blue、green、red以及pink, 那么这个时候版本向量的比较方式示例如下:

由此我们也可以基于上述得到的信息实现我们自己的版本向量, 以下是实现的伪代码:

public enum Ordering {
    Before,
    After,
    Concurrent
}

public class VersionVector {
      private final TreeMap<String, Long> versions;

      public VersionVector increment(String nodeId) {
            TreeMap<String, Long> versions = new TreeMap<>();       
            versions.putAll(this.versions);
            Long version = versions.get(nodeId);
            if(version == null) {
                version = 1L;
            } else {
                version = version + 1L;
            }
            versions.put(nodeId, version);
            return new VersionVector(versions);
      }

      // 自己实现版本的比较
      public Ordering compareTo(VersionVector v2) {
            Set<String> v1Nodes = this.getVersions().keySet();
            Set<String> v2Nodes = v2.getVersions().keySet();
            // 取v1nodes 与 v2nodes 的交集
            Set<String> commonNodes = retainNodes(v1Nodes, v2Nodes);
            // Determine if v1 or v2 has more nodes than common nodes       
            boolean v1Bigger = v1Nodes.size() > commonNodes.size();
            boolean v2Bigger = v2Nodes.size() > commonNodes.size();
            // Compare versions for common nodes
            for (String nodeId : commonNodes) {
                if (v1Bigger && v2Bigger) {
                    break; // No need to compare further
                }
                long v1Version = v1.getVersions().get(nodeId);           
                long v2Version = v2.getVersions().get(nodeId);           
                if (v1Version > v2Version) {
                    v1Bigger = true;
                } else if (v1Version < v2Version) {
                    v2Bigger = true;
                }
            }
            return determineOrdering(v1Bigger, v2Bigger);   
      }

      private Ordering determineOrdering(boolean v1Bigger,boolean v2Bigger) {     
            if (!v1Bigger && !v2Bigger) {
                return Ordering.Before;
            } else if (v1Bigger && !v2Bigger) {
                return Ordering.After;
            } else if (!v1Bigger && v2Bigger) {
                return Ordering.Before;
            } 
            return Ordering.Concurrent;
      }
}

那么引入版本向量之后是如何在Multi-Leader以及Leaderless复制模型中请求写入的流程是怎么样的呢?

首先, 由于每个Replica节点都可能存在处于并发状态的数据值, 因此需要维护一个带版本值的列表, 即:

public class VersionedValue {
      String value;
      VersionVector versionVector;

      @Override
      public boolean equals(Object o) {
            if (this == o) return true;
            if (o == null || getClass() != o.getClass()) return false;
            VersionedValue that = (VersionedValue) o;
            return Objects.equal(value, that.value)
                    && Objects.equal(versionVector, that.versionVector);   
      }
}

public class VersionVectorKVStore {
      Map<String, List<VersionedValue>> kv = new HashMap<>();
}

对于Multi-Leader复制模型, 以双活部署架构为例, 在上述的User A以及UserB分别发起请求写操作落到对应的node节点上,对应的node节点会针对当前的key所对应的版本号递增,其请求流程如下:

那么如果是无主复制模型, 其写入请求流程是怎么样的呢? 关于Leaderless复制模型的数据存储中会存在这样的一个设计, 即putAsPrimary, 即每个 key（或分区/分片）都有一个 preferred primary 节点，写操作通常先走 primary，再复制到其他副本。那为什么要这么做呢? 相比NWR机制, 采用putAsPrimary目的主要是在写入序列化、性能优化、持久化存储成本以及并发冲突等问题简化工程实现的复杂度.

那这个和Leader-Based有啥区别呢? 区别主要在于scope不同, Leader-Based复制模型的scope是属于全局一个主库, 所有写都必须走它, 即中心化; 而putAsPrimary in Leaderless复制模型则是:

• 每个 key（或 shard/partition）都有“天然 primary”，不同 key 可以分散在不同节点。
• 这样系统整体还是去中心化的，不存在全局瓶颈。同时每个 key 又享受到“局部 leader”的好处（性能、一致性）。

基于上述的认知之后, 我们来对比下NWR与putAsPrimary的写入路径之间的差异:

这个时候我们再来看基于版本向量的请求写入流程, 考虑到并发冲突、性能等问题带来的复杂度问题, 我们采用putAsPrimary的写入方式, 基于版本向量写入流程如下:

从上述我们引入一个协调者的身份目的是为了将内部的version信息获取细节屏蔽,仅提供对外客户端写入的API操作, 在上述流程中Replica1当选为key = name 的集群“主节点”, 整个过程中key = name 对应于版本<Replica1, Version> 上是具备单调递增, 类似于前面讲述到Lamport Clock写入流程, 通过primay获取版本, 然后再携带版本进行进行putAsPrimary写入, 这个我们的Primary节点需要进行版本更新并返回给协调者节点.

当putAsPrimay写入成功之后, 再基于协调者向其他Replica节点发起复制写请求, 这个时候我们并不需要更新版本, 而是通过上一次的putAsPrimary获得版本号之后直接复制应用到对应的Replica节点.

也许这个时候我们会有疑问, 感觉和Leader-Based方式一样, 也不需要版本向量呢.但其实不然, 通过上述其实我们会看到基于key路由到指定的primay执行写入其实存在单点问题, 那么当Replica1节点发生不可用或者发生网络分区故障, 那么这个时候就需要从其他Replica2...N中选取一个作为当前key = name的Primary如下:

根据上述的版本向量比较规则可知, 版本向量v2 [Replica1: 2, Replica2:1] > v1 [Replica1: 2], 那么什么情况会处于并发状态呢?

假如现在我们有两个客户端Client1、Client2分别对name的共享资源执行操作, 由于发生网络分区导致name对应的Primary发生变化, 如下:

可以看到Client1在进行写入的时候, name对于的Replica2是作为Primary, 但是由于发生网络分区导致使用Replica1执行putAsPrimary, 同样的Client2将putAsPrimary写入到Replica2节点, 这个时候我们会发现版本向量不具备可比性, 由此得出两个版本存在并发冲突.

既然版本向量能够帮助我们识别并发冲突, 那么当检测到并发冲突的时候, 我们要如何解决呢?

Version Vector如何解决冲突

关于并发冲突的解决方案, 在前面的领导者复制算法模型、单主复制一致性写冲突以及多主复制 & 无领导复制模型 中我们谈过几种解决冲突的方式, 这里我们主要围绕引入版本向量是帮助我们检测到数据值存在并发状态, 那么基于并发状态的数据值要么是数据复制冲突, 要么是请求读写冲突.

自定义冲突逻辑解决

不论是Multi-Leader模型还是Leaderless模型, 我们能够通过版本向量能够检测到数据值存在并发状态, 那么这个时候我们可以考虑在存储层面提供一个冲突解决器, 由对应的业务层面自己实现即可, 如下我们定义一个自定义的冲突解决器伪代码:

public interface ConflictResolver {
    VersionedValue resolve(List<VersionedValue> values);
}

public class ClusterClient{
      public VersionedValue getResolvedValue(String key, ConflictResolver resolver) {
            List<VersionedValue> versionedValues = get(key);
            return resolver.resolve(versionedValues);
      }
}

比如Riak存储就提供了自定义冲突解决器接口交由具体业务场景实现, 以下是链接:

https://docs.riak.com/riak/kv/latest/developing/usage/conflict-resolution/java/index.html

不论是复制冲突还是请求层面冲突, 都可以采用基于业务逻辑层面解决对应的冲突.

自动冲突解决Last Write Wins

对于采用这种方式其实我们在之前的文章其实有讲述过, 简单就是通过版本向量大小来决定采用最新或者最旧的版本向量作为决策数据应用到我们的存储系统中,但是这种方式很明显的不足就是存在数据丢失, 因为判断版本向量大小存在局限性, 就好比上述的两个版本向量[Replica1: 1] 以及 [Replica2: 1], 我们不能单纯地认为Replica1对应的nodeId = 1 以及 Replica2 对应的nodeId = 2 来判断是[Replica2:1] 大于版本[Replica1: 1]. 这种方式仅适用于Lamport Clock作为版本号, 因为通过事件前后的因果关系我们可以推断出版本号的顺序性.即:

class ClusterClient...
  public Optional<TimestampedVersionedValue>
                  getWithLWW(List<TimestampedVersionedValue> values) {
      return values.stream().max(Comparator.comparingLong(v -> v.timestamp));   
  }

同样地如果是唯一性数据或者共享仅被覆盖一次的数据, 那么采用LWW自动解决冲突是一个很好的方式, 不论是请求冲突还是复制冲突我们都能够基于最新或者最旧的版本值进行决策.

上述两种方式都能够为我们数据复制以及请求层面的冲突提供一个解决思路, 但除此之外, 对于数据复制冲突我们的解决方式更多是采用避免冲突的方式, 即分组或者去中心化, 比如基于key的hash分区或者上述的putAsPrimary机制,都是采取避免冲突的解决思路来控制实现的复杂度.当然还有另外一种方式, 即我们前面阐述的反熵机制,由于篇幅关系,反熵机制后续再聊.

那么在请求层面解决冲突还有哪些方式呢? 一种是读取修复Read Repair, 一种是版本太旧直接丢弃.

读取修复Read Repair

与其说Read Repair是解决冲突的方案, 倒不如说是利用已解决冲突的结果来修复其他节点旧版本的数据值, 严格来讲是消除不同节点的数据一致性问题.也就是说在客户端读取数据时触发：当冲突解决完成后，系统还能识别出哪些节点存储的是旧版本数据；在处理客户端此次读取请求的过程中，可将最新版本数据发送给这些存储旧版本的节点。这种机制被称为 “读时修复”（read repair）.

比如在下面的Blue以及Green两个节点均存储了name的两个版本值,其中green节点存储name最新值以及对应的版本号name = new & [blue:1, green: 1]. 当客户端分别从两个节点, 即blue 以及 green 节点中读取数据值, 发现green读取到name的数据值对应的版本号更新, 于是这个时候客户端发起PUT请求修复blue中存储name的旧数据, 如下:

丢弃请求 - Reject Request

存在两个Client 同时向同一个Cluster发送写请求导致的冲突, 比如下面两个用户分别都获取基于Server维度的向量版本[{}],并同时携带请求写入到同一个Cluster中, 那么哪个请求先到则先应用, 后到的请求由于版本比较旧直接丢弃, 如下:

如果我不想让它丢弃请求呢? 在前面讲述Lamport Clock的时候我们利用几个case场景都是围绕针对某个key的顺序性展开, 同样地, 我们这里可以基于ClientId这样的维度维护其递增的版本号, 比如这里的clientId分别为pink以及orange,那么这个时候我们就可以接收请求的写入并携带对应的clientId维护的版本号,如下:

但是版本向量的大小与客户端数量直接相关。这会导致集群节点随时间推移，针对某个特定键（key）累积过多并发值，该问题被称为 “兄弟值激增”（sibling explosion）. 为了解决这个问题, 其实就是转换思路, 通过捕获事件顺序来解决, 比如上述的维度替换为name的维度, 并基于Lamport Clock施加到对应携带写入name请求操作中, 于是就成为了我们的所谓的向量时钟, 即Vector Clock. 这种方式其实和 Riak 数据库采用了一种版本向量的变体 ——“带点版本向量”（dotted version vector）基本类似.

最后,我基于之前的文章整理以下关于并发冲突的解决方案汇总如下:

总结

关于Vector Clock其实我们可以理解为向量 + Lamport Clock的方式, 对于Vector Clock以及 Vector Version我们可以看到以下两者之间的区分对比:

最后这里我也Versioned Value、Lamport Clock、Vector Clock以及Vector Verstion相关的区分汇总如下:

感谢阅读, 如有收获欢迎转发!!!

你好,我是疾风先生, 主要从事互联网搜广推行业, 技术栈为java/go/python, 记录并分享个人对技术的理解与思考, 欢迎关注我的公众号, 致力于做一个有深度,有广度,有故事的工程师,欢迎成长的路上有你陪伴,关注后回复greek可添加私人微信,欢迎技术互动和交流,谢谢!