Redis分布式锁

作者: 学编程的小屁孩 | 来源:发表于2020-03-09 21:42 被阅读0次

前言

分布式锁一般有三种实现方式：1. 数据库乐观锁；2. 基于Redis的分布式锁；3. 基于ZooKeeper的分布式锁。本篇介绍如何正确地实现Redis分布式锁。

可靠性

首先，为了确保分布式锁可用，我们至少要确保锁的实现同时满足以下四个条件：

互斥性。在任意时刻，只有一个客户端能持有锁。
不会发生死锁。即使有一个客户端在持有锁的期间崩溃而没有主动解锁，也能保证后续其他客户端能加锁。
具有容错性。只要大部分的Redis节点正常运行，客户端就可以加锁和解锁。
解铃还须系铃人。加锁和解锁必须是同一个客户端，客户端自己不能把别人加的锁给解了。

代码实现

组件依赖

首先我们要通过Maven引入Jedis开源组件，在pom.xml文件加入下面的代码：

<dependency>
    <groupId>redis.clients</groupId>
    <artifactId>jedis</artifactId>
    <version>2.9.0</version>
</dependency>

加锁代码

正确姿势

Talk is cheap, show me the code。先展示代码，再带大家慢慢解释为什么这样实现：

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;
  }
}

可以看到，我们加锁就一行代码：jedis.set(String key, String value, String nxxx, String expx, int time)，这个set()方法一共有五个形参：

第一个为key，我们使用key来当锁，因为key是唯一的。
第二个为value，我们传的是requestId，很多童鞋可能不明白，有key作为锁不就够了吗，为什么还要用到value？原因就是我们在上面讲到可靠性时，分布式锁要满足第四个条件解铃还须系铃人，通过给value赋值为requestId，我们就知道这把锁是哪个请求加的了，在解锁的时候就可以有依据。requestId可以使用UUID.randomUUID().toString()方法生成。
第三个为nxxx，这个参数我们填的是NX，意思是SET IF NOT EXIST，即当key不存在时，我们进行set操作；若key已经存在，则不做任何操作；
第四个为expx，这个参数我们传的是PX，意思是我们要给这个key加一个过期的设置，具体时间由第五个参数决定。
第五个为time，与第四个参数相呼应，代表key的过期时间。

总的来说，执行上面的set()方法就只会导致两种结果：1. 当前没有锁（key不存在），那么就进行加锁操作，并对锁设置个有效期，同时value表示加锁的客户端。2. 已有锁存在，不做任何操作。

心细的童鞋就会发现了，我们的加锁代码满足我们可靠性里描述的三个条件。首先，set()加入了NX参数，可以保证如果已有key存在，则函数不会调用成功，也就是只有一个客户端能持有锁，满足互斥性。其次，由于我们对锁设置了过期时间，即使锁的持有者后续发生崩溃而没有解锁，锁也会因为到了过期时间而自动解锁（即key被删除），不会发生死锁。最后，因为我们将value赋值为requestId，代表加锁的客户端请求标识，那么在客户端在解锁的时候就可以进行校验是否是同一个客户端。由于我们只考虑Redis单机部署的场景，所以容错性我们暂不考虑。

错误示例1

比较常见的错误示例就是使用jedis.setnx()和jedis.expire()组合实现加锁，代码如下：

public static void wrongGetLock1(Jedis jedis, String lockKey, 
    String requestId, int expireTime) {

    Long result = jedis.setnx(lockKey, requestId); 
    if (result == 1) { 
      // 若在这里程序突然崩溃，则无法设置过期时间，将发生死锁
     jedis.expire(lockKey, expireTime);
    }
}

setnx()方法作用就是SET IF NOT EXIST，expire()方法就是给锁加一个过期时间。乍一看好像和前面的set()方法结果一样，然而由于这是两条Redis命令，不具有原子性，如果程序在执行完setnx()之后突然崩溃，导致锁没有设置过期时间。那么将会发生死锁。网上之所以有人这样实现，是因为低版本的jedis并不支持多参数的set()方法。

错误示例2

public static boolean wrongGetLock2(Jedis jedis, String lockKey, int expireTime) { 
    long expires = System.currentTimeMillis() + expireTime;
    String expiresStr = String.valueOf(expires); 
    // 如果当前锁不存在，返回加锁成功
    if (jedis.setnx(lockKey, expiresStr) == 1) { 
      return true;
    }

   // 如果锁存在，获取锁的过期时间
    String currentValueStr = jedis.get(lockKey); 
    if (currentValueStr != null 
      && Long.parseLong(currentValueStr) < System.currentTimeMillis()) { 
      // 锁已过期，获取上一个锁的过期时间，并设置现在锁的过期时间
     String oldValueStr = jedis.getSet(lockKey, expiresStr); 

    if (oldValueStr != null && oldValueStr.equals(currentValueStr)) { 
       // 并发时只有一个线程的设置值和当前值相同，它才有权利加锁
       return true;
    }
 } 

  // 其他情况，一律返回加锁失败
  return false;
}

这一种错误示例就比较难以发现问题，而且实现也比较复杂。实现思路：使用jedis.setnx()命令实现加锁，其中key是锁，value是锁的过期时间。执行过程：1. 通过setnx()方法尝试加锁，如果当前锁不存在，返回加锁成功。2. 如果锁已经存在则获取锁的过期时间，和当前时间比较，如果锁已经过期，则设置新的过期时间，返回加锁成功。代码如下：

那么这段代码问题在哪里？1. 由于是客户端自己生成过期时间，所以需要强制要求分布式下每个客户端的时间必须同步。 2. 当锁过期的时候，如果多个客户端同时执行jedis.getSet()方法，那么虽然最终只有一个客户端可以加锁，但是这个客户端的锁的过期时间可能被其他客户端覆盖。3. 锁不具备拥有者标识，即任何客户端都可以解锁。

解锁代码

正确姿势

还是先展示代码，再带大家慢慢解释为什么这样实现：

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;
  }
}

可以看到，我们解锁只需要两行代码就搞定了！第一行代码，我们写了一个简单的Lua脚本代码，上一次见到这个编程语言还是在《黑客与画家》里，没想到这次居然用上了。第二行代码，我们将Lua代码传到jedis.eval()方法里，并使参数KEYS[1]赋值为lockKey，ARGV[1]赋值为requestId。eval()方法是将Lua代码交给Redis服务端执行。

那么这段Lua代码的功能是什么呢？其实很简单，首先获取锁对应的value值，检查是否与requestId相等，如果相等则删除锁（解锁）。那么为什么要使用Lua语言来实现呢？因为要确保上述操作是原子性的。那么为什么执行eval()方法可以确保原子性，源于Redis的特性。简单来说，就是在eval命令执行Lua代码的时候，Lua代码将被当成一个命令去执行，并且直到eval命令执行完成，Redis才会执行其他命令。

错误示例1

最常见的解锁代码就是直接使用jedis.del()方法删除锁，这种不先判断锁的拥有者而直接解锁的方式，会导致任何客户端都可以随时进行解锁，即使这把锁不是它的。

public static void wrongReleaseLock1(Jedis jedis, String lockKey) {
    jedis.del(lockKey);
}

错误示例2

这种解锁代码乍一看也是没问题，甚至我之前也差点这样实现，与正确姿势差不多，唯一区别的是分成两条命令去执行，代码如下：

public static void wrongReleaseLock2(Jedis jedis, 
  String lockKey, String requestId) {
    // 判断加锁与解锁是不是同一个客户端
    if (requestId.equals(jedis.get(lockKey))) { 
      // 若在此时，这把锁突然不是这个客户端的，则会误解锁
     jedis.del(lockKey);
    }
}

如代码注释，问题在于如果调用jedis.del()方法的时候，这把锁已经不属于当前客户端的时候会解除他人加的锁。那么是否真的有这种场景？答案是肯定的，比如客户端A加锁，一段时间之后客户端A解锁，在执行jedis.del()之前，锁突然过期了，此时客户端B尝试加锁成功，然后客户端A再执行del()方法，则将客户端B的锁给解除了。

分布式锁

如果有了分布式锁就不会出现超卖问题，举个简单的例子，分布式就像是银行的ATM，你去ATM取钱，如果你去了恰巧里面有人，那么你肯定是进不去的，必须等里面的人出来之后你才能进去，同样的，如果你正在取钱，如果外面还有人等着取钱，那么无论如何他都必须等你取完出来，他才能进去，要么他不取了回家去。

Redis实现的分布式锁

简单的介绍了分布式和分布式锁，之后我们现在来看看如何实现一个分布式锁，先看我之前使用Redis写个一个分布式锁，看看有什么问题？

public Boolean lock(String key, Long waitTime, Long expireTime) {
 
  String value = UUID.randomUUID().toString().replaceAll("-", "").toLowerCase();
  Boolean flag = setNx(key, value, expireTime, TimeUnit.SECONDS);
 
  // 尝试获取锁 成功返回
  if (flag) {
    return flag;
  } else {
    // 现在时间
    long newTime = System.currentTimeMillis();

    // 等待过期时间
    long loseTime = newTime + waitTime;
 
    // 不断尝试获取锁成功返回 11 < 11
    while (System.currentTimeMillis() < loseTime) {
      Boolean testFlag = setNx(key, value, expireTime, TimeUnit.MILLISECONDS);
      if (testFlag) {
        return testFlag;
      }

      try {
        Thread.sleep(1000);
      } catch (InterruptedException e) {
 
      }
    }
  }
   return false;
}

首先尝试获取锁，如果获取到了同时设置过期时间（原子操作，防止死锁），获取到了直接返回，简单看没什么问题，但是如果说，我设置了10秒的过期时间，但是我的业务执行了12秒，那么此时其他线程也进来了，那么这个锁根本就没有起到任何作用，注意：大家千万不要说那我把过期时间设置为1分钟设置10分钟不就好了，千万不要这么干，治标不治本。其实办法肯定是有的，在获取到锁之后，然后开一个守护线程，判断当前线程时候执行结束，如果没有那就重置锁的过期时间。但是我们这里就不说了，而是直接用别人封装好的框架。

Redisson分布式锁

1、首先引入maven

<dependency>
    <groupId>org.redisson</groupId>
    <artifactId>redisson-spring-boot-starter</artifactId>
    <version>3.10.6</version>
</dependency>

注意：我这里引入的是redisson和springboot的集成包，网上一些教程可能是引入如下配置

<dependency>
    <groupId>org.redisson</groupId>
    <artifactId>redisson</artifactId>
    <version>3.11.5</version>
</dependency>

如果你引入的就是redisson的依赖包，如果该依赖包的版本低于3.5会需要你再引入
<dependency>
<groupId>io.netty</groupId>
<artifactId>netty-all</artifactId>
<version>4.1.25.Final</version>
</dependency>
<dependency>
<groupId>com.fasterxml.jackson.core</groupId>
<artifactId>jackson-core</artifactId>
<version>2.9.0</version>
</dependency>
<dependency>
<groupId>com.fasterxml.jackson.core</groupId>
<artifactId>jackson-databind</artifactId>
<version>2.9.0</version>
</dependency>

2、构建Redisson实例（基于SpringBoot）

配置redis信息

spring:
  application:
    name: spring-cloud-product
  redis:
    port: 6379
    host: 127.0.0.1
    password:
    database: 0
    timeout: 2000

配置redisson

新建一个redisson-single.yml的配置文件下面是单机配置

singleServerConfig:
  idleConnectionTimeout: 10000
  pingTimeout: 1000
  connectTimeout: 10000
  timeout: 3000
  retryAttempts: 3
  retryInterval: 1500
  reconnectionTimeout: 3000
  failedAttempts: 3
  password: null
  subscriptionsPerConnection: 5
  clientName: null
  address: "redis://127.0.0.1:6379"
  subscriptionConnectionMinimumIdleSize: 1
  subscriptionConnectionPoolSize: 50
  connectionMinimumIdleSize: 32
  connectionPoolSize: 64
  database: 0
  #在最新版本中dns的检查操作会直接报错 所以我直接注释掉了
  #dnsMonitoring: false
  dnsMonitoringInterval: 5000
threads: 0
nettyThreads: 0
codec: !<org.redisson.codec.JsonJacksonCodec> {}
transportMode : "NIO"

RedissonConfig配置类来配置redisson

@Configuration
public class RedssonConfig {
    @Bean(destroyMethod="shutdown")
    public RedissonClient redisson() throws IOException {
        RedissonClient redisson = Redisson.create(Config.fromYAML(
          new ClassPathResource("redisson-single.yml").getInputStream()));
        return redisson;
    }
}

3、使用Redisson

其实单从使用来说是非常简单的，这里需要跟大家说一点，就是lock()这个方法，我们先看源码，这个方法是可以设置过期时间的，但是它并不会去进行检查任务是否执行结束，如果任务没有执行结束，然后锁的过期时间到了，线程中断，就会出现异常。

public void lock(long leaseTime, TimeUnit unit) {
        try {
            this.lock(leaseTime, unit, false);
        } catch (InterruptedException var5) {
            throw new IllegalStateException();
        }
    }

异常信息

超时异常

4、较为完善的方法

这个方法不需要传递任何参数，它的底层方法会通过我们刚才说的原理，去校验当前任务是否执行结束，如果没有执行结束，那么相应的就会延长锁的过期时间。

底层实现

private void lock(long leaseTime, TimeUnit unit, boolean interruptibly)
  throws InterruptedException {

  long threadId = Thread.currentThread().getId();
  Long ttl = this.tryAcquire(leaseTime, unit, threadId);
  if (ttl != null) {
  RFuture<RedissonLockEntry> future = this.subscribe(threadId);
  this.commandExecutor.syncSubscription(future);

  try {
    // 校验任务是否执行结束
    while(true) {
      ttl = this.tryAcquire(leaseTime, unit, threadId);
      if (ttl == null) {
        return;
      }

      if (ttl >= 0L) {
        try {
         this.getEntry(threadId).getLatch().tryAcquire(ttl, TimeUnit.MILLISECONDS);
         } catch (InterruptedException var13) {
           if (interruptibly) {
             throw var13;
           }

         this.getEntry(threadId).getLatch().tryAcquire(ttl, TimeUnit.MILLISECONDS);
        }
      } else if (interruptibly) {
        this.getEntry(threadId).getLatch().acquire();
      } else {
        this.getEntry(threadId).getLatch().acquireUninterruptibly();
      }
    }
  } finally {
    this.unsubscribe(future, threadId);
  }
 }
}

五、秒杀案例

编写一个秒杀接口

@Autowired
private RedissonClient redissonClient;
 
@Override
public boolean decrementProductStore(Long productId, Integer productQuantity) {
    String key = "dec_store_lock_" + productId;
    RLock lock = redissonClient.getLock(key);
    try {
        //加锁 操作很类似Java的ReentrantLock机制
        lock.lock();
        ProductInfo productInfo = productInfoMapper.selectByPrimaryKey(productId);
        //如果库存为空
        if (productInfo.getProductStock() == 0) {
           return false;
        }
        //简单减库存操作 没有重新写其他接口了
        productInfo.setProductStock(productInfo.getProductStock() - 1);
        productInfoMapper.updateByPrimaryKey(productInfo);
    } catch (Exception e) {
        System.out.println(e.getMessage());
    } finally {
        //解锁
        lock.unlock();
    }
    return true;
}

写一个简单的测试请求

@GetMapping("test")
public String createOrderTest() {
    if (!productInfoService.decrementProductStore(1L, 1)) {
        return "库存不足";
    }
    OrderMaster orderMaster = new OrderMaster();
    //未支付
    orderMaster.setOrderStatus(0);
    //未支付
    orderMaster.setPayStatus(0);
    orderMaster.setBuyerName(name);
    orderMaster.setBuyerAddress("湖南长沙");
    orderMaster.setBuyerPhone("18692794847");
    orderMaster.setOrderAmount(BigDecimal.ZERO);
    orderMaster.setCreateTime(DateUtils.getCurrentDate());
    orderMaster.setOrderId(UUID.randomUUID().toString().replaceAll("-", ""));
    orderMasterService.insert(orderMaster);
    return "创建订单成功";
}

使用ab做接口测试

image.png

ab -n 300 -c 300 请求地址；-n 的含义就是你做多少个请求；-c 的含义就是多少个用户并发请求。数据库中的商品已经全部被秒杀完并未出现超库存的情况。如果对ab不是太了解可以看看这篇文章（ab）。Redisson还有其他好多的方法来解决现在互联网中的好多问题，可以去Redisson官网学习。

附：

RedissonLocker的具体实现类

public class RedissonLockerImpl implements RedissonLocker {
 
    @Autowired
    private RedissonClient redissonClient;
    
    /**************************可重入锁**************************/
    
    /**
     * 拿不到lock就不罢休，不然线程就一直block
     * 没有超时时间,默认30s
     * 
     * @param lockKey
     * @return
     */
    @Override
    public RLock lock(String lockKey) {
        RLock lock = redissonClient.getLock(lockKey);
        lock.lock();
        return lock;
    }
 
    /**
     * 自己设置超时时间
     * 
     * @param lockKey 锁的key
     * @param timeout 秒 如果是-1，直到自己解锁，否则不会自动解锁
     * @return
     */
    @Override
    public RLock lock(String lockKey, int timeout) {
        RLock lock = redissonClient.getLock(lockKey);
        lock.lock(timeout, TimeUnit.SECONDS);
        return lock;
    }
    
    /**
     * 自己设置超时时间
     * 
     * @param lockKey 锁的key
     * @param unit 锁时间单位
     * @param timeout 超时时间
     * 
     */
    @Override
    public RLock lock(String lockKey, TimeUnit unit, int timeout) {
        RLock lock = redissonClient.getLock(lockKey);
        lock.lock(timeout, unit);
        return lock;
    }
 
    /**
     *  尝试加锁，最多等待waitTime，上锁以后leaseTime自动解锁
     * @param lockKey   锁key
     * @param unit      锁时间单位
     * @param waitTime  等到最大时间，强制获取锁
     * @param leaseTime 锁失效时间
     * @return 如果获取成功，则返回true，如果获取失败（即锁已被其他线程获取），则返回false
     */
    @Override
    public boolean tryLock(String lockKey, TimeUnit unit, 
          int waitTime, int leaseTime) {
        RLock lock = redissonClient.getLock(lockKey);
        try {
            return lock.tryLock(waitTime, leaseTime, unit);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        return false;
    }
    
    
    
    /**************************公平锁**************************/
    /**
     *   尝试加锁，最多等待waitTime，上锁以后leaseTime自动解锁
     * @param lockKey   锁key
     * @param unit      锁时间单位
     * @param waitTime  等到最大时间，强制获取锁
     * @param leaseTime 锁失效时间
     * @return 如果获取成功，则返回true，如果获取失败（即锁已被其他线程获取），则返回false
     */
    public boolean fairLock(String lockKey, TimeUnit unit, 
      int waitTime, int leaseTime) {
        RLock fairLock = redissonClient.getFairLock(lockKey);
        try {
            return fairLock.tryLock(waitTime, leaseTime, unit);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        return false;
    }
 
    /**
     * 释放锁
     * @param lockKey 锁key
     */
    @Override
    public void unlock(String lockKey) {
        RLock lock = redissonClient.getLock(lockKey);
        lock.unlock();
    }
 
    /**
     * 释放锁
     */
    @Override
    public void unlock(RLock lock) {
        lock.unlock();
    }
}
业务代码中使用
String lockKey = userId;
// 公平加锁，60秒后锁自动释放
boolean isLocked = false;
try {
    isLocked = redissonLocker.fairLock(lockKey , TimeUnit.SECONDS, 3, 60);
    if (isLocked) { // 如果成功获取到锁就继续执行
        // 执行业务代码操作
        return GlobalResponse.success();
    } else { // 未获取到锁
        return GlobalResponse.fail(500, "请勿重复点击！！");
    }
} catch (Exception e) {
    return GlobalResponse.fail(500, e.getMessage());
} finally {
    if (isLocked) { // 如果锁还存在，在方法执行完成后，释放锁
        redissonLocker.unlock(lockKey);
    }
}

Redis分布式锁

前言

可靠性

代码实现

组件依赖

加锁代码

正确姿势

错误示例1

错误示例2

解锁代码

正确姿势

错误示例1

错误示例2

分布式锁

Redis实现的分布式锁

Redisson分布式锁

1、首先引入maven

2、构建Redisson实例（基于SpringBoot）

配置redis信息

配置redisson

RedissonConfig配置类来配置redisson

3、使用Redisson

4、较为完善的方法

五、秒杀案例

附：

相关文章

网友评论

延伸阅读

深度阅读

栏目导航

热点阅读