标签：
在分布式场景下，有很多种情况都需要实现最终一致性。在设计远程上下文的领域事件的时候，为了保证最终一致性，在通过领域事件进行通讯的方式中，可以共享存储（领域模型和消息的持久化数据源），或者做全局XA事务（两阶段提交，数据源可分开），也可以借助消息中间件（消费者处理需要能幂等）。通过Observer模式来发布领域事件可以提供很好的高并发性能，并且事件存储也能追溯更小粒度的事件数据，使各个应用系统拥有更好的自治性。
本文主要探讨另外一种实现分布式最终一致性的解决方案——采用分布式锁。基于分布式锁的解决方案，比如zookeeper,redis都是相较于持久化（如利用InnoDB行锁，或事务，或version乐观锁）方案提供了高可用性，并且支持丰富化的使用场景。 本文通过Java版本的redis分布式锁开源框架——Redisson来解析一下实现分布式锁的思路。
如果是不跨限界上下文的情况，跟本地领域服务相关的数据一致性，尽量还是用事务来保证。但也有些无法用事务或者乐观锁来处理的情况，这些情况大多是对于一个共享型的数据源，有并发写操作的场景，但又不是对于单一领域的操作。
举个例子，还是用租书来比喻,A和B两个人都来租书，在查看图书的时候，发现自己想要看的书《大设计》库存仅剩一本。书店系统中，书作为一种商品，是在商品系统中，以Item表示出租商品的领域模型，同时每一笔交易都会产生一个订单，Order是在订单系统（交易限界上下文）中的领域模型。这里假设先不考虑跨系统通信的问题（感兴趣的可以参考下领域服务、领域事件），也暂时不考虑支付环节，但是我们需要保证A,B两个人不会都对于《大设计》产生订单就可以，也就是其中一个人是可以成功下单，另外一个人只要提示库存已没即可。此时，书的库存就是一种共享的分布式资源，下订单，减库存就是一个需要保证一致性的写操作。但又因为两个操作不能在同一个本地事务，或者说，不共享持久化的数据源的情况，这时候就可以考虑用分布式锁来实现。本例子中，就需要对于共享资源——书的库存进行加锁，至于锁的key可以结合领域模型的唯一标识，如itemId,以及操作类型（如操作类型是RENT的）设计一个待加锁的资源标识。当然，这里还有一个并发性能的问题，如果是个库存很多的秒杀类型的业务，那么就不能单纯在itemId 加类型加锁，还需要设计排队队列以及合理的调度算法，防止超卖等等，那些就是题外话了。本文只是将这个场景作为一个切入点，具体怎么设计锁，什么场景用还要结合业务。
分布式的思路和线程同步锁ReentrantLock的思路是一样的。我们也要考虑如以下几个问题：
在Redisson介绍前，回顾下Redis的命令，以及不通过任何开源框架，可以基于redis怎么设计一个分布式锁。基于不同应用系统实现的语言，也可以通过其他一些如Jedis,或者Spring的RedisOperations 等，来执行Reids命令Redis command list。
分布式锁主要需要以下redis命令，这里列举一下。在实现部分可以继续参照命令的操作含义。
假设我们现在要给itemId 1234 和下单操作 OP_ORDER 加锁，key是OP_ORDER_1234,结合上面的redis命令，似乎加锁的时候只要一个SETNX OP_ORDER_1234 currentTimestamp ，如果返回1代表加锁成功，返回0 表示锁被占用着。然后再用DEL OP_ORDER_1234解锁，返回1表示解锁成功，0表示已经被解锁过。然而却还存在着很多问题：SETNX会存在锁竞争，如果在执行过程中客户端宕机，也会引起死锁问题，即锁资源无法释放。并且当一个资源解锁的时候，释放锁之后，其他之前等待的锁没有办法再次自动重试申请锁（除非重新申请锁）。解决死锁的问题其实可以可以向Mysql的死锁检测学习，设置一个失效时间，通过key的时间戳来判断是否需要强制解锁。但是强制解锁也存在问题，一个就是时间差问题，不同的机器的本地时间可能也存在时间差，在很小事务粒度的高并发场景下还是会存在问题，比如删除锁的时候，在判断时间戳已经超过时效，有可能删除了其他已经获取锁的客户端的锁。另外，如果设置了一个超时时间，但是确实执行时间超过了超时时间，那么锁会被自动释放，原来持锁的客户端再次解锁的时候会出现问题，而且最为严重的还是一致性没有得到保障。
所以设计的时候需要考虑以下几点：
由于时间戳的设计有很多问题，以及上述几个问题，所以再换一种思路。先回顾几个关于锁的概念和经典java API。通过一些java.util.concurrent的API来处理一些本地队列的同步以及等待信号量的处理。
并发API以及一些框架的使用主要是控制锁的进入和调度，加锁的流程以及锁的逻辑也是非常重要。因为redis支持hash结构，除了key作为锁的标识，还可以利用value的结构
下面参数的含义先说明下 :
// 检查是否key已经被占用，如果没有则设置超时时间和唯一标识，初始化value=1
if (redis.call(‘exists‘, KEYS[1]) == 0)
then
redis.call(‘hset‘, KEYS[1], ARGV[2], 1);
redis.call(‘pexpire‘, KEYS[1], ARGV[1]);
return nil;
end;
// 如果锁重入,需要判断锁的key field 都一直情况下 value 加一
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]);
以上的方法，当返回空是，说明获取到锁，如果返回一个long数值（pttl 命令的返回值），说明锁已被占用，通过返回剩余时间，外部可以做一些等待时间的判断和调整。
也还是先说明一下参数信息：
- KEYS[1] ：需要加锁的key，这里需要是字符串类型。
- KEYS[2] ：redis消息的ChannelName,一个分布式锁对应唯一的一个channelName:“redisson_lock__channel__{” + getName() + “}”
- ARGV[1] ：reids消息体，这里只需要一个字节的标记就可以，主要标记redis的key已经解锁，再结合redis的Subscribe，能唤醒其他订阅解锁消息的客户端线程申请锁。
- ARGV[2] ：锁的超时时间，防止死锁
- ARGV[3] ：锁的唯一标识，也就是刚才介绍的 id（UUID.randomUUID()） + “:” + threadId
// 如果key已经不存在，说明已经被解锁，直接发布（publihs）redis消息
if (redis.call(‘exists‘, KEYS[1]) == 0)
then
redis.call(‘publish‘, KEYS[2], ARGV[1]);
return 1;
end;
// key和field不匹配，说明当前客户端线程没有持有锁，不能主动解锁。
if (redis.call(‘hexists‘, KEYS[1], ARGV[3]) == 0)
then
return nil;
end;
// 将value减1
local counter = redis.call(‘hincrby‘, KEYS[1], ARGV[3], -1);
// 如果counter>0说明锁在重入，不能删除key
if (counter > 0)
then
redis.call(‘pexpire‘, KEYS[1], ARGV[2]); return 0;
else
// 删除key并且publish 解锁消息
redis.call(‘del‘, KEYS[1]); redis.call(‘publish‘, KEYS[2], ARGV[1]);
return 1;
end;
return nil;
这就是解锁过程，当然建议提供强制解锁的接口，直接删除key，以防一些紧急故障出现的时候，关键业务节点受到影响。这里还有一个关键点，就是publish命令，通过在锁的唯一通道发布解锁消息，可以减少其他分布式节点的等待或者空转，整体上能提高加锁效率。至于redis的消息订阅可以有多种方式，基于Jedis的订阅API或者Spring的MessageListener都可以实现订阅，这里就可以结合刚才说的Semaphore,在第一次申请锁失败后acquire,接收到分布式消息后release就可以控制申请锁流程的再次进入。下面结合Redisson源码，相信会有更清晰的认识。
Redisson使用起来很方便，但是需要redis环境支持eval命令，否则一切都是悲剧，比如me.结果还是要用RedisCommands去写一套。例子就如下，获得一个RLock锁对象，然后tryLock 和unlock。trylock方法提供了锁重入的实现，并且客户端一旦持有锁，就会在能正常运行期间一直持有锁，直到主动unlock或者节点故障，主动失效（超过默认的过期时间）释放锁。
public boolean doMyBusiness(Object t) {
RLock lock = redissonClient.getLock(getLockKey(t));
try {
if (lock.tryLock()) {
//do need Business
return true;
} else {
// do other Business or return error.
return false;
}
} finally {
lock.unlock();
}
}
Redisson还提供了设置最长等待时间以及设置释放锁时间的含参tryLock接口 boolean tryLock(long waitTime, long leaseTime, TimeUnit unit) throws InterruptedException; 。Redisson的lock 扩展了java.util.concurrent.locks.Lock的实现，也基本按照了Lock接口的实现方案。lock（）方法会一直阻塞申请锁资源，直到有可用的锁释放。下面一部分会详细解析一部分关键实现的代码。
Redisson 的异步任务(Future,Promise，FutureListener API)，任务计时器(Timeout，TimerTask)，以及通过AbstractChannel连接redis以及写入执行批处理命令等很多都是基于netty框架的。po主因为不能使用eval,所以用Spring提供的redisApi ,RedisOperations来处理redis指令，异步调度等用了Spring的AsyncResult，MessageListener以及一些concurrent api。这里还是先看一下Redisson的实现。
这里以带参数的trylock解析一下，无参的trylock是一种默认参数的实现。先源码走读一下。
@Override
public boolean tryLock(long waitTime, long leaseTime, TimeUnit unit) throws InterruptedException {
long time = unit.toMillis(waitTime);
// 申请锁，返回还剩余的锁过期时间
Long ttl = tryAcquire(leaseTime, unit);
// 如果为空，表示申请锁成功
if (ttl == null) {
return true;
}
// 订阅监听redis消息，并且创建RedissonLockEntry，其中RedissonLockEntry中比较关键的是一个 Semaphore属性对象用来控制本地的锁请求的信号量同步，返回的是netty框架的Future实现。
Future<RedissonLockEntry> future = subscribe();
// 阻塞等待subscribe的future的结果对象，如果subscribe方法调用超过了time，说明已经超过了客户端设置的最大wait time，则直接返回false，取消订阅，不再继续申请锁了。
if (!future.await(time, TimeUnit.MILLISECONDS)) {
future.addListener(new FutureListener<RedissonLockEntry>() {
@Override
public void operationComplete(Future<RedissonLockEntry> future) throws Exception {
if (future.isSuccess()) {
unsubscribe(future);
}
}
});
return false;
}
try {
while (true) {
// 再次尝试一次申请锁
ttl = tryAcquire(leaseTime, unit);
// 获得锁，返回
if (ttl == null) {
return true;
}
// 不等待申请锁，返回
if (time <= 0) {
return false;
}
// 阻塞等待锁
long current = System.currentTimeMillis();
RedissonLockEntry entry = getEntry();
if (ttl >= 0 && ttl < time) {
// 通过信号量(共享锁)阻塞,等待解锁消息.
// 如果剩余时间(ttl)小于wait time ,就在 ttl 时间内，从Entry的信号量获取一个许可(除非被中断或者一直没有可用的许可)。
// 否则就在wait time 时间范围内等待可以通过信号量
entry.getLatch().tryAcquire(ttl, TimeUnit.MILLISECONDS);
} else {
entry.getLatch().tryAcquire(time, TimeUnit.MILLISECONDS);
}
// 更新等待时间(最大等待时间-已经消耗的阻塞时间)
long elapsed = System.currentTimeMillis() - current;
time -= elapsed;
}
} finally {
// 无论是否获得锁,都要取消订阅解锁消息
unsubscribe(future);
}
}上述方法，调用加锁的逻辑就是在tryAcquire(long leaseTime, TimeUnit unit)中
private Long tryAcquire(long leaseTime, TimeUnit unit) {
if (leaseTime != -1) {
return get(tryLockInnerAsync(leaseTime, unit, Thread.currentThread().getId()));
}
return get(tryLockInnerAsync(Thread.currentThread().getId()));
}
tryAcquire(long leaseTime, TimeUnit unit)只是针对leaseTime的不同参数进行不同的转发处理，再提一下，trylock的无参方法就是直接调用了get(tryLockInnerAsync(Thread.currentThread().getId()));
所以下面再看核心的tryLockInnerAsync 基本命令已经在之前解析过，相信这里看起来应该比较轻松，返回的是一个future对象，是为了异步处理IO，提高系统吞吐量。
Future<Long> tryLockInnerAsync(long leaseTime, TimeUnit unit, long threadId) {
internalLockLeaseTime = unit.toMillis(leaseTime);
return commandExecutor.evalWriteAsync(getName(), LongCodec.INSTANCE, RedisCommands.EVAL_LONG,
"if (redis.call(‘exists‘, KEYS[1]) == 0) then " +
"redis.call(‘hset‘, KEYS[1], ARGV[2], 1); " +
"redis.call(‘pexpire‘, KEYS[1], ARGV[1]); " +
"return nil; " +
"end; " +
"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.<Object>singletonList(getName()), internalLockLeaseTime, getLockName(threadId));
}
再说明一下，tryLock(long waitTime, long leaseTime, TimeUnit unit)有leaseTime参数的申请锁方法是会按照leaseTime时间来自动释放锁的。但是没有leaseTime参数的，比如tryLock()或者tryLock(long waitTime, TimeUnit unit)以及lock()是会一直持有锁的。再来看一下没有leaseTime参数的tryLockInnerAsync(Thread.currentThread().getId())
private Future<Long> tryLockInnerAsync(long threadId) {
// 设置了默认的30秒的失效时间
Future<Long> ttlRemaining = tryLockInnerAsync(LOCK_EXPIRATION_INTERVAL_SECONDS, TimeUnit.SECONDS, threadId);
ttlRemaining.addListener(new FutureListener<Long>() {
@Override
public void operationComplete(Future<Long> future) throws Exception {
// 如果future方法没有执行完成（IO被中断等原因）直接返回，不继续处理
if (!future.isSuccess()) {
return;
}
Long ttlRemaining = future.getNow();
// 成功申请到锁，开始一个调度程序
if (ttlRemaining == null) {
scheduleExpirationRenewal();
}
}
});
return ttlRemaining;
}
这里比有leaseTime参数的trylock就多了异步scheduleExpirationRenewal调度。可以继续看一下,这里的expirationRenewalMap就是之前降到的一个ConcurrentMap结构。下面的这个调度方式很精妙。除非被unlock的cancleTask方法触发，否则会一直循环重置过期时间。
private static final ConcurrentMap<String, Timeout> expirationRenewalMap = PlatformDependent.newConcurrentHashMap();
private void scheduleExpirationRenewal() {
// 保证任务不会被重复创建
if (expirationRenewalMap.containsKey(getName())) {
return;
}
// 添加一个netty的Timeout回调任务，每（internalLockLeaseTime / 3）毫秒执行一次
Timeout task = commandExecutor.getConnectionManager().newTimeout(new TimerTask() {
@Override
public void run(Timeout timeout) throws Exception {
// 异步调用redis的pexpire命令，重置过期时间
expireAsync(internalLockLeaseTime, TimeUnit.MILLISECONDS);
// 移除，确保下一次调用
expirationRenewalMap.remove(getName());
scheduleExpirationRenewal(); // 再次循环调用
}
}, internalLockLeaseTime / 3, TimeUnit.MILLISECONDS);
// expirationRenewalMap如果已经有getName()任务，停止任务，也是为了在极端的并发情况下，保证任务不会被重复创建
if (expirationRenewalMap.putIfAbsent(getName(), task) != null) {
task.cancel();
}
}
这个任务，其实还有一个问题，个人觉得在expirationRenewalMap.containsKey判断时也加上isLocked判断会比较好，以防止unlock时出现redis节点异常的时候，任务没有办法自动停止，或者设置一个最大执行次数的限制也可以，否则极端情况下也会耗尽本地节点的CPU资源。
解锁的逻辑相对简单，如下，redis 命令相信看起来也会比较轻松了。
@Override
public void unlock() {
Boolean opStatus = commandExecutor.evalWrite(getName(), LongCodec.INSTANCE, RedisCommands.EVAL_BOOLEAN,
"if (redis.call(‘exists‘, KEYS[1]) == 0) then " +
"redis.call(‘publish‘, KEYS[2], ARGV[1]); " +
"return 1; " +
"end;" +
"if (redis.call(‘hexists‘, KEYS[1], ARGV[3]) == 0) then " +
"return nil;" +
"end; " +
"local counter = redis.call(‘hincrby‘, KEYS[1], ARGV[3], -1); " +
"if (counter > 0) then " +
"redis.call(‘pexpire‘, KEYS[1], ARGV[2]); " +
"return 0; " +
"else " +
"redis.call(‘del‘, KEYS[1]); " +
"redis.call(‘publish‘, KEYS[2], ARGV[1]); " +
"return 1; "+
"end; " +
"return nil;",
Arrays.<Object>asList(getName(), getChannelName()), LockPubSub.unlockMessage, internalLockLeaseTime, getLockName(Thread.currentThread().getId()));
if (opStatus == null) {
throw new IllegalMonitorStateException("attempt to unlock lock, not locked by current thread by node id: "
+ id + " thread-id: " + Thread.currentThread().getId());
}
// 解锁成功之后取消更新锁expire的时间任务
if (opStatus) {
cancelExpirationRenewal();
}
这里的 cancelExpirationRenewal对应着取消 scheduleExpirationRenewal的重置expire时间任务。
void cancelExpirationRenewal() {
Timeout task = expirationRenewalMap.remove(getName());
if (task != null) {
task.cancel();
}
}
再看一下Redisson是如何处理unlock的redis消息的。这里的消息内容就是unlockMessage = 0L和unlock方法中publish的内容是对应的。
public class LockPubSub extends PublishSubscribe<RedissonLockEntry> {
public static final Long unlockMessage = 0L;
@Override
protected RedissonLockEntry createEntry(Promise<RedissonLockEntry> newPromise) {
return new RedissonLockEntry(newPromise);
}
@Override
protected void onMessage(RedissonLockEntry value, Long message) {
if (message.equals(unlockMessage)) {
// 释放一个许可，唤醒等待的entry.getLatch().tryAcquire去再次尝试获取锁。
value.getLatch().release();
// 如果entry还有其他Listeners回调，也唤醒执行。
synchronized (value) {
Runnable runnable = value.getListeners().poll();
if (runnable != null) {
if (value.getLatch().tryAcquire()) {
runnable.run();
} else {
value.getListeners().add(runnable);
}
}
}
}
}
}
Redisson还支持Redis的多种集群配置，一主一备，一主多备，单机等等。也是通过netty的EventExecutorGroup，Promise，Future等API实现调度的。
在思考是否采用分布式锁以及采用哪种实现方案的时候，还是要基于业务，技术方案一定是基于业务基础，服务于业务，并且衡量过投入产出比的。所以如果有成熟的解决方案，在业务可承受规模肯定是不要重复造轮子，当然还要经过严谨的测试。在po主用Spring的redis api实现时，也遇到了一些问题。
比如hIncrBy 的字符集问题，在使用命令的时候，当然可以直接set a 1然后incr a 1，这个问题可以参考ERR value is not an integer or out of range 问题，但在使用RedisConnection的时候，需要通过转码，byte[] value =SafeEncoder.encode(String.valueOf(“1”)) 再 connection.hSet(key, field, value)这样才可以，或者自己通过String转成正确的编码也可以。
还有刚才说的调度pexpire任务，在unlock异常的时候，任务池中的任务无法自动结束。另外就是Spring的MessageListener的onMessage(Message message, byte[] pattern)回调方法message.getBody()是byte数组，消息内容转化的时候要处理一下。
希望文章能够给大家一起思路或者能解决一些困惑。另外，望能够支持下po主公司的APP——蜂潮运动，各位工程师们在辛苦工作之余也要注重身体，加强锻炼。跑步是一种时尚，下载我们的APP，让生活充满健康和乐趣。如果有兴趣可以和po主继续深入交流下技术或者探讨下运动。
标签：
原文地址：http://blog.csdn.net/lijingyao8206/article/details/51150936
