分布式环境下，互斥性与幂等性问题，分析与解决思路
时间：2021-11-11 13:53:34
手机看文章
扫描二维码
随时随地手机看文章
[导读]随着互联网信息技术的飞速发展，数据量不断增大，业务逻辑也日趋复杂，对系统的高并发访问、海量数据处理的场景也越来越多。如何用较低成本实现系统的高可用、易伸缩、可扩展等目标就显得越发重要。为了解决这一系列问题，系统架构也在不断演进。传统的集中式系统已经逐渐无法满足要求，分布式系统被使...
随着互联网信息技术的飞速发展，数据量不断增大，业务逻辑也日趋复杂，对系统的高并发访问、海量数据处理的场景也越来越多。如何用较低成本实现系统的高可用、易伸缩、可扩展等目标就显得越发重要。
为了解决这一系列问题，系统架构也在不断演进。传统的集中式系统已经逐渐无法满足要求，分布式系统被使用在更多的场景中。分布式系统由独立的服务器通过网络松散耦合组成。在这个系统中每个服务器都是一台独立的主机，服务器之间通过内部网络连接。分布式系统有以下几个特点：
然而，在分布式系统中，其环境的复杂度、网络的不确定性会造成诸如时钟不一致、“拜占庭将军问题”（Byzantine failure）等。存在于集中式系统中的机器宕机、消息丢失等问题也会在分布式环境中变得更加复杂。基于分布式系统的这些特征，有两种问题逐渐成为了分布式环境中需要重点关注和解决的典型问题：
今天我们就针对这两个问题来进行分析。
- 互斥性问题 -
先看两个常见的例子：例1：某服务记录关键数据X，当前值为100。A请求需要将X增加200；同时，B请求需要将X减100。在理想的情况下，A先读取到X=100，然后X增加200，最后写入X=300。B请求接着从读取X=300，减少100，最后写入X=200。然而在真实情况下，如果不做任何处理，则可能会出现：A和B同时读取到X=100；A写入之前B读取到X；B比A先写入等情况。例2：某服务提供一组任务，A请求随机从任务组中获取一个任务；B请求随机从任务组中获取一个任务。在理想的情况下，A从任务组中挑选一个任务，任务组删除该任务，B从剩下的的任务中再挑一个，任务组删除该任务。同样的，在真实情况下，如果不做任何处理，可能会出现A和B挑中了同一个任务的情况。以上的两个例子，都存在操作互斥性的问题。互斥性问题用通俗的话来讲，就是对共享资源的抢占问题。如果不同的请求对同一个或者同一组资源读取并修改时，无法保证按序执行，无法保证一个操作的原子性，那么就很有可能会出现预期外的情况。因此操作的互斥性问题，也可以理解为一个需要保证时序性、原子性的问题。在传统的基于数据库的架构中，对于数据的抢占问题往往是通过数据库事务（ACID）来保证的。在分布式环境中，出于对性能以及一致性敏感度的要求，使得分布式锁成为了一种比较常见而高效的解决方案。事实上，操作互斥性问题也并非分布式环境所独有，在传统的多线程、多进程情况下已经有了很好的解决方案。因此在研究分布式锁之前，我们先来分析下这两种情况的解决方案，以期能够对分布式锁的解决方案提供一些实现思路。
- 多线程解决方案及原理 -
《Thinking in Java》书中写到：
- 原理 -
ReentrantLockReentrantLock主要利用CAS CLH队列来实现。它支持公平锁和非公平锁，两者的实现类似。
ReentrantLock的基本实现可以概括为：先通过CAS尝试获取锁。如果此时已经有线程占据了锁，那就加入CLH队列并且被挂起。当锁被释放之后，排在CLH队列队首的线程会被唤醒，然后CAS再次尝试获取锁。在这个时候，如果：
下面分析下两个片段：
在尝试获取锁的时候，会先调用上面的方法。如果状态为0，则表明此时无人占有锁。此时尝试进行set，一旦成功，则成功占有锁。如果状态不为0，再判断是否是当前线程获取到锁。如果是的话，将状态 1，因为此时就是当前线程，所以不用CAS。这也就是可重入锁的实现原理。
为了解决这一系列问题，系统架构也在不断演进。传统的集中式系统已经逐渐无法满足要求，分布式系统被使用在更多的场景中。分布式系统由独立的服务器通过网络松散耦合组成。在这个系统中每个服务器都是一台独立的主机，服务器之间通过内部网络连接。分布式系统有以下几个特点：
- 可扩展性：可通过横向水平扩展提高系统的性能和吞吐量。
- 高可靠性：高容错，即使系统中一台或几台故障，系统仍可提供服务。
- 高并发性：各机器并行独立处理和计算。
- 廉价高效：多台小型机而非单台高性能机。
然而，在分布式系统中，其环境的复杂度、网络的不确定性会造成诸如时钟不一致、“拜占庭将军问题”（Byzantine failure）等。存在于集中式系统中的机器宕机、消息丢失等问题也会在分布式环境中变得更加复杂。基于分布式系统的这些特征，有两种问题逐渐成为了分布式环境中需要重点关注和解决的典型问题：
- 互斥性问题。
- 幂等性问题。
今天我们就针对这两个问题来进行分析。
- 互斥性问题 -
先看两个常见的例子：例1：某服务记录关键数据X，当前值为100。A请求需要将X增加200；同时，B请求需要将X减100。在理想的情况下，A先读取到X=100，然后X增加200，最后写入X=300。B请求接着从读取X=300，减少100，最后写入X=200。然而在真实情况下，如果不做任何处理，则可能会出现：A和B同时读取到X=100；A写入之前B读取到X；B比A先写入等情况。例2：某服务提供一组任务，A请求随机从任务组中获取一个任务；B请求随机从任务组中获取一个任务。在理想的情况下，A从任务组中挑选一个任务，任务组删除该任务，B从剩下的的任务中再挑一个，任务组删除该任务。同样的，在真实情况下，如果不做任何处理，可能会出现A和B挑中了同一个任务的情况。以上的两个例子，都存在操作互斥性的问题。互斥性问题用通俗的话来讲，就是对共享资源的抢占问题。如果不同的请求对同一个或者同一组资源读取并修改时，无法保证按序执行，无法保证一个操作的原子性，那么就很有可能会出现预期外的情况。因此操作的互斥性问题，也可以理解为一个需要保证时序性、原子性的问题。在传统的基于数据库的架构中，对于数据的抢占问题往往是通过数据库事务（ACID）来保证的。在分布式环境中，出于对性能以及一致性敏感度的要求，使得分布式锁成为了一种比较常见而高效的解决方案。事实上，操作互斥性问题也并非分布式环境所独有，在传统的多线程、多进程情况下已经有了很好的解决方案。因此在研究分布式锁之前，我们先来分析下这两种情况的解决方案，以期能够对分布式锁的解决方案提供一些实现思路。
- 多线程解决方案及原理 -
《Thinking in Java》书中写到：
基本上所有的并发模式在解决线程冲突问题的时候，都是采用序列化访问共享资源的方案。在多线程环境中，线程之间因为公用一些存储空间，冲突问题时有发生。解决冲突问题最普遍的方式就是用互斥锁把该资源或对该资源的操作保护起来。Java JDK中提供了两种互斥锁Lock和synchronized。不同的线程之间对同一资源进行抢占，该资源通常表现为某个类的普通成员变量。因此，利用ReentrantLock或者synchronized将共享的变量及其操作锁住，即可基本解决资源抢占的问题。下面来简单聊一聊两者的实现原理。
- 原理 -
ReentrantLockReentrantLock主要利用CAS CLH队列来实现。它支持公平锁和非公平锁，两者的实现类似。
- CAS：Compare and Swap，比较并交换。CAS有3个操作数：内存值V、预期值A、要修改的新值B。当且仅当预期值A和内存值V相同时，将内存值V修改为B，否则什么都不做。该操作是一个原子操作，被广泛的应用在Java的底层实现中。在Java中，CAS主要是由sun.misc.Unsafe这个类通过JNI调用CPU底层指令实现。
- CLH队列：带头结点的双向非循环链表(如下图所示)：
ReentrantLock的基本实现可以概括为：先通过CAS尝试获取锁。如果此时已经有线程占据了锁，那就加入CLH队列并且被挂起。当锁被释放之后，排在CLH队列队首的线程会被唤醒，然后CAS再次尝试获取锁。在这个时候，如果：- 非公平锁：如果同时还有另一个线程进来尝试获取，那么有可能会让这个线程抢先获取；
- 公平锁：如果同时还有另一个线程进来尝试获取，当它发现自己不是在队首的话，就会排到队尾，由队首的线程获取到锁。
下面分析下两个片段：
final boolean nonfairTryAcquire(int acquires) {
final Thread current = Thread.currentThread();
int c = getState();
if (c == 0) {
if (compareAndSetState(0, acquires)) {
setExclusiveOwnerThread(current);
return true;
}
}
else if (current == getExclusiveOwnerThread()) {
int nextc = c acquires;
if (nextc < 0) // overflow
throw new Error("Maximum lock count exceeded");
setState(nextc);
return true;
}
return false;
}在尝试获取锁的时候，会先调用上面的方法。如果状态为0，则表明此时无人占有锁。此时尝试进行set，一旦成功，则成功占有锁。如果状态不为0，再判断是否是当前线程获取到锁。如果是的话，将状态 1，因为此时就是当前线程，所以不用CAS。这也就是可重入锁的实现原理。
final boolean acquireQueued(final Node node, int arg) {
boolean failed = true;
try {
boolean interrupted = false;
for (;;) {
final Node p = node.predecessor();
if (p == head 
下载文档
