In-depth analysis of NOSQL database distributed algorithms

< /p>

Data consistency

As we all know, distributed systems often encounter network isolation or delay. In this case, the isolated part is not available, so It is impossible to maintain high availability without sacrificing consistency. This fact is often referred to as the “CAP theory”. However, consistency is a very expensive thing in a distributed system, so it is often necessary to make some concessions, not only for availability, but also for many trade-offs. In order to study these trade-offs, we noticed that the consistency problem of distributed systems is caused by data isolation and replication, so we will start by studying the characteristics of replication:

• Availability. In the case of network isolation, the remaining part can still handle read and write requests.
• Latency in reading and writing. Read and write requests can be processed in a short time.
• reading and writing malleability. The pressure of reading and writing can be balanced by multiple nodes.
• Fault tolerance. The processing of read and write requests does not depend on any particular node.
• data persistence. A node failure under certain conditions will not cause data loss.
• Consistency. Consistency is much more complicated than the previous features, and we need to discuss several different viewpoints in detail. But we will not involve too much consistency theory and concurrency model, because this is beyond the scope of this article, I will only use a streamlined system composed of some simple features.

• read and write consistency. From a read-write point of view, the basic goal of a database is to make the time for replicas to be as short as possible (that is, the time for updates to be transmitted to all replicas) to ensure eventual consistency. In addition to this weaker guarantee, there are some stronger consistency features:

• read-after-write consistency. The effect of the write operation on the data item X can always be seen by the subsequent read operation on X.
• Read after reading consistency. After a read operation on data item X, subsequent read operations on X should return the same or newer value as the first return value.

< /p>

• Write consistency. Partitioned databases often have write conflicts. The database should be able to handle this conflict and ensure that multiple write requests will not be processed by different partitions. In this regard, the database provides several different consistency models:

< li style="margin:0px; padding:0px; list-style:disc">atomic writing. If the database provides an API, a write operation can only be a single atomic assignment. The way to avoid write conflicts is to find the “latest version” of each data. This enables all nodes to obtain the same version at the end of the update, regardless of the order of the update. Network failures and delays often result in inconsistent update order of each node. The data version can be represented by a timestamp or a value specified by the user. Cassandra uses this method.
• atomized read-modify- Write. Applications sometimes need to perform read-modify-write sequential operations instead of separate atomic write operations. If two clients read the data of the same version, modify and write the modified data back, according to the atomic write model, the update that is later in time will overwrite the previous one. This behavior is incorrect in some cases (for example, two clients add new values ​​to the same list value). The database provides at least two solutions:

< /p>

• Conflict prevention. Read-modify-write can be regarded as a special case transaction, so distributed locks or a consistent protocol such as PAXOS can solve this problem. This technology supports atomic read-rewrite semantics and transactions with arbitrary isolation levels. Another method is to avoid distributed concurrent write operations and route all write operations to a specific data item to a single node (it can be a global master node or a partition master node). In order to avoid conflicts, the database must sacrifice availability in the case of network isolation. This method is commonly used in many systems that provide strong consistency guarantees (such as most relational databases, HBase, MongoDB).
• Conflict detection. The database tracks the conflicts of concurrent updates, and chooses to roll back one of them or maintain both versions to the client to resolve. Concurrent updates are usually tracked with a vector clock (this is an optimistic lock), or to maintain a complete version history. This method is used in Riak, Voldemort, CouchDB.

Now let us take a closer look at commonly used replication techniques and classify them according to the characteristics described. The first picture depicts the logical relationship between different technologies and the trade-off coordinates between the consistency, scalability, availability, and latency of the system. The second figure depicts each technology in detail.

The copy factor is 4. The read-write coordinator can be an external client or an internal agent node.

We will go through all the techniques from weak to strong based on consistency:

(A , Anti-entropy) The consistency is the weakest, based on the following strategy. When writing, select any node to update. When reading, if the new data has not been transmitted to the node that reads through the anti-entropy protocol in the background, then the old data is still read. (The anti-entropy protocol will be described in detail in the next section). The main features of this method are:

• The high propagation delay makes it not easy to use in data synchronization, so the typical usage is only as an auxiliary function To detect and fix unplanned inconsistencies. Cassandra uses an anti-entropy algorithm to transfer database topology and other metadata information between nodes.

• Weak consistency guarantee: even in the absence of failure , There will also be write conflicts and inconsistencies in reading and writing.

• High availability and robustness under network isolation. Instead of updating one by one with asynchronous batch processing, this makes for excellent performance.

• The durability guarantee is weak because the new data initially only has a single copy.

(B) An improvement to the above model is to asynchronously send updates to all available nodes when any node receives an update data request. This is also considered to be directed anti-entropy.

< blockquote style="margin-top:0px; margin-right:0px; margin-bottom:1.5em; padding:0px 0px 0px 1em; list-style:none; border-left-width:4px; border-left-style: solid; border-left-color:rgb(221,221,221); color:rgb(119,119,119); margin-left:3em!important">

• Compared with pure anti-entropy, this approach only uses a little The performance sacrifice greatly improves consistency. However, formal consistency and durability remain unchanged.

• If some nodes are unavailable at the time due to network failure or node failure, the update will eventually be passed to the node through the anti-entropy propagation process.

(C) In the former mode, the use of prompt handover technology can better handle the failure of a certain node. The expected update of the failed node is recorded on the additional agent node, and it is indicated that the update will be delivered to the node once the characteristic node is available. This improves consistency and reduces replication convergence time.

(D, one-time reading Write) Because the responsible node that prompts the handover may also become invalid before the update is passed out. In this case, it is necessary to ensure consistency through the so-called read repair. Each read operation starts an asynchronous process, requesting a data digest (like a signature or hash) from all nodes that store this data. If the digest returned by each node is found to be inconsistent, the data version on each node will be unified. . We use one-time reading and writing to name the technologies that combine A, B, C, and D-none of them provide strict consistency guarantees, but as a self-contained method, it can be used in practice.

(E, read several writes Several) The above strategy is a heuristic enhancement that reduces the replication convergence time. In order to ensure stronger consistency, usability must be sacrificed to ensure a certain amount of read-write overlap. The usual practice is to write W copies instead of one at the same time, and read R copies when reading.

• First, you can configure the number of write copies W>1.

• Secondly, because R+W>N, there must be overlap between the node that is written and the node that is read, so multiple copies of data are read At least one of them is relatively new data (W=2, R=3, N=4 in the above figure). This can ensure consistency (read and write consistency for a single user) when read and write requests are performed in sequence (read after writing is executed), but cannot guarantee global read consistency. Using the example shown in the figure below, R=2, W=2, N=3, because the update of the two copies of the write operation is non-transactional. When the update is not completed, the read may read both of them. Old value or one new and one old:

• For a certain read delay requirement, setting different values ​​of R and W can adjust the write delay and durability, and vice versa.
• if W< =N/2, multiple concurrent writes will be written to several different nodes (for example, write operation A writes N/2 before writing, and B writes N/2 after writing). Setting W>N/2 can ensure that conflicts are detected in time when atomic reads and writes conform to the rollback model.
• Strictly speaking, although this model can tolerate the failure of individual nodes, it is not good for network isolation fault tolerance. In practice, the “approximate quantity pass” method is often used to improve the usability in certain situations by sacrificing consistency.

(F, Read all and write several) The read consistency problem can be alleviated by accessing all copies (read data or check summary) when reading data. This ensures that as long as the data on at least one node is updated with new data, it can be seen by readers. But in the case of network isolation, this guarantee will not work.

(G, master-slave) This technique is often used to provide atomic writes or conflict detection persistence levels of read and write changes. In order to achieve the level of conflict prevention, a centralized management method or lock must be used. The simplest strategy is to use master-slave asynchronous replication. All write operations for a specific data item are routed to a central node and executed sequentially. In this case, the master node will become a bottleneck, so the data must be divided into independent slices (different slices have different masters) in order to provide scalability.

(H, Transactional Read Quorum Write Quorum and Read One Write All) The method of updating multiple copies can avoid write conflicts by using transaction control technology. The well-known method is to use a two-phase submission protocol. But the two-phase commit is not completely reliable, because the failure of the coordinator may cause resource congestion. The PAXOS submission protocol is a more reliable choice, but at the expense of performance. A small step forward on this basis is to read one copy and write all copies. This method puts the updates of all copies in one transaction. It provides strong fault tolerance and consistency but will lose some performance and availability.

Some trade-offs in the above analysis need to be emphasized.

• consistency and usability. The rigorous trade-off has been given by the CAP theory. In the case of network isolation, the database must either centralize the data or accept the risk of data loss.
• Consistency and scalability. It can be seen that even if the read-write consistency guarantee reduces the scalability of the replica set, only in the atomic write model can write conflicts be handled in a relatively scalable manner. The atomic read/write model avoids conflicts by adding temporary global locks to the data. This shows that the dependence between data or operations, even in a small range or a short time, will damage the scalability. Therefore, carefully designing the data model and storing the data fragments separately is very important for scalability.
• Consistency and delay. As mentioned above, when the database needs to provide strong consistency or durability, it should be biased towards reading and writing all replica technologies. However, it is obvious that consistency is inversely proportional to request latency, so using several copy technologies is a more acceptable method.
• Failover and consistency/scalability/latency . What is interesting is that the conflict between fault tolerance and consistency, scalability, and delay is not severe. By reasonably giving up some performance and consistency, the cluster can tolerate up to node failures. This compromise is evident in the difference between the two-phase submission and the PAXOS protocol. Another example of this trade-off is to add specific consistency guarantees, such as “reading and writing” using a strict session process, but this adds to the complexity of failover.

< /p>

anti-entropy protocol, rumor propagation algorithm

Let’s start with the following scenario:

There are many nodes, and each piece of data will have a copy on some of the nodes. Each node can process update requests separately, and each node periodically synchronizes the state with other nodes, so that after a period of time, all copies will tend to be consistent. How does the synchronization process work? When does the synchronization start? How to choose the synchronization object? How to exchange data? We assume that the two nodes always overwrite the old data with the newer version of the data or both versions are reserved for processing by the application layer.

This problem is common in data consistency maintenance and Scenarios such as cluster state synchronization (such as the propagation of cluster member information). Although the introduction of a coordinator who monitors the database and develops a synchronization plan can solve this problem, a decentralized database can provide better fault tolerance. The main method of decentralization is to use a well-designed infection protocol, which is relatively simple, but provides a good convergence time, and can tolerate any node failure and network isolation. Although there are many types of contagion algorithms, we only focus on the anti-entropy protocol because NoSQL databases use it.

The anti-entropy protocol assumes that the synchronization will follow a fixed The schedule is executed, and each node chooses another node to exchange data randomly or according to a certain rule to eliminate differences. There are three anti-style anti-entropy protocols: push, pull, and mix. The principle of the push protocol is to simply select a random node and send the data state to it. It is obviously stupid to push all the data in a real application, so nodes generally work in the manner shown in the figure below.

Node A is initiated as synchronization The person prepares a data summary, which contains the fingerprint of the data on A. After node B receives the summary, it compares the data in the summary with the local data, and returns a summary of the data differences to A. Finally, A sends an update to B, and B updates the data again. The pull mode and hybrid mode are similar to this, as shown in the figure above.

The anti-entropy protocol provides sufficient convergence Time and scalability. The figure below shows the results of a simulation that propagates an update in a cluster of 100 nodes. In each iteration, each node only contacts a randomly selected peer node.

It can be seen that the convergence of the pull method is better than that of the push method, which can be proved theoretically. Moreover, there is a problem of “convergent tail” in the push method. After many iterations, although almost all nodes have been traversed, a few of them are still unaffected. Compared with the simple push and pull method, the hybrid method is more efficient, so this method is usually used in practical applications. Anti-entropy is scalable because the average conversion time grows as a logarithmic function of the cluster size.

Although these techniques seem simple, they still There are many studies focusing on the performance of anti-entropy protocols under different constraints. One of them uses network topology to replace random selection with a more efficient structure. Under the condition of limited network bandwidth, adjust the transmission rate or use advanced rules to select the data to be synchronized. The summary calculation also faces challenges. The database maintains a recently updated log to facilitate summary calculations.

Eventually Consistent Data Types

In the previous section we assumed that two nodes always merge their data versions. However, it is not easy to resolve update conflicts, and it is unexpectedly difficult for all copies to eventually reach a semantically correct value. A well-known example is that deleted entries in the Amazon Dynamo database can be reproduced.

Let’s assume an example to illustrate this problem: The database maintains a logical global counter, and each node can increase or decrease the count. Although each node can maintain its own value locally, these local counts cannot be combined by simple addition and subtraction. Suppose an example: there are three nodes A, B, and C, and each node performs an addition operation. If A gets a value from B and adds it to the local copy, then C gets the value from B, and then C gets the value from A, then the final value of C is 4, which is wrong. The solution to this problem is to use a data structure similar to a vector clock to maintain a pair of counters for each node:

[js] view plain copy

1. 1classCounter{
2. 2 int []plus
3. 3 int[]minus
4. 4     int  NODE_ID   
5. 5   
6. 6     increment() {   
7. 7         plus[NODE_ID]++   
8. 8     }   
9. 9   
10. 10    decrement() {   
11. 11        minus[NODE_ID]++   
12. 12    }   
13. 13   
14. 14    get() {   
15. 15        return sum(plus) – sum(minus)   
16. 16    }   
17. 17   
18. 18    merge(Counter other) {   
19. 19        for i in 1..MAX_ID {   
20. 20            plus[i] = max(plus[i], other.plus[i])   
21. 21            minus[i] = max(minus[i], other.minus[i])   
22. 22        }  
23. 23    }   
24. 24 }  

Cassandra用类似的方法计数。利用基于状态的或是基于操作的复制理论也可以设计出更复杂的最终一致的数据结构。例如，中就提及了一系列这样的数据结构，包括：

• 计数器（加减操作）
• 集合（添加和移除操作）
• 图（增加边或顶点，移除边或顶点）
• 列表（插入某位置或者移除某位置）

最终一致数据类型的功能通常是有限的，还会带来额外的性能开销。

数据放置

这部分主要关注控制在分布式数据库中放置数据的算法。这些算法负责把数据项映射到合适的物理节点上，在节点间迁移数据以及像内存这样的资源的全局调配。

均衡数据

我们还是从一个简单的协议开始，它可以提供集群节点间无缝的数据迁移。这常发生于像集群扩容（加入新节点），故障转移（一些节点宕机）或是均衡数据（数据在节点间的分布不均衡）这样的场景。如下图A中所描绘的场景 – 有三个节点，数据随便分布在三个节点上（假设数据都是key-value型）。

如果数据库不支持数据内部均衡，就要在每个节点上发布数据库实例，如上面图B所示。这需要手动进行集群扩展，停掉要迁移的数据库实例，把它转移到新节点上，再在新节点上启动，如图C所示。尽管数据库能够监控到每一条记录，包括MongoDB, Oracle Coherence, 和还在开发中的 Redis Cluster 在内的许多系统仍然使用的是自动均衡技术。也即，将数据分片并把每个数据分片作为迁移的最小单位，这是基于效率的考虑。很明显分片数会比节点数多，数据分片可以在各节点间平均分布。按照一种简单的协议即可实现无缝数据迁移，这个协议可以在迁移数据分片的时候重定向客户的数据迁出节点和迁入节点。下图描绘了一个Redis Cluster中实现的get（key）逻辑的状态机。

假定每个节点都知道集群拓扑，能够把任意key映射到相应的数据分片，把数据分片映射到节点。如果节点判断被请求的key属于本地分片，就会在本地查找（上图中上面的方框）。假如节点判断请求的key属于另一个节点X，他会发送一个永久重定向命令给客户端（上图中下方的方框）。永久重定向意味着客户端可以缓存分片和节点间的映射关系。如果分片迁移正在进行，迁出节点和迁入节点会标记相应的分片并且将分片的数据加锁逐条加锁然后开始移动。迁出节点首先会在本地查找key，如果没有找到，重定向客户端到迁入节点，假如key已经迁移完毕的话。这种重定向是一次性的，并且不能被缓存。迁入节点在本地处理重定向，但定期查询在迁移还没完成前被永久重定向。

动态环境中的数据分片和复制

我们关注的另一个问题是怎么把记录映射到物理节点。比较直接的方法是用一张表来记录每个范围的key与节点的映射关系，一个范围的key对应到一个节点，或者用key的hash值与节点数取模得到的值作为节点ID。但是hash取模的方法在集群发生更改的情况下就不是很好用，因为增加或者减少节点都会引起集群内的数据彻底重排。导致很难进行复制和故障恢复。

有许多方法在复制和故障恢复的角度进行了增强。最著名的就是一致性hash。网上已经有很多关于一致性hash的介绍了，所以在这里我只提供一个基本介绍，仅仅为了文章内容的完整性。下图描绘了一致性hash的基本原理：

一致性hash从根本上来讲是一个键值映射结构 – 它把键（通常是hash过的）映射到物理节点。键经过hash之后的取值空间是一个有序的定长二进制字符串，很显然每个在此范围内的键都会被映射到图A中A、B、C三个节点中的某一个。为了副本复制，将取值空间闭合成一个环，沿环顺时针前行直到所有副本都被映射到合适的节点上，如图B所示。换句话说，Y将被定位在节点B上，因为它在B的范围内，第一个副本应该放置在C，第二个副本放置在A，以此类推。

这种结构的好处体现在增加或减少一个节点的时候，因为它只会引起临接区域的数据重新均衡。如图C所示，节点D的加入只会对数据项X产生影响而对Y无影响。同样，移除节点B（或者B失效）只会影响Y和X的副本，而不会对X自身造成影响。但是，这种做法在带来好处的同时也有弱点，那就是重新均衡的负担都由邻节点承受了，它们将移动大量的数据。通过将每个节点映射到多个范围而不是一个范围可以一定程度上减轻这个问题带来的不利影响，如图D所示。这是一个折中，它避免了重新均衡数据时负载过于集中，但是与基于模块的映射相比，保持了总均衡数量适当降低。

给大规模的集群维护一个完整连贯的hash环很不容易。对于相对小一点的数据库集群就不会有问题，研究如何在对等网络中将数据放置与网络路由结合起来很有意思。一个比较好的例子是Chord算法，它使环的完整性让步于单个节点的查找效率。 Chord算法也使用了环映射键到节点的理念，在这方面和一致性hash很相似。不同的是，一个特定节点维护一个短列表，列表中的节点在环上的逻辑位置是指数增长的（如下图）。这使得可以使用二分搜索只需要几次网络跳跃就可以定位一个键。

这张图画的是一个由16个节点组成的集群，描绘了节点A是如何查找放在节点D上的key的。 (A) 描绘了路由，(B) 描绘了环针对节点A、B、C的局部图像。在参考资料中有更多关于分散式系统中的数据复制的内容。

按照多个属性的数据分片

当只需要通过主键来访问数据的时候，一致性hash的数据放置策略很有效，但是当需要按照多个属性来查询的时候事情就会复杂得多。一种简单的做法（MongoDB使用的）是用主键来分布数据而不考虑其他属性。这样做的结果是依据主键的查询可以被路由到接个合适的节点上，但是对其他查询的处理就要遍历集群的所有节点。查询效率的不均衡造成下面的问题：

有一个数据集，其中的每条数据都有若干属性和相应的值。是否有一种数据分布策略能够使得限定了任意多个属性的查询会被交予尽量少的几个节点执行？

HyperDex数据库提供了一种解决方案。基本思想是把每个属性视作多维空间中的一个轴，将空间中的区域映射到物理节点上。一次查询会被对应到一个由空间中多个相邻区域组成的超平面，所以只有这些区域与该查询有关。让我们看看参考资料中的一个例子：

每一条数据都是一条用户信息，有三个属性First Name 、Last Name 和Phone Number。这些属性被视作一个三维空间，可行的数据分布策略是将每个象限映射到一个物理节点。像“First Name = John”这样的查询对应到一个贯穿4个象限的平面，也即只有4个节点会参与处理此次查询。有两个属性限制的查询对应于一条贯穿两个象限的直线，如上图所示，因此只有2个节点会参与处理。

这个方法的问题是空间象限会呈属性数的指数函数增长。结果就会是，只有几个属性限制的查询会投射到许多个空间区域，也即许多台服务器。将一个属性较多的数据项拆分成几个属性相对较少的子项，并将每个子项都映射到一个独立的子空间，而不是将整条数据映射到一个多维空间，这样可以一定程度上缓解这个问题：

这样能够提供更好的查询到节点的映射，但是增加了集群协调的复杂度，因为这种情况下一条数据会散布在多个独立的子空间，而每个子空间都对应各自的若干个物理节点，数据更新时就必须考虑事务问题。

钝化副本

有的应用有很强的随机读取要求，这就需要把所有数据放在内存里。在这种情况下，将数据分片并把每个分片主从复制通常需要两倍以上的内存，因为每个数据都要在主节点和从节点上各有一份。为了在主节点失效的时候起到代替作用，从节点上的内存大小应该和主节点一样。如果系统能够容忍节点失效的时候出现短暂中断或性能下降，也可以不要分片。

下面的图描绘了4个节点上的16个分片，每个分片都有一份在内存里，副本存在硬盘上：

灰色箭头突出了节点2上的分片复制。其他节点上的分片也是同样复制的。红色箭头描绘了在节点2失效的情况下副本怎样加载进内存。集群内副本的均匀分布使得只需要预留很少的内存就可以存放节点失效情况下激活的副本。在上面的图里，集群只预留了1/3的内存就可以承受单个节点的失效。特别要指出的是副本的激活（从硬盘加载入内存）会花费一些时间，这会造成短时间的性能下降或者正在恢复中的那部分数据服务中断。

系统协调

在这部分我们将讨论与系统协调相关的两种技术。分布式协调是一个比较大的领域，数十年以来有很多人对此进行了深入的研究。这篇文章里只涉及两种已经投入实用的技术。关于分布式锁，consensus协议以及其他一些基础技术的内容可以在很多书或者网络资源中找到。

故障检测

故障检测是任何一个拥有容错性的分布式系统的基本功能。实际上所有的故障检测协议都基于心跳通讯机制，原理很简单，被监控的组件定期发送心跳信息给监控进程（或者由监控进程轮询被监控组件），如果有一段时间没有收到心跳信息就被认为失效了。除此之外，真正的分布式系统还要有另外一些功能要求：

• 自适应。 故障检测应该能够应对暂时的网络故障和延迟，以及集群拓扑、负载和带宽的变化。但这有很大难度，因为没有办法去分辨一个长时间没有响应的进程到底是不是真的失效了，因此，故障检测需要权衡故障识别时间（花多长时间才能识别一个真正的故障，也即一个进程失去响应多久之后会被认为是失效）和虚假警报率之间的轻重。这个权衡因子应该能够动态自动调整。
• 灵活性。 乍看上去，故障检测只需要输出一个表明被监控进程是否处于工作状态的布尔值，但在实际应用中这是不够的。我们来看参考资料中的一个类似MapReduce的例子。有一个由一个主节点和若干工作节点组成的分布式应用，主节点维护一个作业列表，并将列表中的作业分配给工作节点。主节点能够区分不同程度的失败。如果主节点怀疑某个工作节点挂了，他就不会再给这个节点分配作业。其次，随着时间推移，如果没有收到该节点的心跳信息，主节点就会把运行在这个节点上的作业重新分配给别的节点。最后，主节点确认这个节点已经失效，并释放所有相关资源。
• 可扩展性和健壮性。 失败检测作为一个系统功能应该能够随着系统的扩大而扩展。他应该是健壮和一致的，也即，即使在发生通讯故障的情况下，系统中的所有节点都应该有一个一致的看法（即所有节点都应该知道哪些节点是不可用的，那些节点是可用的，各节点对此的认知不能发生冲突，不能出现一部分节点知道某节点A不可用，而另一部分节点不知道的情况）

所谓的累计失效检测器可以解决前两个问题，Cassandra对它进行了一些修改并应用在产品中。其基本工作流程如下：

• 对于每一个被监控资源，检测器记录心跳信息到达时间Ti。
• 计算在统计预测范围内的到达时间的均值和方差。
• 假定到达时间的分布已知（下图包括一个正态分布的公式），我们可以计算心跳延迟（当前时间t_now和上一次到达时间Tc之间的差值） 的概率，用这个概率来判断是否发生故障。可以使用对数函数来调整它以提高可用性。在这种情况下，输出1意味着判断错误（认为节点失效）的概率是10%，2意味着1%，以此类推。

根据重要程度不同来分层次组织监控区，各区域之间通过谣言传播协议或者中央容错库同步，这样可以满足扩展性的要求，又可以防止心跳信息在网络中泛滥。如下图所示（6个故障检测器组成了两个区域，互相之间通过谣言传播协议或者像ZooKeeper这样的健壮性库来联系）：

协调者竞选

协调者竞选是用于强一致性数据库的一个重要技术。首先，它可以组织主从结构的系统中主节点的故障恢复。其次，在网络隔离的情况下，它可以断开处于少数的那部分节点，以避免写冲突。

Bully 算法是一种相对简单的协调者竞选算法。 MongoDB 用了这个算法来决定副本集中主要的那一个。 Bully 算法的主要思想是集群的每个成员都可以声明它是协调者并通知其他节点。别的节点可以选择接受这个声称或是拒绝并进入协调者竞争。被其他所有节点接受的节点才能成为协调者。节点按照一些属性来判断谁应该胜出。这个属性可以是一个静态ID，也可以是更新的度量像最近一次事务ID（最新的节点会胜出）。

下图的例子展示了bully算法的执行过程。使用静态ID作为度量，ID值更大的节点会胜出：

1. 最初集群有5个节点，节点5是一个公认的协调者。
2. 假设节点5挂了，并且节点2和节点3同时发现了这一情况。两个节点开始竞选并发送竞选消息给ID更大的节点。
3. 节点4淘汰了节点2和3，节点3淘汰了节点2。
4. 这时候节点1察觉了节点5失效并向所有ID更大的节点发送了竞选信息。
5. 节点2、3和4都淘汰了节点1。
6. 节点4发送竞选信息给节点5。
7. 节点5没有响应，所以节点4宣布自己当选并向其他节点通告了这一消息。

协调者竞选过程会统计参与的节点数目并确保集群中至少一半的节点参与了竞选。这确保了在网络隔离的情况下只有一部分节点能选出协调者（假设网络中网络会被分割成多块区域，之间互不联通，协调者竞选的结果必然会在节点数相对比较多的那个区域中选出协调者，当然前提是那个区域中的可用节点多于集群原有节点数的半数。如果集群被隔离成几个区块，而没有一个区块的节点数多于原有节点总数的一半，那就无法选举出协调者，当然这样的情况下也别指望集群能够继续提供服务了）。

原文地址：http://juliashine.com/distributed-algorithms-in-nosql-databases/

作者介绍：可观，资深架构师，擅长WEB开发、服务器端开发、软件开发管理等等。 （审校/刘亚琼）