Two-phase commit(http://en.wikipedia.org/wiki/Two-phase_commit_protocol)是分布式事务最基础的协议，Three-phase commit(http://en.wikipedia.org/wiki/Three-phase_commit_protocol)主要解决Two-phase commit中协调者宕机问题。

Two-phase commit的算法实现 (from <<Distributed System: Principles and Paradigms>>)：

协调者(Coordinator)：

    write START_2PC to local log;

    multicast VOTE_REQUEST to all participants;

    while not all votes have been collected {

        wait for any incoming vote;

        if timeout {

            write GLOBAL_ABORT to local log;

            multicast GLOBAL_ABORT to all participants;

            exit;

        }

        record vote;

    }

    if all participants sent VOTE_COMMIT and coordinator votes COMMIT {

        write GLOBAL_COMMIT to local log;

        multicast GLOBAL_COMMIT to all participants;

    } else {

        write GLOBAL_ABORT to local log;

        multicast GLOBAL_ABORT to all participants;

    }

参与者(Participants)

    write INIT to local log;

    wait for VOTE_REQUEST from coordinator;

    if timeout {

        write VOTE_ABORT to local log;

        exit;

    }

    if participant votes COMMIT {

        write VOTE_COMMIT to local log;

        send VOTE_COMMIT to coordinator;

        wait for DECISION from coordinator;

        if timeout {

            multicast DECISION_REQUEST to other participants;

            wait until DECISION is received;  /* remain blocked*/

            write DECISION to local log;

        }

        if DECISION == GLOBAL_COMMIT

            write GLOBAL_COMMIT to local log;

        else if DECISION == GLOBAL_ABORT

            write GLOBAL_ABORT to local log;

    } else {

        write VOTE_ABORT to local log;

        send VOTE_ABORT to coordinator;

    }

另外，每个参与者维护一个线程专门处理其它参与者的DECISION_REQUEST请求，处理线程流程如下：

    while true {

        wait until any incoming DECISION_REQUEST is received;

        read most recently recorded STATE from the local log;

        if STATE == GLOBAL_COMMIT

            send GLOBAL_COMMIT to requesting participant;

        else if STATE == INIT or STATE == GLOBAL_ABORT;

            send GLOBAL_ABORT to requesting participant;

        else

            skip;  /* participant remains blocked */

    }

从上述的协调者与参与者的流程可以看出，如果所有参与者VOTE_COMMIT后协调者宕机，这个时候每个参与者都无法单独决定全局事务的最终结果(GLOBAL_COMMIT还是GLOBAL_ABORT)，也无法从其它参与者获取，整个事务一直阻塞到协调者恢复；如果协调者出现类似磁盘坏这种永久性错误，该事务将成为被永久遗弃的孤儿。问题的解决有如下思路：

1. 协调者持久化数据定期备份。为了防止协调者出现永久性错误，这是一种代价最小的解决方法，不容易引入bug，但是事务被阻塞的时间可能特别长，比较适合银行这种正确性高于一切的系统。

2. Three-phase Commit。这是理论上的一种方法，实现起来复杂且效率低。思路如下：假设参与者机器不可能出现超过一半同时宕机的情况，如果协调者宕机，我们需要从活着的超过一半的参与者中得出事务的全局结果。由于不可能知道已经宕机的参与者的状态，所以引入一个新的参与者状态PRECOMMIT，参与者成功执行一个事务需要经过INIT, READY, PRECOMMIT，最后到COMMIT状态；如果至少有一个参与者处于PRECOMMIT或者COMMIT，事务成功；如果至少一个参与者处于INIT或者ABORT，事务失败；如果所有的参与者都处于READY(至少一半参与者活着)，事务失败，即使原先宕机的参与者恢复后处于PRECOMMIT状态，也会因为有其它参与者处于ABORT状态而回滚。PRECOMMIT状态的引入给了宕机的参与者回滚机会，所以Three-phase commit在超过一半的参与者活着的时候是不阻塞的。不过，Three-phase Commit只能算是是理论上的探索，效率低并且没有解决网络分区问题。

3. Paxos解决协调者单点问题。Jim Gray和Lamport合作了一篇论文讲这个方法，很适合互联网公司的超大规模集群，Google的Megastore事务就是这样实现的，不过问题在于Paxos和Two-phase Commit都不简单，需要有比较靠谱（代码质量高）的小团队设计和编码才行。后续的blog将详细阐述该方法。

总之，分布式事务只能是系统开发者的乌托邦式理想，Two-phase commit的介入将导致涉及多台机器的事务之间完全串行，没有代价的分布式事务是不存在的。

前面我的一篇文章http://hi.baidu.com/knuthocean/blog/item/12bb9f3dea0e400abba1673c.html引用了对Google App Engine工程师关于Bigtable/Megastore replication的文章。当时留下了很多疑问，比如：为什么Google Bigtable 是按照column family级别而不是按行执行replication的？今天重新思考了Bigtable replication问题，有如下体会：

1. Bigtable/GFS的设计属于分层设计，和文件系统/数据库分层设计原理一致，通过系统隔离解决工程上的问题。这种分层设计带来了两个问题，一个是性能问题，另外一个就是Replication问题。由于存储节点和服务节点可能不在一台机器，理论上总是存在性能问题，这就要求我们在加载/迁移Bigtable子表(Bigtable tablet)的时候考虑本地化因素；另外，GFS有自己的replication机制保证存储的可靠性，Bigtable通过分离服务节点和存储节点获得了很大的灵活性，且Bigtable的宕机恢复时间可以做到很短。对于很多对实时性要求不是特别高的应用Bigtable由于服务节点同时只有一个，既节约资源又避免了单点问题。然后，Bigtable tablet服务过于灵活导致replication做起来极其困难。比如，tablet的分裂和合并机制导致多个tablet(一个只写，其它只读)服务同一段范围的数据变得几乎不可能。

2. Google replication分为两种机制，基于客户端和基于Tablet Server。分述如下：

2-1). 基于客户端的replication。这种机制比较简单，实现如下：客户端读/写操作均为异步操作，每个写操作都尝试写两个Bigtable集群，任何一个写成功就返回用户，客户端维护一个retry list，不断重试失败的写操作。读操作发到两个集群，任何一个集群读取成功均可。然后，这样做有两个问题：

    a. 客户端不可靠，可能因为各种问题，包括程序问题退出，retry list丢失导致两个集群的数据不一致；

    b. 多个客户端并发操作时无法保证顺序性。集群A收到的写操作可能是"DEL item; PUT item"；集群B的可能是"PUT item; DEL item"。

2-2). 基于Tablet Server的replication。这种机制实现较为复杂，目的是为了保证读服务，写操作的延时仍然可能比较长。两个集群，一个为主集群，提供读/写服务；一个为slave集群，提供只读服务，两个集群维持最终一致性。对于一般的读操作，尽量读取主集群，如果主集群不可以访问则读取slave集群；对于写操作，首先将写操作提交到主集群的Tablet Server，主集群的Tablet Server维护slave集群的元数据信息，并维护一个后台线程不断地将积攒的用户表格写操作提交到slave集群进行日志回放(group commit)。对于一般的tablet迁移，操作逻辑和Bigtable论文中的完全一致；主集群如果发生了机器宕机，则除了回放commit log外，还需要完成宕机的Tablet Server遗留的后台备份任务。之所以要按照column family级别而不是按行复制，是为了提高压缩率从而提高备份效率。如果主集群写操作日志的压缩率大于备份数据的压缩率，则可能出现备份不及时，待备份数据越来越多的问题。

假设集群A为主集群，集群B是集群A的备份，集群切换时先停止集群A的写服务，将集群A余下的备份任务备份到集群B后切换到集群B；如果集群A不可访问的时间不可预知，可以选择直接切换到集群B，这样会带来一致性问题。且由于Bigtable是按列复制的，最后写入的一些行的事务性无法保证。不过由于写操作数据还是保存在集群A的，所以用户可以知道丢了哪些数据，很多应用可以通过重新执行A集群遗留的写操作进行灾难恢复。Google的App Engine也提供了这种查询及重做丢失的写操作的工具。

想法不成熟，有问题联系：knuthocean@163.com

负载平衡策略

Dynamo的负载平衡取决于如何给每台机器分配虚拟节点号。由于集群环境的异构性，每台物理机器包含多个虚拟节点。一般有如下两种分配节点号的方法：

1. 随机分配。每台物理节点加入时根据其配置情况随机分配S个Token(节点号)。这种方法的负载平衡效果还是不错的，因为自然界的数据大致是比较随机的，虽然可能出现某段范围的数据特别多的情况（如baidu, sina等域名下的网页特别多），但是只要切分足够细，即S足够大，负载还是比较均衡的。这个方法的问题是可控性较差，新节点加入/离开系统时，集群中的原有节点都需要扫描所有的数据从而找出属于新节点的数据，Merkle Tree也需要全部更新；另外，增量归档/备份变得几乎不可能。

2. 数据范围等分+随机分配。为了解决方法1的问题，首先将数据的Hash空间等分为Q = N * S份 (N=机器个数，S=每台机器的虚拟节点数），然后每台机器随机选择S个分割点作为Token。和方法1一样，这种方法的负载也比较均衡，且每台机器都可以对属于每个范围的数据维护一个逻辑上的Merkle Tree，新节点加入/离开时只需扫描部分数据进行同步，并更新这部分数据对应的逻辑Merkle Tree，增量归档也变得简单。该方法的一个问题是对机器规模需要做出比较合适的预估，随着业务量的增长，可能需要重新对数据进行划分。

不管采用哪种方法，Dynamo的负载平衡效果还是值得担心的。

客户端缓存及前后台任务资源分配

客户端缓存机器信息可以减少一次在DHT中定位目标机器的网络交互。由于客户端数量不可控，这里缓存采用客户端pull的方式更新，Dynamo中每隔10s或者读/写操作发现缓存信息不一致时客户端更新一次缓存信息。

Dynamo中同步操作、写操作重试等后台任务较多，为了不影响正常的读写服务，需要对后台任务能够使用的资源做出限制。Dynamo中维护一个资源授权系统。该系统将整个机器的资源切分成多个片，监控60s内的磁盘读写响应时间，事务超时时间及锁冲突情况，根据监控信息算出机器负载从而动态调整分配给后台任务的资源片个数。

Dynamo的优点

1. 设计简单，组合利用P2P的各种成熟技术，模块划分好，代码复用程度高。

2. 分布式逻辑与单机存储引擎逻辑基本隔离。很多公司有自己的单机存储引擎，可以借鉴Dynamo的思想加入分布式功能。

3. NWR策略可以根据应用自由调整，这个思想已经被Google借鉴到其下一代存储基础设施中。

4. 设计上天然没有单点，且基本没有对系统时钟一致性的依赖。而在Google的单Master设计中，Master是单点，需要引入复杂的分布式锁机制来解决，且Lease机制需要对机器间时钟同步做出假设。

Dynamo的缺陷

1. 负载平衡相比单Master设计较不可控；负载平衡策略一般需要预估机器规模，不能无缝地适应业务动态增长。

2. 系统的扩展性较差。由于增加机器需要给机器分配DHT算法所需的编号，操作复杂度较高，且每台机器存储了整个集群的机器信息及数据文件的Merkle Tree信息，机器最大规模只能到几千台。

3. 数据一致性问题。多个客户端的写操作有顺序问题，而在GFS中可以通过只允许Append操作得到一个比较好的一致性模型。

4. 数据存储不是有序，无法执行Mapreduce；Mapreduce是目前允许机器故障，具有强扩展性的最好的并行计算模型，且有开源的Hadoop可以直接使用，Dynamo由于数据存储依赖Hash无法直接执行Mapreduce任务。