Redis实现分布式锁

并发编程之美 2020-09-27

739

介绍

为了保证共享资源在高并发情况下同一时间只能被一个线程执行，在传统单体应用单机部署的情况下，可以使用Java并发处理相关的锁，synchronized或ReentrantLock进行互斥控制。但是在分布式系统中，应用分布在不同的机器上，这使得单机部署的并发控制锁失效。为了解决这个问题就需要一种跨JVM的互斥机制来控制共享资源的访问，这就是分布式锁钥解决的问题。

一、分布式锁的原理及特性

之前写了一篇文章，介绍了分布式锁的原理及特性。接下来我们以扣减库存这个场景为例，再详解介绍下分布式锁的原理和特性。

一个电商下订单系统，目前是单机部署的，下单逻辑是库存足够才允许下单成功，不能出现扣减库存为负数的情况。如果是在秒杀场景下，系统的并发量非常的高，所以会预先将商品库存保存在redis中，用户下单的时候会先检查库存是否足够，在更新redis中的库存。系统架构如下：

但是这样一来就会产生一个问题：假如当商品库存只剩1个的时候，同时来了两个请求，其中一个执行到第3步，更新数据库的库存为0，还没执行第4步的时候，另一个请求执行到了第2步，发现库存为1，就继续执行第3步，更新数据库的库存为-1，这样就出现了库存超卖的问题。

单机部署的情况我们很容易就想到使用线程锁把2、3、4步锁住，让他们顺序执行完后，另一个线程才能进来执行第2步。结构如下：

如上图在执行第2步的时候，我们可以使用synchronized或者ReentrantLock来锁住，然后在第4步执行完后释放锁。

但是随着系统并发的上升，一台机器扛不住了，我们的架构采用增加一台机器，进行多机部署，如下图：

假如此时两个用户的请求同时到来，但是落在了不同的机器上，因为上图中的两个A系统，运行在两个不同的JVM里面，他们加的锁只对属于自己JVM里面的线程有效，对于其他JVM的线程是无效的。那么这两个请求还是可以同时执行了，还是会出现库存超卖的问题。

那么此时的问题就是，我们需要一个全局唯一的锁，可以保证两台机器加的锁是同一个锁，才能解决如上库存超卖的问题，此时这个场景就是分布式锁的使用场景了。

分布式锁在整个系统提供一个全局、唯一的获取锁的机制，然后每个系统在需要加锁时，都通过这个机制去获取锁，这样不同的系统拿到的就可以认为是同一把锁。具体实现分布式锁的方式可以是以下几种：

1. 数据库实现分布式锁

乐观锁实现方式
悲观锁实现方式，性能非常不好

2. Memcache

利用Memcache的add命令。此命令是原子性操作，只有在key不存在的情况下才能add成功，也就意味着线程得到了锁

3. 基于Redis实现分布锁

利用Redis的setnx命令。只有在key不存在的情况下才能set成功

4. 基于zookeeper实现分布式锁

利用zookeeper的临时顺序节点，来实现分布式锁和等待队列

5. Chubby

Google公司实现的粗粒度分布式锁服务，底层利用了Paxos一致性算法

采用分布式锁实现库存扣减的逻辑如下：

所以现在我们知道了库存超卖场景在分布式部署系统的情况下使用Java原生的锁机制无法保证线程安全，所以我们需要用到分布式锁的解决方案。

那么如何实现分布式锁呢？接下来我们介绍下使用redis实现分布式锁的方案。

二、Redis实现分布式锁

Reids的很多命令都可以实现分布式锁，最常用的是setnx命令，setnx的含义是只有当key不存在时设置key的值为value，当key存在时，不做任何反应。当返回1时获取锁成功，当获取锁失败时，每隔1秒自动尝试再次获取锁，看能否获取到锁，等别人的锁过期了或者释放了锁，才能获取到锁。

Redis实现分布式锁流程如下：

使用stringRedisTmplate模拟分布式锁的实现，先来个错误版的：

String lockKey = "lock:name"
try {
  Boolean result = stringRedisTmplate.opsForValue().setIfAbsent(lockKey, "value");
  stringRedisTmplate.expire(lockKey,10,TimeUnit.SECONDS);
  if (!result) {
    return "获取锁失败";
  }
  // 处理业务逻辑
} finally {
  stringRedisTmplate.delete(lockKey);
}

上面是一个错误的示例，存在的问题是setnx和expire设置过期时间是两步操作，非原子性操作，当某线程执行setnx得到了锁，还没来得及设置过期时间时，redis节点挂掉了，这把锁就永久有效了，其他线程就无法再获得锁了。怎么解决呢？

使用stringRedisTmplate将setnx和expire合并到一起，它的底层是使用的lua脚本，是原子性操作：

Boolean result = stringRedisTmplate.opsForValue().setIfAbsent(lockKey,"value",10,TimeUnit.SECONDS);

这样就解决了无法释放锁的问题，那么这个又存在什么问题呢？我们设置的过期时间是10秒，假设第一个线程从加锁到解锁需要15秒，执行业务逻辑需要10秒，我们用下图模拟下在高并发场景下的流程过程：

上面实现的分布式锁不具备拥有者标识，线程1在执行业务逻辑期间锁过期了，线程2获得了锁，线程1执行完业务逻辑解锁时解锁了线程2的锁，同理后面线程3解锁了线程4的。即发生了在高并发场景下锁永远失效，导致了锁误删的问题。

可以给锁加个唯一标识，比如请求ID：

String lockKey = "lock:name"
String requestId = UUID.randomUUID().toString();
try {
  Boolean result = stringRedisTmplate.opsForValue().setIfAbsent(lockKey,requestId,10,TimeUnit.SECONDS);
  if (!result) {
    return "获取锁失败";
  }
   处理业务逻辑
} finally {
  if (requestId.equals(stringRedisTmplate.opsForValue().get(lockKey))) {
      stringRedisTmplate.delete(lockKey);
  }
}

这样就解决误删的问题，但有个问题是在我们还没有处理完业务逻辑，我们设置的锁已经过期了，在高并发场景下可能存在锁永远失效的问题，那么这种问题一般是怎么解决呢？

解决思路一般是子线程监测锁并且重置锁过期时间，当线程获取到了锁，开启一个分线程开启一个定时器，每隔一段时间与检测锁有没有过期，如果锁还存在，则把锁的过期时间进行重置。当主线程执行完释放掉锁后，主线程结束，对应的分线程也就结束了。针对锁过期时间重置的问题可以使用redisson框架来解决，下面我们看下redisson框架来实现分布式锁。

三、Redisson框架实现分布式锁

如果你的项目中Redis是多机部署的，那么使用Redisson实现分布式锁是非常合适的，这是Redis官方提供的Java组件，代码示例：

springboot下配置一个redisson客户端：

@SpringBootApplication
public class Application {
    public static void main(String[] args) {
        SpringApplication.run(Application.class, args);
    }


    @Bean
    public Redisson redisson() {
        // 单机模式
        Config config = new Config();
        config.useSingleServer().setAddress("redis://127.0.0.1:6379").setDatabase(0);
        return (Redisson)Redisson.create(config);
    }
}

加锁解锁代码：

RLock lock = redisson.getLock(lockKey);
lock.tryLock(30, TimeUnit.SECONDS);
lock.unlock();

使用就是这么简单，redisson所有的指令都是通过Lua脚本执行的，保证了原子性。

redisson设置了key默认的过期时间是30分钟，前面提到了一种业务场景，当业务执行时间超过了超时时间可以使用redisson来解决这个问题，redisson重置锁的过期时间过程如下：

一般timer定时检测的时间是设置的锁的过期时间的三分之一。它会在你获取锁之后，每隔三分之一超时时间就会把锁的超时时间重置。这样当业务逻辑还没处理完之前，锁是不会过期的，并且如果机器宕机的话，则重置锁时长的timer定时检测也就消失了，锁最多再过一个超时时长也就释放了。我们可以看下它的实现代码：

// 加锁逻辑
private <T> RFuture<Long> tryAcquireAsync(long leaseTime, TimeUnit unit, final long threadId) {
    if (leaseTime != -1) {
        return tryLockInnerAsync(leaseTime, unit, threadId, RedisCommands.EVAL_LONG);
    }
    // 调用lua脚本，设置key的过期时间
    RFuture<Long> ttlRemainingFuture = tryLockInnerAsync(commandExecutor.getConnectionManager().getCfg().getLockWatchdogTimeout(), TimeUnit.MILLISECONDS, threadId, RedisCommands.EVAL_LONG);
    ttlRemainingFuture.addListener(new FutureListener<Long>() {
        @Override
        public void operationComplete(Future<Long> future) throws Exception {
            if (!future.isSuccess()) {
                return;
            }


            Long ttlRemaining = future.getNow();
             lock acquired
            if (ttlRemaining == null) {
                // 定时重置锁超时时间逻辑
                scheduleExpirationRenewal(threadId);
            }
        }
    });
    return ttlRemainingFuture;
}




<T> RFuture<T> tryLockInnerAsync(long leaseTime, TimeUnit unit, long threadId, RedisStrictCommand<T> command) {
    internalLockLeaseTime = unit.toMillis(leaseTime);


    return commandExecutor.evalWriteAsync(getName(), LongCodec.INSTANCE, command,
              "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));
}






// 定时重置锁超时时间最终会调用了这里
private void scheduleExpirationRenewal(final long threadId) {
    if (expirationRenewalMap.containsKey(getEntryName())) {
        return;
    }


     这个任务会延迟10s执行
    Timeout task = commandExecutor.getConnectionManager().newTimeout(new TimerTask() {
        @Override
        public void run(Timeout timeout) throws Exception {


             这个操作会将key的过期时间重新设置为30s
            RFuture<Boolean> future = renewExpirationAsync(threadId);


            future.addListener(new FutureListener<Boolean>() {
                @Override
                public void operationComplete(Future<Boolean> future) throws Exception {
                    expirationRenewalMap.remove(getEntryName());
                    if (!future.isSuccess()) {
                        log.error("Can't update lock " + getName() + " expiration", future.cause());
                        return;
                    }


                    if (future.getNow()) {
                         reschedule itself
                         通过递归调用本方法，无限循环延长过期时间
                        scheduleExpirationRenewal(threadId);
                    }
                }
            });
        }


    }, internalLockLeaseTime  3, TimeUnit.MILLISECONDS);


    if (expirationRenewalMap.putIfAbsent(getEntryName(), new ExpirationEntry(threadId, task)) != null) {
        task.cancel();
    }
}

四、RedLock算法

除了要考虑分布式锁的实现方式，还要考虑实际redis是如何部署工作的。redis有3种部署方式：

单机部署
redis sentinel哨兵模式
redis cluster集群模式

如果采用单机模式的redis实现分布式锁，会存在单点问题，只要redis挂掉了，分布式锁就没用了。如果采用redis sentinel哨兵模式，加锁的时候只对一个节点加锁，即使通过sentinel做了高可用，如果master节点挂掉，发生了主从切换，也可能发生锁丢失的情况。redis cluster集群模式也无法避免锁丢失，当slave节点还没同步到锁数据时，master节点挂点了，同样发生锁丢失情况。

基于以上原因，redis的作者提出了RedLock算法，主要思想是在多个集群节点的mster节点都去加相同的锁，锁的过期时间设置的较短，一般几十毫秒，当过半的master节点加锁成功，则认为整个加锁是成功的，要是加锁失败了，则mster节点一次删除这个锁。意思就是只要有一个master节点加了锁，所有mster节点都要不断轮询去尝试获取锁。

这里采取的是最终一致性。redis的设计决定了数据并不是强一致性的，即使是redlock，在极端情况下，也不能保证所有master加锁的流程都正确。此外redis分布式锁的实现需要不断尝试获取锁，也是比较消耗性能的。

redisson也提供了对redlock算法的支持，用法也很简单：

RedissonClient redisson = Redisson.create(config);
RLock lock1 = redisson.getFairLock("lock1");
RLock lock2 = redisson.getFairLock("lock2");
RLock lock3 = redisson.getFairLock("lock3");
RedissonRedLock multiLock = new RedissonRedLock(lock1, lock2, lock3);
multiLock.lock();
multiLock.unlock();

五、分布式锁的设计思路

分布式锁的设计思路和线程同步锁ReentrantLock的思路是一样的。但是也需要考虑如以下几个问题：

互斥性
死锁情况
可重入性
容错性：锁永久有效问题
加锁解锁同一客户端：锁永久失效问题
锁的性能：数据库悲观锁
CAP：RedLock算法

接下来介绍下Jedis客户端工具如何实现正确的加锁与解锁，加深我们对Redis分布式锁的原理以及设计思路的理解。

正确的加锁姿势

public class RedisTool {    
   private static final String LOCK_SUCCESS = "OK";    
   private static final String SET_IF_NOT_EXIST = "NX";   
   private static final String SET_WITH_EXPIRE_TIME = "PX";    
   /**
     * 尝试获取分布式锁
     * @param jedis Redis客户端
     * @param lockKey 锁
     * @param requestId 请求标识
     * @param expireTime 超期时间
     * @return 是否获取成功     
     */
    public static boolean tryGetDistributedLock(Jedis jedis, String lockKey, String requestId, int expireTime) {
        String result = jedis.set(lockKey, requestId, SET_IF_NOT_EXIST, SET_WITH_EXPIRE_TIME, expireTime);
        if (LOCK_SUCCESS.equals(result)) { 
            return true;
        }        
        return false;
    }
}

set方法加入了NX参数，可以保证如果已有key存在，则函数不会调用成功，也就是只有一个客户端能持有锁，满足互斥性。其次，我们对锁设置了过期时间，即使锁的持有者后续发生崩溃而没有解锁，锁也会因为到了过期时间而自动解锁，不会发生死锁。最后，因为我们将value赋值为requestId，代表加锁的客户端请求标识，那么在客户端在解锁的时候就可以进行校验是否是同一个客户端。

正确的解锁姿势

public class RedisTool {
   private static final Long RELEASE_SUCCESS = 1L;    
   /**
     * @param jedis Redis客户端
     * @param lockKey 锁
     * @param requestId 请求标识
     * @return 是否释放成功     
     */
    public static boolean releaseDistributedLock(Jedis jedis, String lockKey, String requestId) {
        String script = "if redis.call('get', KEYS[1]) == ARGV[1] then return redis.call('del', KEYS[1]) else return 0 end";
        Object result = jedis.eval(script, Collections.singletonList(lockKey), Collections.singletonList(requestId));
        if (RELEASE_SUCCESS.equals(result)) {
            return true;
        }       
        return false;
    }
}