分布式锁相比拟多线程锁，更加高级一些。它的作用范畴，也由单机转换为分布式，是罕用的资源协调伎俩。罕用的有redis分布式做和zk分布式锁。但它们有什么区别呢？咱们在平时应用中，又该如何抉择。
- 解析
这个问题对要求较高，它不仅要理解实现办法，还要对原理有所把握。所以问题答复起来，分为很多档次。
家喻户晓，Redis标榜的是轻量级，直观上分布式锁是比拟好实现的，比方应用setnx，但一旦退出高可用这个属性，Redis锁的实现难度就会爆炸式回升。
再加上锁的其余几个属性：乐观乐观、读写锁等，事件会更加的简单。
如果你全都通晓，聊一天都聊不完。
- 尝试剖析以下
先来一个，比拟通俗、入门的剖析：
redis的分布式锁，能够基于setnx指令实现（但其实更倡议应用带nx参数的set指令）
zk的分布式锁，是基于长期节点的有序性和节点的监听机制实现的
这种答复形式，间接把本人给绕进去了，因为这波及到十分多的细节。他人只是问区别，为什么把本人往源码级别绕呢？
倡议这样剖析：
Redis，应用redisson封装的RedLock
Zk，应用curator封装的InterProcessMutex
比照：
实现难度上：Zookeeper >= redis
服务端性能：redis > Zookeeper
客户端性能：Zookeeper > redis
可靠性：Zookeeper > redis
细聊：
2.1 实现难度
对于间接操纵底层API来说，实现难度都是差不多的，都须要思考很多边界场景。但因为Zk的ZNode人造具备锁的属性，所以间接上手撸的话，很简略。
Redis须要思考太多异样场景，比方锁超时、锁的高可用等，实现难度较大。
2.2 服务端性能
Zk基于Zab协定，须要一半的节点ACK，才算写入胜利，吞吐量较低。如果频繁加锁、开释锁，服务端集群压力会很大。
Redis基于内存，只写Master就算胜利，吞吐量高，Redis服务器压力小。
2.3 客户端性能
Zk因为有告诉机制，获取锁的过程，增加一个监听器就能够了。防止了轮询，性能耗费较小。
Redis并没有告诉机制，它只能应用相似CAS的轮询形式去争抢锁，较多空转，会对客户端造成压力。
2.4 可靠性
这个就很显著了。Zookeeper就是为协调而生的，有严格的Zab协定控制数据的一致性，锁模型强壮。
Redis谋求吞吐，可靠性上稍逊一筹。即便应用了Redlock，也无奈保障100%的健壮性，但个别的利用不会遇到极其场景，所以也被罕用。
- 扩大
Zk的分布式锁样代码样例：
import org.apache.curator.framework.CuratorFramework;import org.apache.curator.framework.recipes.locks.InterProcessMutex;import java.util.concurrent.TimeUnit;public class ExampleClientThatLocks{ private final InterProcessMutex lock; private final FakeLimitedResource resource; private final String clientName; public ExampleClientThatLocks(CuratorFramework client, String lockPath, FakeLimitedResource resource, String clientName) { this.resource = resource; this.clientName = clientName; lock = new InterProcessMutex(client, lockPath); } public void doWork(long time, TimeUnit unit) throws Exception { if ( !lock.acquire(time, unit) ) { throw new IllegalStateException(clientName + " could not acquire the lock"); } try { System.out.println(clientName + " has the lock"); resource.use(); } finally { System.out.println(clientName + " releasing the lock"); lock.release(); // always release the lock in a finally block } }}RedLock的分布式锁应用样例：
String resourceKey = "goodgirl";RLock lock = redisson.getLock(resourceKey);try { lock.lock(5, TimeUnit.SECONDS); //真正的业务 Thread.sleep(100);} catch (Exception ex) { ex.printStackTrace();} finally { if (lock.isLocked()) { lock.unlock(); }}再附一段RedLock的外部lock和unlock的代码实现，以便对你对其复杂度有肯定的理解。
@Override <T> RFuture<T> tryLockInnerAsync(long leaseTime, TimeUnit unit, long threadId, RedisStrictCommand<T> command) { internalLockLeaseTime = unit.toMillis(leaseTime); return commandExecutor.evalWriteAsync(getName(), LongCodec.INSTANCE, command, "local mode = redis.call('hget', KEYS[1], 'mode'); " + "if (mode == false) then " + "redis.call('hset', KEYS[1], 'mode', 'read'); " + "redis.call('hset', KEYS[1], ARGV[2], 1); " + "redis.call('set', KEYS[2] .. ':1', 1); " + "redis.call('pexpire', KEYS[2] .. ':1', ARGV[1]); " + "redis.call('pexpire', KEYS[1], ARGV[1]); " + "return nil; " + "end; " + "if (mode == 'read') or (mode == 'write' and redis.call('hexists', KEYS[1], ARGV[3]) == 1) then " + "local ind = redis.call('hincrby', KEYS[1], ARGV[2], 1); " + "local key = KEYS[2] .. ':' .. ind;" + "redis.call('set', key, 1); " + "redis.call('pexpire', key, ARGV[1]); " + "local remainTime = redis.call('pttl', KEYS[1]); " + "redis.call('pexpire', KEYS[1], math.max(remainTime, ARGV[1])); " + "return nil; " + "end;" + "return redis.call('pttl', KEYS[1]);", Arrays.<Object>asList(getName(), getReadWriteTimeoutNamePrefix(threadId)), internalLockLeaseTime, getLockName(threadId), getWriteLockName(threadId)); }@Override protected RFuture<Boolean> unlockInnerAsync(long threadId) { String timeoutPrefix = getReadWriteTimeoutNamePrefix(threadId); String keyPrefix = getKeyPrefix(threadId, timeoutPrefix); return commandExecutor.evalWriteAsync(getName(), LongCodec.INSTANCE, RedisCommands.EVAL_BOOLEAN, "local mode = redis.call('hget', KEYS[1], 'mode'); " + "if (mode == false) then " + "redis.call('publish', KEYS[2], ARGV[1]); " + "return 1; " + "end; " + "local lockExists = redis.call('hexists', KEYS[1], ARGV[2]); " + "if (lockExists == 0) then " + "return nil;" + "end; " + "local counter = redis.call('hincrby', KEYS[1], ARGV[2], -1); " + "if (counter == 0) then " + "redis.call('hdel', KEYS[1], ARGV[2]); " + "end;" + "redis.call('del', KEYS[3] .. ':' .. (counter+1)); " + "if (redis.call('hlen', KEYS[1]) > 1) then " + "local maxRemainTime = -3; " + "local keys = redis.call('hkeys', KEYS[1]); " + "for n, key in ipairs(keys) do " + "counter = tonumber(redis.call('hget', KEYS[1], key)); " + "if type(counter) == 'number' then " + "for i=counter, 1, -1 do " + "local remainTime = redis.call('pttl', KEYS[4] .. ':' .. key .. ':rwlock_timeout:' .. i); " + "maxRemainTime = math.max(remainTime, maxRemainTime);" + "end; " + "end; " + "end; " + "if maxRemainTime > 0 then " + "redis.call('pexpire', KEYS[1], maxRemainTime); " + "return 0; " + "end;" + "if mode == 'write' then " + "return 0;" + "end; " + "end; " + "redis.call('del', KEYS[1]); " + "redis.call('publish', KEYS[2], ARGV[1]); " + "return 1; ", Arrays.<Object>asList(getName(), getChannelName(), timeoutPrefix, keyPrefix), LockPubSub.UNLOCK_MESSAGE, getLockName(threadId)); }所以，倡议应用曾经封装好的组件。如果你非要应用setnx或者set指令去做这些事，xjjdog只能说是想被虐。基本原理咱们能够做到理解，这些细节，不下点功夫是理不清的。
说了这半天，咱们选型的时候，该如何做呢？这要看你的基础设施。如果你的利用用到了zk，而且集群性能很强劲，优选zk。如果你只有redis，不想为了个分布式锁，引入臃肿的zk，那就用redis。
零根底学习Java编程，举荐退出我的十年Java学习园地。