云计算 第 16 课 分布式键值对存储中的一致性问题

前面我们使用过两种分布式存储的方式：sharding 和 replication，其中一个很重要的问题就是如何在分布式键值对存储中保证数据一致性，这一次我们来学习和理解关键的机制和概念。

---

学习目标

1. 理解在分布式数据存储中一致性的不同层级
2. 实现并评估强一致性模型和最终一致性模型
3. 比较不同一致性层级的权衡
4. 了解在不同地区部署 replication 的利弊

在大型分布式应用中，应用数据会被复制到全球不同地方的机器上，其原因是：

• 可靠性/可用性：分布式数据存储保证了在单点故障时的正常操作
• 性能：可以避免一个数据存储超载
• 延迟：不同地区的请求可以访问最近的数据存储

这种把同样的数据复制到不同地区的不同机器上的分布式解决方案，在设计的时候需要考虑以下问题：

1. 从不同地区访问时的目标性能（吞吐量和延迟）
2. 需要多少台 replica，它们应该放在哪里
3. 什么时候给哪台机器发送读请求
4. 什么时候给哪台机器发写请求
5. 一次写入操作如何传递到所有 replica 上
6. 用什么层级的锁来支持特定的数据处理
7. 不同的数据存储多久需要同步一次

具体需要怎么做一定要考虑实际场景，对一致性要求没那么高的话，宽松一点未尝不可；但如果是类似银行的交易情况的话，就无论如何需要保证一致性。

举个例子：

假设我们有三个数据中心，和三个客户端，客户端 1 访问数据中心 1，客户端 2 访问数据中心 2，客户端 3 访问数据中心 3（数据中心的数据是一样的，也就是说堆任何一个数据中心数据的修改需要同步到全部的数据中心），架构如下图所示：

记某条特定的记录为 X，当前的值是 2，接着发生了下面的操作：

客户端 1 更新了 X 的值，完成之后另外两个客户端读取 X 的值，这时候，根据不同的一致性设定，另外两个客户端读取到的值可能是 1 也可能是 2。这就是一个『不一致』的例子

一致性模型

这里我们主要讨论以下五种一致性模型，从强到弱：

1. Strict 严格一致性
2. Strong 强一致性/线性一致性
3. Sequential 顺序一致性
4. Causal 因果一致性
5. Eventual 最终一致性

严格一致性

严格一致性中，每个操作都有一个绝对的全局的时间戳，所有的操作都必须按照时间戳来执行。所有的读都需要获取到最新写入的值。但是在分布式系统中，时钟本身就很难同步，并且两条指令间所间隔的时间可能比让其他节点知道这个操作所需要的时间还要短。因此严格一致性基本上是不可能的。

严格一致性其实从物理定律上来说就是不能实现的（它要求写操作能够瞬间传播出去），所有的共享访问事件都有绝对时间顺序

顺序一致性

任何执行结果都是相同的，就好像所有进程对数据存储的读、写操作是按某种序列顺序执行的，并且每个进程的操作按照程序所制定的顺序出现在这个序列中。也就是说，任何读、写操作的交叉都是可接受的，但是所有进程都看到相同的操作交叉。

如下图所示的例子，当 X 的值在数据中心 1 中更新时，对于数据中心 2 和 3 中的 值 X 的访问实际上是锁住的（不能读也不能写），只有当数据中心 2 和 3 也更新了 X 的值时，才可以继续访问：

强一致性/线性一致性

线性一致性需要一个全局时钟(vector clock)，因为它要求两个不重合的调用前一个必须在后一个之前对系统起作用，即所有线程的执行结果的线性化序列必须满足 program order，这种全局意义上的 program order又称作 real time order。这里起作用的意思就是能被系统内所有及系统外部的观察者看见。

顺序一致性则将这个条件放弱，即不重合的前一个调用不一定在后一个之前起作用，但是在同一个线程内部这个要求则是必须的（即常说的在每个线程上的投影必须满足Program order），这样就免去了全局时钟同步的需要。静态一致性是线性一致性的另一种弱化，它要求被一个静态时间点（段）隔开的两个执行部分必须满足real time order。所谓的静态时间点（段）就是指在这个时间点（段）上（内）没有悬停的方法调用（要么没有任何方法，要么这些调用都有相应的应答）。要注意的是，无论是线性一致性，串行一致性也好，静态一致性也好，要求执行的结果是可以顺序化的（串行化），即所有的进程都能看到同一个统一的串行执行顺序。

在这次的项目中我们需要实现强一致性（很难！），我们会提供显式的时间戳用来给不同操作排序。我们需要保证在任何时候，每个客户端从不同的数据中心都能读取到同样的值。

举个例子

如上图所示，因为有全局的时间戳，所以两次取钱的操作肯定有先后顺序（假设左边的更早发生），我们需要保证同时只能有一个写入操作，下一个写入操作只有在全部数据中心都完成同步之后才能进行。

因果一致性

所有进程必须以相同的顺序看到具有潜在因果关系的写操作。不同机器上的进程可以以不同的顺序看到并发的写操作（Hutto和Ahamad 1990）。

假设P1和P2是有因果关系的两个进程，例如P2的写操作信赖于P1的写操作，那么P1和P2对x的修改顺序，在P3和P4看来一定是一样的。但如果P1和P2没有关系，那么P1和P2对x的修改顺序，在P3和P4看来可以是不一样的。

具体来说，操作 1 是 PUT(A, 2)，操作 2 是 GET(A)，那么这两个操作就是有因果关系的。不同的数据中心可以定期同步或者通过指定的触发器同步。因果一致性提供了比强一致性好得多的性能。

如上图所示，如果用户 1 在 US-West 服务器上更新了某张照片，用户 2 不一定能在 Ireland 的服务器上获取到这次更新。但是所有的用户看到的对这张照片的评论的顺序是一样的（因为评论被认为是因果相关的）

最终一致性

这是以客户为中心的一致性模型，指的是在一段时间内没有数据更新操作的话，那么所有的副本将逐渐成为一致的。例如OpenStack Swift就是采用这种模型。以一次写多次读的情况下，这种模型可以工作得比较好。

唯一需要注意的是，如果多个数据中心相同记录的值都是更新过的，那么就需要保证最后是以最新的那次更新为准进行同步（新的写入要覆盖旧的写入）

具体选择哪种一致性需要根据应用场景和需求来综合考虑，不同的层级会有不同的性能表现，具体需要自行权衡

最后是一个介绍的视频，这个视频前半部分说得不错（有具体例子），后半部分不如直接看我的日志。

任务介绍

需要针对不同的场景，对后端进行优化。这次我们需要实现分布式的前端(Coordinators) 和分布式的后端(datastores)，注意，这里因为不同 datastore 的位置也不一样，相互之间的延迟也是不一样的。

场景一

可以通过社交网络进行转账，需要记录所有的交易操作，这里需要使用『强一致性』模型

场景二

可以通过社交网络认识新朋友，也可以让其他人看到自己最新认识的朋友。因为可以被不同人查看，所以会有很多的读操作，这里可以使用『最终一致性』模型

跨区域数据存储的一致性

这次我们需要管理三个前端实例(Coordinator.java)和三个后端实例(KeyValueStore.java)。不同的前后端负责处理不同区域的请求，如果代码写得好，前后端不同实例的代码应该是相同的。

和上节课一样，我们需要支持 PUT 和 GET 操作，具体的请求及描述如下：

• CoordinatorDNS:8080/consistency?consistency=CONSISTENCY_LEVEL
  • 参数的值是 strong 或 eventual，需要据此对应设置一致性模型
• CoordinatorDNS:8080/put?key=KEY&value=VALUE&timetamp=TIME
  • 需要保存到数据库中的键值对以及时间戳，利用时间戳来保证请求有序，时间戳是 long 类型数值
• CoordinatorDNS:8080/get?key=KEY&timetamp=TIME
  • 需要访问的 key 以及时间戳，应返回对应的 value（从同一个区域的 datastore 中获取），实现强一致模型的时候时间戳是有用的

时间戳是 long 类型，值大表示后发生。注意这里不提供地区的参数，需要自行从通区域的 datastore 中获取。

辅助函数

以下的 API 调用都是不同步的，也就是说任意线程可以同时调用这些 API

• KeyValueLib.PUT(String datastoreDNS, String key, String value, String timestamp, String consistency) throws IOException
  • 有延迟（具体见下表），当 datastore 接收到请求时，可以从 KeyValueStore 获取到的信息是
    • key : String
    • value : String
    • timestamp : Long
    • region : int
    • consistency : String
• KeyValueLib.GET(String datastoreDNS, String key, String timestamp, String consistency) throws IOException
  • 无延迟，在强一致性模型中仍旧需要时间戳来保证顺序，当 datastore 接收到请求时，可以从 KeyValueStore 获取到的信息是
    • key : String
    • timestamp : Long
    • consistency : String
• KeyValueLib.FORWARD(String coordinatorDNS, String key, String value, String timestamp) throws IOException
  • 有延迟，会把请求转发给特定的 Coordinator，可以通过 forward 参数来判断是由客户端发送的请求(forward=null)或是转发来的请求(forward=true)，可以用这个 API 来在不同的 Coordinator 之间转发 PUT 请求，用来同步不同 datastore 的数据，当另外的 Coordinator 接收到请求时，可以从 KeyValueStore 获取到的信息是
    • key : String
    • value : String
    • timestamp : Long
    • region : int
    • forward = "true" : String
• KeyValueLib.AHEAD(String key, String timestamp) throws IOException
  • 无延迟，告知每个地区的 datastore 在某个时刻(timestamp)有一个对于特定记录(key)的 PUT 请求，自己选择用或不用，以及如何用。可以用来给不同的 datastore 上锁，这样在同步之前对应的读写就被锁了。当 datastore 接收到请求时，可以从 KeyValueStore 获取到的信息是
    • key : String
    • timestamp : Long
• KeyValueLib.COMPLETE(String key, String timestamp) throws IOException
  • 无延迟，告知每个地区的 datastore 在某个时刻(timestamp)对于特定记录(key)的 PUT 请求已经完成，自己选择用或不用，以及如何用。可以用来给不同的 datastore 解锁，表示同步完成，其他的客户端可以访问这个 key。当 datastore 接收到请求时，可以从 KeyValueStore 获取到的信息是
    • key : String
    • timestamp : Long

只能使用上面提到 API 完成 Coordinator 和 Datastores 的交互

注意事项

注意上面有些方法是有延迟的，主要是因为地区的不同，具体三个地区为：

• us-east, 数值 1.
• us-west, 数值 2.
• ap-southeast(Singapore), 数值 3.

可以通过 KeyValueLib.region 来判断当前所在的地区（一个整型的值），这样就可以不同的机器用相同的代码，具体的延迟如下

以下几点需要注意

1. 只有在跨区域的时候才有延迟，在同一个区域的 coordinator 和 datastore 没有延迟
2. 所有的延迟都是双向的，也就是从 us-west 到 us-east 的延迟也是 200ms
3. 在 Coordinator 之间的延迟也如上表所示（使用 FORWARD 请求的时候），也就是说，从 us-east 的 coordinator 到 us-west 的 coordinator 的延迟，和从 us-east 的 coordinator 到 us-west 的 datastore 的延迟是一样的（都是 200ms）
4. 没有说明的延迟都是可以忽略的，不会影响收到的时间戳，也就是说，同区域的访问，以及使用 AHEAD 和 COMPLETE 方法都可以认为是没有延迟的。
5. 提到的延迟都是存在的，并且波动不会很大（目测是在线程中 wait 了）
6. 没有时间戳一样的请求
7. 在强一致模型中，可以认为收到的时间戳是递增的，但是在最终一致性模型中则不一定

系统需求

并行执行

并行处理请求，在 datastore 和 coordinator 上都需要

确定的行为

没有竞争条件，注意，现在要同时处理前端和后端，会难很多

非阻塞 PUT 操作

非阻塞的 PUT，需要用队列缓存着

无缓存

Coordinator 中不能有缓存

Dynamic Schema

可以根据请求来决定是使用强一致性还是最终一致性模型

通过 API 交互

只能使用给定的 API 进行交互

同区域 GET 服务

所以的数据只在同区域的数据库中请求

只有主 Coordinator 能处理对应的 PUT 请求

利用 hash 函数保证每个 key 都有一个对应的主 Coordinator，这种机制使得具体的锁机制不会太复杂。注意 key "a" 的主 Coordinator 是 us-east，key "b" 是 us-west，key "c" 是 singapore。也就是说，对应 key "a"，来说，只有 us-east 的 coordinator 能发送 PUT 请求给所有的 datacenter，如果其他 coordinator 收到 key "a" 的请求，就需要转发(FORWARD)给 us-east 的 coordinator。

注意使用参数中的时间戳来进行排序

强一致性需求

多个区域多个 coordinator 条件下实现强一致性并不简单，下面是具体的需求

强一致性

任何时候任何复制节点中相同的 key 对应相同的 value

严格排序

按时间戳顺序处理请求

原子操作

所有的 PUT 操作都应该是原子的，不能同时更新

访问控制

一个 key 在被更新时其他节点中的对应 key 不能访问

提示和建议

1. 因为读和写都有严格的一致性要求，这部分是最难实现的
2. PUT 请求总是被主 coordinator 处理，但是 GET 请求则可以直接在同区域的数据库中完成，注意在 datacenter 中也需要实现锁机制
3. KeyValueLib.AHEAD 和 KeyValueLib.COMPLETE 方法很有用，她们可以在无延迟的情况下通知所有的 datacenter

最终一致性需求

最终一致性更强调性能，不保证请求的顺序，也不保证请求的一致性。唯一的要求是新的写入操作不能被旧的写入操作覆盖。下面是具体的需求：

非阻塞读和写

读和写操作都没有任何阻塞

隐式操作顺序

时间戳在这里只是用于检查有没有覆盖新写入的值，如果是旧的写入请求，可以直接扔掉。

提示和建议

最终一致性的实现比较简单，因为不需要保证操作的顺序，所以也不需要任何的锁机制，只需要保证新的值不要被旧的覆盖即可，唯一能保证的是最终会一致，中间过程中就随便搞即可。

项目日志

开始之前要仔细阅读需求。开始之前要仔细阅读需求。开始之前要仔细阅读需求。（我都翻译得如此简练了，还是要弄懂才好）

简单来说，情况是这样，有三个地区，每个地区都有一台 Coordinator 和 KeyValueStore，也就是一共要六台机器。除此之外，还有另一台客户端机器，用来测试和提交。

• 打上标签: Project:3.3
• KeyValueStore Datacenter: ami-71e8ef1b, t1.micro
• US-East Coordinator: ami-c8f5f2a2, t1.micro
• US-West Coordinator: ami-2ff5f245, t1.micro
• AP-Southeast(Singapore) Coordinator: ami-36e8ef5c, t1.micro
• Client: ami-d72a2cbd, m1.small

具体任务列表如下：

1. 开启 3 台 Datastore 实例。为了保证稳定性，全部在 us-east 区开启。需要允许 8080 端口上的所有流量。注意在 Coordinator 中填写地址的时候不要弄错了，设定好哪个 Datastore 属于哪个 Coordinator（打好标签什么的）
2. 只需要修改 /home/ubuntu/Project3_3/vertx/bin/KeyValueStore.java 文件使其可以工作，在该文件夹下执行 ./vertx run KeyValueStore.java 即可运行。注意在开始测试之前要在所有的 KeyValueStore 都做一次
3. 开启 3 台 Coordinator 实例。为了保证稳定性，全部在 us-east 区开启。需要允许 8080 端口上的所有流量。
4. 只需要修改 /home/ubuntu/Project3_3/vertx/bin/Coordinator.java 文件使其工作, 在该文件夹下执行 ./vertx run Coordinator.java 即可运行。注意在开始测试之前要在所有的 Coordinator 都做一次
5. 实现强一致性和最终一致性，除了系统需求之外，这两个部分还有各自的要求

解题记录

分别开启 datacenter 和 coordinator 的实例，先把代码搞下来：

• Database: scp -i demo.pem ubuntu@ec2-52-91-191-111.compute-1.amazonaws.com:~/Project3_3/vertx/bin/KeyValueStore.java ./
• Coordinator: scp -i demo.pem ubuntu@ec2-54-164-37-130.compute-1.amazonaws.com:~/Project3_3/vertx/bin/Coordinator.java ./

最终一致性模型

因为最终一致性比较简单，所以就先来做这一部分，在 Coordinator 部分中，我们需要处理的是 PUT 和 GET 这两个请求，做法也比较简单，基本的流程列举如下：

对于 PUT 操作：

1. 取得 key 所对应的 primary Coordinator（也就是之前使用的哈希函数）注意这里的值是 1/2/3
2. 判断是否是转发过来的，或者是不是 key 对应的 primary Coordinator
   • 如果是，则进入第 3 步
   • 如果不是，则转发到对应的 Coordinator 上
3. 开三个新线程，分别写入到对应三个 DataCenter

对于 GET 操作，直接从本机所在地区的 datacenter 中取回数据即可。

之前我们不需要操心后台的数据存储，这次我们需要自己完成后台的任务，相对来说 DataCenter 部分也不算太难（对于最终一致性来说），我们同样只需要处理 PUT 和 GET 这两个请求，基本的流程列举如下：

对于 PUT 操作：

1. 利用 ConcurrentHashMap 来作为存储数据的结构
2. 如果数据库中没有对应的 key，那么直接把键值对存入 Map 中（注意保留时间戳）
3. 如果数据库中有对应 key，那么比较数据库中的时间戳与当前请求的时间戳，挑大的那个放回数据库（这一部分操作需要同步）

对于 GET 操作，直接从数据库中读取当前数值即可，没有其他限制

强一致性模型

我们首先需要决定的是，具体负责同步的是 Coordinator 还是 DataCenter。如果是 Coordinator，那么和之前 云计算 第 15 课 分区和复制 是一致的，可以考虑采用优先队列的方式进行处理，但是这样就出现了一个问题：如何保证 DataCenter 的访问次序呢，尤其是在有延迟的情况下。

答案是：几乎无解。有的话一定会很复杂。

所以需要改变策略，把大部分的操作放到 DataCenter 中来处理。既然决定要在 DataCenter 中实现强一致性，肯定少不了优先队列。回过头来看之前的解决方案，实际上没有用到『锁』，而是采用了等待的机制，当轮到该线程执行的时候，这个线程才开始执行。

实话说，现在我就有点看不明白之前的代码了，之前依赖于多个线程互相同步，很容易出问题（尤其是在在有了延迟的情况下），所以这次我们考虑换一个机制。

具体换成什么呢？目前的方法是这样的：Coordinator 中通过锁来控制 PUT 请求（GET 的话无所谓，直接扔到后台即可，这里有个假设很重要，就是时间戳是递增的，可以利用这个特性来简化同步机制）。在 DataCenter 中，我们会在遇到每个请求的基础上多开一个线程，用来调度和处理其他线程的操作（这样所有的控制都在一处，代码也会好写一些）。

我们具体来看看 PUT 的操作流程（和前面的最终一致性有少许类似）：

1. 取得 key 所对应的 primary Coordinator（也就是之前使用的哈希函数）注意这里的值是 1/2/3
2. 判断是否是转发过来的，或者是不是 key 对应的 primary Coordinator
   • 如果是，则进入第 3 步
   • 如果不是，则转发到对应的 Coordinator 上
3. 创建一个值为 3 的 CountDownLatch 锁，这样只有当三个 PUT 操作都完成之后，才会继续操作
4. 给所有的 DataCenter 发送 AHEAD 信息，告诉他们需要锁住了
5. 然后开启三个新的线程，对应完成 PUT 的操作，操作完成之后 CountDownLatch 锁的值减一
6. 当 CountDownLatch 锁的值变为零时，程序可以继续执行，发送 COMPLETE 信息，告诉 Datacenter 操作已经完成

然后来看看 GET 的操作：几乎不需要改动（具体的顺序会在 DataCenter 处理）

因为只涉及到一个锁，所以整个 Coordinator 的逻辑还是比较清晰的。接下来我们看看 DataCenter 的设计（会复杂一些）。首先需要思考除了用来存储数据的哈希表，我们还需要什么东西。从上次的设计来看，我们需要为每个 key 保存对应的优先队列，还需要对应的锁（之前想太复杂的机制反而放到这里非常适用），最后需要另外一个时间戳队列来处理 AHEAD 和 COMPLETE 请求（前面的锁可以看做是读写锁，这个队列可以看做是顺序锁）。

这里先来看看对应 AHEAD 和 COMPLETE 请求的操作，我们需要做的就是在接收到 AHEAD 请求的时候在对应 key 的优先队列中加入时间戳，在接收到 COMPLETE 请求的时候在对应 key 的优先队列中移除时间戳。

然后来看看对应 PUT 和 GET 请求的操作，因为我们已经有了一个用来调度的线程，所以要做的就是把请求放到对应 key 的队列中（代码其实非常简单，我的实现就两行）。

1. 遍历所有的 key，处理对应的优先队列
2. 根据读写锁来判断当前可以进行的操作
3. 根据顺序锁来判断在最新一次写入之前可以完成的读操作
4. 启动对应的线程

这里需要注意的是要随时更新数据，保证数据同步。

注意 RESET 函数也是需要写的，我没写浪费了好多时间。

写完之后我们就可以把代码上传了

• Database: scp -i demo.pem ./KeyValueStore.java ubuntu@ec2-54-84-235-244.compute-1.amazonaws.com:~/Project3_3/vertx/bin/
• Coordinator: scp -i demo.pem ./Coordinator.java ubuntu@ec2-54-88-115-252.compute-1.amazonaws.com:~/Project3_3/vertx/bin/

提交方法

1. 启动一个 m1.small(ami-d72a2cbd) 客户端实例
2. 进入文件夹 /home/ubuntu/Project3_3/
3. 其中包含文件 consistency_checker.sh, config.prop, submitter.sh 和 references.
4. 在 config.prop 填写对应的地址，执行 ./consistency_checker strong|eventual 进行测试，测试之前确保所有的实例都已经启动（并运行）
5. 把 Coordinator.java 和 KeyValueStore.java 文件复制到 /home/ubuntu/Project3_3 文件夹中（如果创建了其他文件，也需要一并包含），并注意填写 references，然后执行 submitter 进行提交
6. 网络状况也会对结果产生影响，试着避开高峰期进行测试和提交

额外任务

前面的测试只是保证结果的正确性，所以额外任务就是——性能！

测试本身非常简单粗暴，就是在强一致性模式下发送大量请求，然后系统会根据表现给出评分（注意需要保证强一致性本身没有问题再进行测试）

除了多加机器，添加缓存或者负载均衡技术，还有什么方法呢？

这时候就需要考虑流量的模式了，不同地区的流量模式不一定一样。

我们的系统会收到的请求其实是简单的『发送者-接收者』键值对，一般来说每个人不会频繁改变自己的所在位置，所以对于同一个人来说，多数的请求都会在同一个位置产生，仔细想想这里面的门道

当然这只是一个其中的思路，除了不能缓存之外，其他都可以尝试！

提交方法

填写好 config.prop 并执行 ./bonus_checker.sh。测试时间会比较长（至少要十分钟），完成之后使用 ./submitter.sh bonus 来提交。一定要记得把代码放进去，不然会扣掉全部分数

参考资料

• 存储一致性总结
• 分布式中的一致性模型