共计 4808 个字符，预计需要花费 13 分钟才能阅读完成。
分布式锁相比拟多线程锁，更加高级一些。它的作用范畴，也由单机转换为分布式，是罕用的资源协调伎俩。罕用的有 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 学习园地。
