我们在编程有很多场景使用本地锁和分布式锁，但是是否考虑这些锁的原理是什么？本篇讨论下实现分布式锁的常见办法及他们实现原理。
​

一、使用锁的原则

使用本地锁和分布式锁是为了解决并发导致脏数据的场景，使用锁的最高境界是通过流程设计避免使用锁，锁会牺牲掉系统性能为代价的。

二、常见的分布式锁实现

分布式锁总结：
产品性能：redis>zookeeper>mysql，获取锁成功率：mysql悲观>zk>redis

2.1、mysql实现分布式锁原理

2.1.1、乐观锁实现

• 乐观锁实现方式：

通过在mysql加入version或者updatetime时间戳的方式实现。下面主要介绍新增流程，修改的更简单不做介绍。乐观锁实现出现事务回滚的时候要处理掉新增场景中无效数据。乐观锁其实并没有用到锁的概念，其实他是一个版本同步的技巧实现。
**version-新增实现：**在数据库新增的时候先新增一条数据，然后读取这个数据返回给修改页面，当修改页面提交的时候version对比如果一致说明数据合法，如果不一致说明数据不合法。然后version自增写入数据库。
**updatetime-新增实现：**同version规则一致。

• 乐观锁无法解决的问题

乐观锁无法解决并发多线程问题，这种适合解决并发比较低的场景的数据库数据一致性问题。

2.1.2、悲观锁实现

• **悲观锁实现：**悲观锁实现简单 select * from table for update,悲观锁是所有锁中理论最靠谱的一种，但是性能差。在并发场景不推荐使用;
• **悲观锁的问题：**性能问题、导致锁表

​

2.2、zookeeper实现分布式锁实现及原理

2.2.0、Curator recipes实现哪些锁

• **InterProcessMutex：**可重入、独占锁InterProcessSemaphoreMutex：不可重入、独占锁
• **InterProcessReadWriteLock：**读写锁
• InterProcessSemaphoreV2 ： 共享信号量
• **InterProcessMultiLock：**多重共享锁 （将多个锁作为单个实体管理的容器）

​

2.2.1、zookeeper分布式实现

/**
 * 功能：zk - zk Curator客户端 - 实现分布式锁测试
 * 作者：丁志超
 */  
public class ZkCuratorLock{  
	
	//实例化客户端
	private static RetryPolicy retryPolicy  = new ExponentialBackoffRetry(1000,3);
    private static CuratorFramework client = CuratorFrameworkFactory.builder()
            .connectString("ip:2181")
            .sessionTimeoutMs(3000)
            .connectionTimeoutMs(5000)
            .retryPolicy(retryPolicy)
            .build();
    
    //zk分布式锁创建节点在零时目录zklock下创建
    static String lockPath = "/zklock";
    //实例化分布式锁
    final static InterProcessLock lock = new InterProcessSemaphoreMutex(client, lockPath);
	
	
	public static void main(String[] args) {
		//获取锁
		try {
			lock.acquire();
		} catch (Exception e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
		}finally {
			//释放锁
			try {
				lock.release();
			} catch (Exception e) {
			}
		}
	}
	
} 

<!-- zookeeper curator客户端  -->
<dependency>
  <groupId>org.apache.curator</groupId>
  <artifactId>curator-recipes</artifactId>
  <version>2.12.0</version>
</dependency>

2.2、zookeeper分布式锁实现原理

2.2.1、顺序节点实现原理

• 创建一个锁目录lock
• 线程A获取锁会在lock目录下，创建临时顺序节点
• 获取锁目录下所有的子节点，然后获取比自己小的兄弟节点，如果不存在，则说明当前线程顺序号最小获得锁
• 线程B创建临时节点并获取所有兄弟节点，只监听当前获取锁的节点的变化，节约效率。
• 线程A处理完，删除自己的节点，线程B监听到变更事件，判断自己是最小的节点获得锁

由于节点的临时属性，如果创建znode的那个客户端崩溃了，那么相应的znode会被自动删除。这样就避免了设置过期时间的问题，Curator客户端是基于顺序节点实现的分布式锁。

2.3、zookeeper实现分布式锁面临的问题

2.3.1、是否有脑裂问题？

由于zookeeper是基于Ha方式部署的，其写全部在主节点进行，只要主节点服务正常就不会出现脑裂问题。

2.3.2、心跳超时问题？

由于zookeeper客户端与服务端session保持会话需要心跳机制保证。如果客户端或者服务端GC导致心跳超时，此时零时节点会被zookeeper全部踢掉。zookeeper的零时节点绑定在会话session上面，session不存在客户端创建的零时节点全部会被删除。

2.4、zookeeper Curator实现分布式锁源码分析

//1、Curator锁数据结构
private static class LockData
{
        //当前拥有锁的线程
        final Thread owningThread;
        //当前锁的路径
        final String lockPath;
        //锁计数器
        final AtomicInteger lockCount = new AtomicInteger(1);
 }


//2、这段是Curator的精华 - 获取锁成功后验证 及 获取锁失败后等待
private boolean internalLockLoop(long startMillis, Long millisToWait, String ourPath) throws Exception
    {
        boolean     haveTheLock = false;
        boolean     doDelete = false;
        try
        {
            if ( revocable.get() != null )
            {
                client.getData().usingWatcher(revocableWatcher).forPath(ourPath);
            }

            while ( (client.getState() == CuratorFrameworkState.STARTED) && !haveTheLock )
            {
                List<String>        children = getSortedChildren();
                String              sequenceNodeName = ourPath.substring(basePath.length() + 1); // +1 to include the slash
                
                //成功获取锁
                PredicateResults    predicateResults = driver.getsTheLock(client, children, sequenceNodeName, maxLeases);
                if ( predicateResults.getsTheLock() )
                {
                    haveTheLock = true;
                }
                else
                {
                    //没有获取锁，监听拥有锁节点的变化
                    String  previousSequencePath = basePath + "/" + predicateResults.getPathToWatch();
                    
                    //获取锁失败后等待超时如果不设置超时一直等待
                    synchronized(this)
                    {
                        try 
                        {
                            // use getData() instead of exists() to avoid leaving unneeded watchers which is a type of resource leak
                            client.getData().usingWatcher(watcher).forPath(previousSequencePath);
                            if ( millisToWait != null )
                            {
                                millisToWait -= (System.currentTimeMillis() - startMillis);
                                startMillis = System.currentTimeMillis();
                                if ( millisToWait <= 0 )
                                {
                                    doDelete = true;    // timed out - delete our node
                                    break;
                                }

                                wait(millisToWait);
                            }
                            else
                            {
                                wait();
                            }
                        }
                        catch ( KeeperException.NoNodeException e ) 
                        {
                            // it has been deleted (i.e. lock released). Try to acquire again
                        }
                    }
                }
            }
        }
        catch ( Exception e )
        {
            ThreadUtils.checkInterrupted(e);
            doDelete = true;
            throw e;
        }
        finally
        {
            if ( doDelete )
            {
                deleteOurPath(ourPath);
            }
        }
        return haveTheLock;
    }
    
//3、获取锁算法思想 - maxLeases默认值是1，要求获取锁的线程永远是list的第一个线程，保证获取锁顺序性
public PredicateResults getsTheLock(CuratorFramework client, List<String> children, String sequenceNodeName, int maxLeases) throws Exception
    {
        int             ourIndex = children.indexOf(sequenceNodeName);
        validateOurIndex(sequenceNodeName, ourIndex);

        boolean         getsTheLock = ourIndex < maxLeases;
        String          pathToWatch = getsTheLock ? null : children.get(ourIndex - maxLeases);

        return new PredicateResults(pathToWatch, getsTheLock);
    }
    

​

2.5、redis实现分布式锁

2.5.0、redisson实现哪些分布式锁

• RedissonLock 功能常见常用的获取非公平锁的类
• RedissonMultiLock 功能是一个方法有很多锁构成，把这些锁统一管理
• RedissonSpinLock 功能是自旋方式获取锁的RedissonReadLock
• RedissonWriteLock 功能redisson实现的读写锁 读实现：hget 写：setNx
• RedissonRedLock 功能与RedissonMultiLock相似

2.5.1、redis分布式实现

2.5.1.1、redis setNx实现

//setNx实现分布式锁
public class SetNxLock {
		public static void main(String[] args) {
			Jedis jedis = new Jedis("localhost");
			jedis.setnx("key", "value");
			try {
				if(jedis.exists("key")) {
					jedis.expire("key", 10);
					System.out.println("我获取了锁，干点活！");
					jedis.del("key");
				}
			} catch (Exception e) {
				
			} finally {
				jedis.del("key");
			}
		}
}  

2.5.1.2、redis lua脚本实现

/**
 * 功能：redis - lua - lua实现分布式锁 使用redis.clients客户端
 * 作者：丁志超
 */  
public class LuaLock {

		public static void main(String[] args) {
			 lock("122333", "33331","10000" );
			 unlock("122333", "33331");
	}
		
	/**
	 * 加锁语法
	 * key:redis key
	 * value:redis value
	 * time: redis timeouts 锁过期时间一般大于最耗时的业务消耗的时间 
	 * 语法参考文档：https://www.runoob.com/redis/redis-scripting.html
	 * */	
	public static String lock(String key, String value,String timeOut ) {
			/**
	         *  -- 加锁脚本，其中KEYS[]为外部传入参数
	         *  -- KEYS[1]表示key 
	         *  -- ARGV[1]表示value
	         *  -- ARGV[2]表示过期时间
	         */
		    String lua_getlock_script = "if redis.call('SETNX','"+key+"','"+value+"') == 1 then" +
                "     return redis.call('pexpire','"+key+"','"+timeOut+"')" +
                " else" +
                "     return 0 " +
                "end";
			
			Jedis jedis = new Jedis("localhost");
			//在缓存中添加脚本但不执行
			String scriptId = jedis.scriptLoad(lua_getlock_script);
			//查询脚本是否添加
			Boolean isExists = jedis.scriptExists(scriptId);
			//执行脚本 返回1表示成功，返回0表示失败
			Object num = jedis.eval(lua_getlock_script);;
			return String.valueOf(num);
	}
	
	
	
	/**
	 * 释放锁语法
	 * key:redis key
	 * value:redis value
	 * time: redis timeouts 锁过期时间一般大于最耗时的业务消耗的时间 
	 * 语法参考文档：https://www.runoob.com/redis/redis-scripting.html
	 * */	
	public static String unlock(String key, String value ) {
			/**
	         *  -- 加锁脚本，其中KEYS[]为外部传入参数
	         *  -- KEYS[1]表示key 
	         *  -- ARGV[1]表示value
	         *  -- ARGV[2]表示过期时间
	         */
		    String lua_unlock_script  =
		              "if redis.call('get','"+key+"') == '"+value+"' then " +
		                      " return redis.call('del','"+key+"') " +
		                      "else  return 0 " +
		                      "end";
			
			Jedis jedis = new Jedis("localhost");
			//在缓存中添加脚本但不执行
			String scriptId = jedis.scriptLoad(lua_unlock_script);
			//查询脚本是否添加
			Boolean isExists = jedis.scriptExists(scriptId);
			//执行脚本 返回1表示成功，返回0表示失败
			Object num = jedis.eval(lua_unlock_script);;
			return String.valueOf(num);
	}
}  

2.5.1.3、redis redisson实现

/**
 * 功能：redis - Redisson - Redisson实现分布式锁
 * 作者：丁志超
 */  
public class RedissonLock {

	public static void main(String[] args) {
		Config config = new Config(); 
     	config.useSingleServer().setAddress("localhost");
     	RedissonClient redissonClient = Redisson.create(config);
		RLock rLock = redissonClient.getLock("key");
		try {
			rLock.tryLock(10, TimeUnit.SECONDS);
			System.out.println("我获取了锁，该我干活了。");
		} catch (InterruptedException e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
		}finally {
			if(rLock.isLocked()) {
				rLock.unlock();
			}
		}
		
	}
}  

2.5.1、单节点redis分布式实现原理

2.5.1.1、单节点setNx、lua脚本分布式锁实现原理

**setNx线程模型：**setNx线程模型必须是单线程（包括接收、解析、处理线程），才能保证其没有并发问题。这个猜想没有找到源码及相关文档的证明。redis线程模型文章

 SET resource_name my_random_value NX PX 30000

**lua脚本：**lua脚本其实就是一个类似于mysql的存储过程，其最大的意义是降低网络交互。性能要优于单独使用redis命令。

if redis.call("get",KEYS[1]) == ARGV[1] then
    return redis.call("del",KEYS[1])
else
    return 0
end

2.5.1.2、redisson分布式锁实现原理

//redisson锁实体数据结构
public class RedissonLockEntry  {
    
    //计数器
    private int counter;
    
    //信号类 控制多少线程同时获取锁
    private final Semaphore latch;
    private final RPromise<RedissonLockEntry> promise;
    //线程队列
    private final ConcurrentLinkedQueue<Runnable> listeners = new ConcurrentLinkedQueue<Runnable>();
}


//源码类位置 redisson RedissonLock.class
//redisson实现分布式锁源码解析
public boolean tryLock(long waitTime, long leaseTime, TimeUnit unit) throws InterruptedException {
        long time = unit.toMillis(waitTime);
        long current = System.currentTimeMillis();
        long threadId = Thread.currentThread().getId();
        //尝试获取锁其实现见后面的方法
        Long ttl = tryAcquire(waitTime, leaseTime, unit, threadId);
        // lock acquired
        if (ttl == null) {
            return true;
        }
        
        //如果锁超时直接返回失败
        time -= System.currentTimeMillis() - current;
        if (time <= 0) {
            acquireFailed(waitTime, unit, threadId);
            return false;
        }
        
        
        current = System.currentTimeMillis();   
        //通过线程ID获取锁结构体
        RFuture<RedissonLockEntry> subscribeFuture = subscribe(threadId);
        if (!subscribeFuture.await(time, TimeUnit.MILLISECONDS)) {
            if (!subscribeFuture.cancel(false)) {
                subscribeFuture.onComplete((res, e) -> {
                    if (e == null) {
                        unsubscribe(subscribeFuture, threadId);
                    }
                });
            }
            acquireFailed(waitTime, unit, threadId);
            return false;
        }

        try {
            //再次检查锁是否超时，如果超时释放锁
            time -= System.currentTimeMillis() - current;
            if (time <= 0) {
                acquireFailed(waitTime, unit, threadId);
                return false;
            }
            
            //在非超时周期内通过自旋方式获取锁
            while (true) {
                long currentTime = System.currentTimeMillis();
                ttl = tryAcquire(waitTime, leaseTime, unit, threadId);
                // lock acquired
                if (ttl == null) {
                    return true;
                }
                
                //超时释放锁
                time -= System.currentTimeMillis() - currentTime;
                if (time <= 0) {
                    acquireFailed(waitTime, unit, threadId);
                    return false;
                }

                // waiting for message
                currentTime = System.currentTimeMillis();
                if (ttl >= 0 && ttl < time) {
                    subscribeFuture.getNow().getLatch().tryAcquire(ttl, TimeUnit.MILLISECONDS);
                } else {
                    subscribeFuture.getNow().getLatch().tryAcquire(time, TimeUnit.MILLISECONDS);
                }

                time -= System.currentTimeMillis() - currentTime;
                if (time <= 0) {
                    acquireFailed(waitTime, unit, threadId);
                    return false;
                }
            }
        } finally {
            unsubscribe(subscribeFuture, threadId);
        }
//        return get(tryLockAsync(waitTime, leaseTime, unit));
    }


//redisson获取锁最底层实现，使用lua脚本实现，如果有key返回key的剩余生命时间
<T> RFuture<T> tryLockInnerAsync(long waitTime, long leaseTime, TimeUnit unit, long threadId, RedisStrictCommand<T> command) {
        return evalWriteAsync(getRawName(), LongCodec.INSTANCE, command,
                //若 key 存在返回 1 ，否则返回 0               
                "if (redis.call('exists', KEYS[1]) == 0) then " +
                        //给key增加超时时间    
                        "redis.call('hincrby', KEYS[1], ARGV[2], 1); " +
                        //设置key的生命周期      
                        "redis.call('pexpire', KEYS[1], ARGV[1]); " +
                        "return nil; " +
                        "end; " +
                        //查询key存在      
                        "if (redis.call('hexists', KEYS[1], ARGV[2]) == 1) then " +
                        "redis.call('hincrby', KEYS[1], ARGV[2], 1); " +
                        "redis.call('pexpire', KEYS[1], ARGV[1]); " +
                        "return nil; " +
                        "end; " +
                        "return redis.call('pttl', KEYS[1]);",
                Collections.singletonList(getRawName()), unit.toMillis(leaseTime), getLockName(threadId));
} 

//释放锁
@Override
protected RFuture<Void> acquireFailedAsync(long waitTime, TimeUnit unit, long threadId) {
        long wait = threadWaitTime;
        if (waitTime != -1) {
            wait = unit.toMillis(waitTime);
        }

        //这块看的有点懵，等学习lua在看
        return evalWriteAsync(getRawName(), LongCodec.INSTANCE, RedisCommands.EVAL_VOID,
                // 移除list超时的key及对应的线程
                "local queue = redis.call('lrange', KEYS[1], 0, -1);" +
                // find the location in the queue where the thread is
                "local i = 1;" +
                "while i <= #queue and queue[i] ~= ARGV[1] do " +
                    "i = i + 1;" +
                "end;" +
                // go to the next index which will exist after the current thread is removed
                "i = i + 1;" +
                // decrement the timeout for the rest of the queue after the thread being removed
                "while i <= #queue do " +
                    "redis.call('zincrby', KEYS[2], -tonumber(ARGV[2]), queue[i]);" +
                    "i = i + 1;" +
                "end;" +
                // remove the thread from the queue and timeouts set
                //移除超时的线程              
                "redis.call('zrem', KEYS[2], ARGV[1]);" +
                "redis.call('lrem', KEYS[1], 0, ARGV[1]);",
                Arrays.<Object>asList(threadsQueueName, timeoutSetName),
                getLockName(threadId), wait);
}

2.5.2、redis集群分布式锁实现原理

上面介绍的setNx、lua实现分布式锁都是基于单节点的redis本地实现方式，只要解决好本地的线程并发问题即可。当时当redis集群中使用分布式锁怎么办？redis集群实现分布式锁基于redlock算法，下面介绍其实现原理及缺陷。
​

2.5.2.1、redlock实现前提

在算法的分布式版本中，我们假设我们有N个 Redis 主节点，这些节点是完全独立的。上面已经介绍单个节点如何获取释放锁。那么在集群中每台独立的redis也会使用这种方式获取释放锁。我们假设5台redis节点，这个数值不是固定的是业务需要的选择。

2.5.2.2、redlock实现步骤

1. 它以毫秒为单位获取当前时间。
2. 它尝试顺序获取所有 N 个实例中的锁，在所有实例中使用相同的键名和随机值。在第 2 步中，当在每个实例中设置锁时，客户端使用一个比总锁自动释放时间更小的超时来获取它。例如，如果自动释放时间为 10 秒，则超时可能在 ~ 5-50 毫秒范围内。这可以防止客户端长时间处于阻塞状态，试图与已关闭的 Redis 节点通信：如果实例不可用，我们应该尽快尝试与下一个实例通信。
3. 客户端通过从当前时间中减去步骤 1 中获得的时间戳来计算获取锁所用的时间。当且仅当客户端能够在大多数实例（至少 3 个）中获取锁，并且获取锁所用的总时间小于锁有效时间，则认为该锁已获取。
4. 如果获得了锁，则其有效时间被认为是初始有效时间减去经过的时间，如步骤 3 中计算的那样。
5. 如果客户端由于某种原因获取锁失败（或者它无法锁定 N/2+1 个实例或有效时间为负），它将尝试解锁所有实例（即使是它认为没有锁定的实例）能够锁定）。

**总结：**上面是redis官方介绍的，为了保证原汁原味我给翻译下了，下面按照我理解的总结。红色部分是我对官方理解不一致的地方。

1. redis时间精度很高以毫秒为单位获取时间，这点也透露出来redis对系统时间的敏感性，也是马丁质疑他的一个点。
2. redis在N个节点中同时去获取锁，如果在超时周期获取不到锁就释放锁，防止通讯堵塞。这也是马丁质疑他的一点马丁认为这点会导致多通讯次数增加服务器成本。
3. 如果客户端能在N个redis节点中在（N/2+1）个节点中获取锁，且获取锁各个节点耗时最久的时间小于锁的有效时间就认为客户端获取了锁。
4. redis获取锁的有效时间等于锁有效时间减去获取锁花费的时间在减去集群节点时间差。怎么理解？木桶效应比如锁生命周期是10秒，最快节点1秒获取，最慢的3秒获取。此时锁生命周期就是 10-（3-1）=8在减去获取锁过程的时间，屏蔽集群的时间差异。
5. 无法获取N/2+1 个实例或获取锁超时即获取锁失败。

2.5.2.3、redlock如何保证安全

算法安全吗？我们可以尝试了解在不同场景中会发生什么。​
首先让我们假设客户端能够在大多数情况下获取锁。所有实例都将包含一个具有相同生存时间的密钥。但是，密钥是在不同的时间设置的，因此密钥也会在不同的时间到期。但是如果第一个键在时间 T1（我们在联系第一台服务器之前采样的时间）设置为最差，而最后一个键在时间 T2（我们从最后一个服务器获得回复的时间）设置为最差，我们肯定集合中第一个过期的键至少会存在MIN_VALIDITY=TTL-(T2-T1)-CLOCK_DRIFT。所有其他密钥将在稍后到期，因此我们确信至少这次密钥将同时设置。
在设置了大部分键的期间，另一个客户端将无法获取锁，因为如果 N/2+1 个键已经存在，则 N/2+1 SET NX 操作将无法成功。因此，如果获取了锁，则不可能同时重新获取它（违反互斥属性）。
​

但是，我们还希望确保多个客户端同时尝试获取锁不能同时成功。
如果客户端使用接近或大于锁最大有效时间（我们基本用于 SET 的 TTL）的时间锁定了大多数实例，它会认为锁无效并解锁实例，因此我们只需要考虑客户端能够在小于有效时间的时间内锁定大多数实例的情况。在这种情况下，对于上面已经表达的参数，MIN_VALIDITY没有客户端应该能够重新获取锁。因此，只有当锁定多数的时间大于 TTL 时间时，多个客户端才能同时锁定 N/2+1 个实例（“时间”为第 2 步的结束），从而使锁定无效。
​

**总结：**上面是redis官方对redlok如何保证安全的理解，我这边做个自己理解的注释。
想说的就是这个算法MIN_VALIDITY=TTL-(T2-T1)-CLOCK_DRIFT，节点T2-T1这个变量就是为了屏蔽掉集群节点的时间差异，如果得到的结果生命周期小于0或者无法获取N/2+1个节点就认为获取锁失败了。
​

2.5.2.4、马丁·克莱普曼喷redLock算法

我认为 Redlock 算法是一个糟糕的选择，因为它“既不是鱼也不是家禽”：它对于效率优化锁来说是不必要的重量级和昂贵的，但是对于正确性取决于锁的情况来说它不够安全。​
特别是，该算法对时序和系统时钟做出了危险的假设（基本上假设同步系统具有有限的网络延迟和有限的操作执行时间），如果不满足这些假设，则会违反安全属性。此外，它缺乏生成防护令牌的设施（保护系统免受网络长时间延迟或暂停进程的影响）。
​

如果您只需要尽力而为的锁定（作为效率优化，而不是为了正确性），我建议坚持使用简单的 Redis单节点锁定算法（条件设置如果不存在以获得锁定， atomic delete-if-value-matches 以释放锁），并在您的代码中非常清楚地记录锁只是近似值，有时可能会失败。不要费心设置五个 Redis 节点的集群。
​

另一方面，如果您需要锁定以确保正确性，请不要使用 Redlock。相反，请使用适当的共识系统，例如ZooKeeper，可能通过 实现锁的Curator 配方之一。（至少，使用具有合理事务保证的数据库。）并且请在锁定下的所有资源访问中强制使用防护令牌。
​

正如我开头所说的，Redis 是一个很好的工具，如果你使用得当。以上都不会削弱 Redis 对其预期目的的有用性。Salvatore多年来一直致力于该项目，其成功当之无愧。但是每个工具都有局限性，了解它们并相应地进行计划很重要。
​

**总结：**马丁·克莱普曼说的也对，比如redis在5台机器中3台已经获取锁，但是其中3台有一台挂了，然后重启redis还是认为成功获取锁。redlock是一种分布式场景解决一致性的方案，其也受制CAP理论。其保证AP可用性、分区容错性。redis是最大的最求性能，这是他一直信奉的原则，redlock在极苛刻的条件下出现问题。但是不代表其不科学，zk也有自己丢数据的场景，mysql一样也有自己丢数据的场景。关键是这个事情发生的概率，所以我个人观点是不怎么支持马丁·克莱普曼的观点。

三、分布式锁性能测试大比拼

光说不练说明没有理解，光练不说说明没有系统的看全面，最后给各类分布式锁性能做一个测试，由于受制于测试环境资源，我们测试的值可能和你们的不一样，但是我们追求的是测试方法的科学性而不是绝对值。本次测试用本地单体应用，实际生产环境多个节点要结合自己的环境特点测试。测试代码及压测脚本
测试方式：
1.单应用-本地部署（I58G mac）一套应用jmeter压测这个应用
2.redis master集群-4核心8G3节点
3.zk集群 - 4核心8G*3节点

四、相关文档

3.1、redis分布式锁实现原理官方文档
3.2、马丁·克莱普曼对redlock质疑