常规项目都是采用Redission来实现分布式锁，进行分布式系统中资源竞争加锁操作。需要单独引入Jar包，偶然发现SpringBoot中的integration也实现多种载体的分布式锁控制。

代码集成

引入

// 分布式锁
implementation 'org.springframework.boot:spring-boot-starter-integration'
implementation('org.springframework.integration:spring-integration-redis')

采用最常见的redis来作为分布式锁的底层载体。

锁注册

在@Configuration配置类中，添加分布式锁注册信息。

@Bean
open fun redisLockRegistry(redisConnectionFactory: RedisConnectionFactory): RedisLockRegistry {

    return RedisLockRegistry(redisConnectionFactory, "fcDistroLock", 20000L)
}

有两个核心参数，第一个指定的是分布式锁的前缀，第二个是指定分布式锁的过期时间。过期时间建议不要指定到过长，防止拖慢整体的业务响应速度。

加锁

在使用之前需要知道加锁的三个核心方法。

lock	直接加锁，一直等待
tryLock(无参数）	尝试加锁，未获取到锁，直接返回失败
tryLock(long time, TimeUnit unit)	尝试加锁，等待一定时间后未获取到锁，直接返回失败

建议使用带参数的尝试加锁，设置一个合适的超时时间。建议使用模式如下

Lock lock = ...;
if (lock.tryLock(time)) {
  try {
    // manipulate protected state
  } finally {
    lock.unlock();
  }
} else {
  // perform alternative actions
}}

有一个点需要注意，当加锁失败时，需要考虑补偿机制。例如用户余额扣减失败，需要重新进行推送；或者加锁失败，抛出异常回滚本地事务等。

使用上非常简单。

细粒度加锁

可以通过上图可以看到，我们加锁的对象是用户id，并不是所有用户。代表不同用户之间操作是不受分布式事务限制。这里同步会衍生另外一个问题，如果用户id特别多，就会占用非常多的资源。这里就需要定时手动清除加锁对象，或者加锁成功后直接清除。个人建议使用定时清除，有助于减少对象的创建，提高系统吞吐量。

@Scheduled(cron = "0 0 0/1 * * ?")
fun scheduleRemoveRedisLock() {

    redisLockRegistry.expireUnusedOlderThan(1000 * 60 * 60)
}

RedisLockRegistry其实已经提供清除的方法，我们只需要指定清除的有效期即可。项目中指定的是清除1个小时之前的加锁对象。

核心逻辑

打开tryLock的实现类RedisLock很容易发现，每个加锁id都对应1个RedisLock，1个RedisLock中包含1个ReentrantLock，用来进行本地资源互斥。

public boolean tryLock(long time, TimeUnit unit) throws InterruptedException {
   long now = System.currentTimeMillis();
   if (!this.localLock.tryLock(time, unit)) { // 获取本地互斥锁
      return false;
   }
   try {
      long expire = now + TimeUnit.MILLISECONDS.convert(time, unit);
      boolean acquired;
      while (!(acquired = obtainLock()) && System.currentTimeMillis() < expire) { //NOSONAR
         Thread.sleep(100); //NOSONAR 防止请求过快，进行100Ms的休眠
      }
      if (!acquired) {
         this.localLock.unlock();
      }
      return acquired;
   }
   catch (Exception e) {
      this.localLock.unlock();
      rethrowAsLockException(e);
   }
   return false;
}

两个条件跳出循环获取锁的过程。

1. 超过等待时间
2. redis返回是否获取到锁

Redis锁逻辑判断

private static final String OBTAIN_LOCK_SCRIPT =
      "local lockClientId = redis.call('GET', KEYS[1])\n" +
            "if lockClientId == ARGV[1] then\n" +
            "  redis.call('PEXPIRE', KEYS[1], ARGV[2])\n" +
            "  return true\n" +
            "elseif not lockClientId then\n" +
            "  redis.call('SET', KEYS[1], ARGV[1], 'PX', ARGV[2])\n" +
            "  return true\n" +
            "end\n" +
            "return false";

利用Redis的原子性进行锁资源判断，通过是否相同应用id来支持重入锁。

整体使用

使用上非常简单，没有锁续期，没有读写锁，也没有考虑重入锁的计数问题。功能上还是比Redission差不少，在一些业务相对比较简单的场景可以尝试使用SpringBoot自带的分布式锁。如果需要面对更细粒度的控制，提高性能，更复杂的锁控制，就需要使用到Redission来进行分布式锁的编写了。