redis分布式锁实现---基于redisson封装自己的分布式锁

Posted 2021-08-31 wen-pan

一、介绍

对于使用Redis做分布式锁的简单实现，在上篇我们已经介绍了如何通过Redis命令 + lua脚本来一步步的实现一个简单的分布式锁。并且对于每种实现方案的优缺点进行了逐一分析。其实这些缺陷也是Redis做分布式锁的常见缺点。

在生产上我们一般不会自己从头开始去实现一个分布式锁，毕竟需要考虑的问题以及成本太高了。好在已经有很多框架帮我们实现好了各种分布式锁（比如Redisson），一般来说在我们自己的产品中只需要将redisson提供的相关功能进行封装，提供一些动态配置项来适配redisson提供的对应着redis的几种部署模式的配置即可。

二、Redisson介绍

引用至：百度百科

Redisson采用了基于NIO的Netty框架，不仅能作为Redis底层驱动客户端，具备提供对Redis各种组态形式的连接功能，对Redis命令能
以同步发送、异步形式发送、异步流形式发送或管道形式发送的功能，LUA脚本执行处理，以及处理返回结果的功能，还在此基础上融入了更
高级的应用方案，不但将原生的RedisHash，List，Set，String，Geo，HyperLogLog等数据结构封装为Java里大家最熟悉的映射
（Map），列表（List），集（Set），通用对象桶（Object Bucket），地理空间对象桶（Geospatial Bucket），基数估计算法
（HyperLogLog）等结构，在这基础上还提供了分布式的多值映射（Multimap），本地缓存映射（LocalCachedMap），有序集
（SortedSet），计分排序集（ScoredSortedSet），字典排序集（LexSortedSet），列队（Queue），阻塞队列（Blocking 
Queue），有界阻塞列队（Bounded Blocking Queue），双端队列（Deque），阻塞双端列队（Blocking Deque），阻塞公平列队
（Blocking Fair Queue），延迟列队（Delayed Queue），布隆过滤器（Bloom Filter），原子整长形（AtomicLong），原子双
精度浮点数（AtomicDouble），BitSet等Redis原本没有的分布式数据结构。不仅如此，Redisson还实现了Redis文档中提到像分布式
锁Lock这样的更高阶应用场景。事实上Redisson并没有不止步于此，在分布式锁的基础上还提供了联锁（MultiLock），读写锁
（ReadWriteLock），公平锁（Fair Lock），红锁（RedLock），信号量（Semaphore），可过期性信号量
（PermitExpirableSemaphore）和闭锁（CountDownLatch）这些实际当中对多线程高并发应用至关重要的基本部件。正是通过实现基
于Redis的高阶应用方案，使Redisson成为构建分布式系统的重要工具。

在提供这些工具的过程当中，Redisson广泛的使用了承载于Redis订阅发布功能之上的分布式话题（Topic）功能。使得即便是在复杂的分
布式环境下，Redisson的各个实例仍然具有能够保持相互沟通的能力。在以这为前提下，结合了自身独有的功能完善的分布式工具，
Redisson进而提供了像分布式远程服务（Remote Service），分布式执行服务（Executor Service）和分布式调度任务服务
（Scheduler Service）这样适用于不同场景的分布式服务。** 使得Redisson成为了一个基于Redis的Java中间件 **
（Middleware）。

简单来说就是：redisson是一个基于java编程框架netty进行扩展、封装、增强redis的一个Java中间件。即我们可以通过redisson这个中间件去方便的操作Redis。由于redisson底层是基于netty的，所以想了解Redisson源码***首先必须要熟悉netty网络编程框架***。

三、Redis的部署方式

理解redisson前我们需要先了解一下Redis的几种部署模式。

1、Redis五种部署方式

• 单节点部署

• 主从方式部署

• 哨兵方式部署

• 集群方式部署

• 云托管模式部署

2、简单介绍

• 最开始的Redis是单个节点部署的
• 基于单节点部署为了保证数据的备份，一般会添加一个节点作为slave来备份master节点上的数据，这里就衍生出了主从部署方式
• 虽然有了主从结构，但是从节点只提供数据备份和读取功能，一旦master宕机了那么服务将会变得不可用。
• 所以，基于主从结构的缺点，衍生出了哨兵模式，通过哨兵去监控主节点，一旦主节点挂了，那么将重新从所有的slave节点中选出一个最合适的节点当做新的master节点。
• 虽然哨兵模式能够进行故障转移，在一定程度上保证了redis服务的高可用。但是对于Redis的压力方面并没有得到很大的改善（读取操作可以负载均衡到slave节点帮助master节点分担压力，但是所有的写入操作的压力都在master节点），并且单个master节点的磁盘容量、内存容量也是有限。所以为了进一步的拓展Redis的性能，于是衍生出了集群模式（多个master节点和多个slave节点）。
• 一些云服务商像阿里云、微软云和亚马逊云都基于原生redis做了高可用部署，为了能连接云服务商的redis服务，这里redisson也提供了API操作方式。

四、基于redisson去封装自己的分布式锁

项目gitee地址：https://gitee.com/mr_wenpan/stone-monster-backend

1、如何封装

• 首先我们要知道对于分布式锁的获取、释放、判断是否被占有等这些操作在redisson中已经帮我们做了很好的实现，并且各种问题都已经考虑到了，并且还能够基于配置信息去适配不同的Redis部署模式，可以说是非常全面了。
• 我们要做的其实就是读取服务中的redis相关配置（比如：在application.yml配置文件中指定的Redis的部署模式、slave节点master节点地址、连接池大小、超时时间、负载均衡模式等）。
• 然后基于这些用户指定的配置去构建redisson的配置文件即可。
• redisson会基于我们的配置去连接redis、创建连接池、控制超时时间等。

其实就是简单一句话，读取application.yml配置文件中的相关配置，设置到redisson提供的对应配置类上即可（redisson的配置项比较繁杂）。

2、具体实现

以Redis部署模式为单机版为例！

①、引入相关依赖

<dependencies>
  <!--基于redisson实现的分布式锁进行封装，需要引入redisson依赖-->
  <dependency>
    <groupId>org.redisson</groupId>
    <artifactId>redisson</artifactId>
  </dependency>
  <!--自动配置-->
  <dependency>
    <groupId>org.springframework.boot</groupId>
    <artifactId>spring-boot-autoconfigure</artifactId>
  </dependency>
  <dependency>
    <groupId>org.springframework.boot</groupId>
    <artifactId>spring-boot-configuration-processor</artifactId>
    <optional>true</optional>
  </dependency>
  <dependency>
    <groupId>org.projectlombok</groupId>
    <artifactId>lombok</artifactId>
    <scope>provided</scope>
  </dependency>
  <dependency>
    <groupId>org.apache.commons</groupId>
    <artifactId>commons-lang3</artifactId>
  </dependency>
</dependencies>

②、创建配置文件映射application.yml中的相关配置

@Data
@ConfigurationProperties(prefix = LockConfigProperties.PREFIX)
public class LockConfigProperties {

    public static final String PREFIX = "stone.redis.lock";

    /**
     * Redis的运行模式,默认使用单机模式
     */
    private String pattern = ServerPattern.SINGLE.getPattern();
    /**
     * 单节点模式
     */
    private SingleConfig singleServer;
    /**
     * 集群模式
     */
    private ClusterConfig clusterServer;
    /**
     * 云托管模式
     */
    private ReplicatedConfig replicatedServer;
    /**
     * 哨兵模式
     */
    private SentinelConfig sentinelServer;
    /**
     * 主从模式
     */
    private MasterSlaveConfig masterSlaveServer;
    /**
     * 客户端名称
     */
    private String clientName = LockConstant.LOCK_CLIENT_NAME;
    /**
     * 启用SSL终端识别
     */
    private boolean sslEnableEndpointIdentification = true;
    /**
     * SSL实现方式，确定采用哪种方式（JDK或OPENSSL）来实现SSL连接
     */
    private String sslProvider = LockConstant.JDK;
    /**
     * SSL信任证书库路径
     */
    private String sslTruststore;
    /**
     * SSL信任证书库密码
     */
    private String sslTruststorePassword;
    /**
     * SSL钥匙库路径
     */
    private String sslKeystore;
    /**
     * SSL钥匙库密码
     */
    private String sslKeystorePassword;
    /**
     * 锁信息配置
     */
    private Property property = new Property();
    /**
     * RTopic共享线程数量
     */
    private int threads = 0;
    /**
     * Redisson使用的所有redis客户端之间共享的线程数量
     */
    private int nettyThreads = 0;
    /**
     * 仅在没有leaseTimeout参数定义的情况下获取锁定时才使用此参数。看门狗过期时间
     */
    private long lockWatchdogTimeout = 30000;
    /**
     * 是否串行处理消息
     */
    private boolean keepPubSubOrder = true;
    /**
     * 否在Redis端使用Lua脚本缓存
     */
    private boolean useScriptCache = false;
 		
  	@Data
    public static class SingleConfig {
        /**
         * 节点地址
         */
        private String address;
        /**
         * 节点端口
         */
        private int port = 6379;
        /**
         * 发布和订阅连接的最小空闲连接数
         */
        private int subConnMinIdleSize = 1;
        /**
         * 发布和订阅连接池大小
         */
        private int subConnPoolSize = 50;
        /**
         * 最小空闲连接数
         */
        private int connMinIdleSize = 32;
        /**
         * 连接池大小
         */
        private int connPoolSize = 64;
        /**
         * 是否启用DNS监测
         */
        private boolean dnsMonitoring = false;
        /**
         * DNS监测时间间隔，单位：毫秒，该配置需要dnsMonitoring设为true
         */
        private int dnsMonitoringInterval = 5000;
        /**
         * 连接空闲超时，单位：毫秒
         */
        private int idleConnTimeout = 10000;
        /**
         *
         */
        private boolean keepAlive = false;
        /**
         * 连接超时，单位：毫秒
         */
        private int connTimeout = 10000;
        /**
         * 命令等待超时，单位：毫秒
         */
        private int timeout = 3000;
        /**
         * 命令失败重试次数 如果尝试达到 retryAttempts（命令失败重试次数） 
         * 仍然不能将命令发送至某个指定的节点时，将抛出错误。如果尝试在此限制之内发送成功，则开始启用
         * timeout（命令等待超时） 计时。
         */
        private int retryAttempts = 3;
        /**
         * 命令重试发送时间间隔，单位：毫秒
         */
        private int retryInterval = 1500;
        /**
         * 数据库编号
         */
        private int database = 0;
        /**
         * 密码
         */
        private String password;
        /**
         * 单个连接最大订阅数量
         */
        private int subPerConn = 5;
    }
}

③、编写自动配置类基于上面的配置构建redisson的RedissonClient

将上面从application.yml配置文件中读取到的配置信息，用来构建一个RedissonClient并注入到容器中。后面我们获取不同类型的锁会用到这个RedissonClient。

@Configuration
@EnableConfigurationProperties(LockConfigProperties.class)
public class RedisLockAutoConfiguration {

    /**
     * 注入Redis分布式锁的配置属性
     */
    @Autowired
    private LockConfigProperties lockConfig;

    /**
     * 按不同的配置决定注入的RedissonClient，后面加锁需要基于RedissonClient去封装
     *
     * @return org.redisson.api.RedissonClient
     * @author [e-mail] 2021/7/25 3:24 下午
     */
    @Bean(name = "lockRedissonClient", destroyMethod = "shutdown")
    @ConditionalOnMissingBean
    RedissonClient redisson() throws Exception {
        // ====================================不同模式的公有配置====================================
        Config config = new Config();
        // RTopic共享线程数量
        config.setThreads(lockConfig.getThreads());
        // Redisson使用的所有redis客户端之间共享的线程数量
        config.setNettyThreads(lockConfig.getNettyThreads());
        // 仅在没有leaseTimeout参数定义的情况下获取锁定时才使用此参数。看门狗过期时间
        config.setLockWatchdogTimeout(lockConfig.getLockWatchdogTimeout());
        // 是否串行处理消息
        config.setKeepPubSubOrder(lockConfig.getKeepPubSubOrder());
        // 否在Redis端使用Lua脚本缓存
        config.setUseScriptCache(lockConfig.getUseScriptCache());
        // 根据用户在配置文件中配置的pattern模式获取分布式锁的服务模式（单机、集群、哨兵、主从）
        ServerPattern serverPattern = ServerPatternFactory.getServerPattern(lockConfig.getPattern());

        // ====================================不同模式的特有配置====================================
        // 为了创建不同的连接池（MasterConnectionPool、MasterPubSubConnectionPool、
        // SlaveConnectionPool和PubSubConnectionPool）
        // 参考：https://blog.csdn.net/zilong_zilong/article/details/78609423

        // 适配不同的Redis部署模式
        switch (serverPattern) {
            // redis采用单机模式部署
            case SINGLE:
                // 指定使用单节点部署方式，通过config获取单机模式server配置
                SingleServerConfig singleServerConfig = config.useSingleServer();
                ServerConfigInitFactory.initSingleConfig(singleServerConfig, lockConfig);
                break;
                // todo 这里后面可接着实现主从模式、集群模式、哨兵模式、云托管模式下的分布式锁
            default:
                break;
        }
        // 根据自己定义的配置创建一个RedissonClient实例并注入容器（这一步非常耗时，预计1~2秒）
        return Redisson.create(config);
    }
}

④、公平锁锁实现

好了，上面三步我们已经将redisson配置配置好了，并且将RedissonClient注入到了容器中，下面我们就可以基于RedissonClient去获取redisson实现的各种类型的锁了（重入锁、公平锁、读锁、写锁、红锁、联锁等）。

以公平锁为例，这里我们自己对redisson提供的公平锁实现做一个简单的封装，以便在我们项目中各个模块需要使用到公平锁的时候能够统一调用方式，无需在每个服务中都再自己手动配置redisson的繁杂配置（因为在我们自己的基于redisson封装的锁组件模块中已经对redisson的那些繁杂的配置赋了默认值）。只需要引入我们自己基于redisson封装的锁模块，然后在application.yml中配置一下redis地址、端口、密码等几个常用配置即可方便的使用分布式锁。

1、定义获取锁需要的必须信息

我们需要定义一个对象来存放获取锁时所需要的必要信息（比如：要获取锁的名称、锁的过期时间、等待时间、时间单位等）

// 锁基本信息
@Data
@Builder
@NoArgsConstructor
public class LockInfo {

    /**
     * 锁的名称
     */
    private String name;
    /**
     * 获取锁的线程唯一标识（可重入锁使用）
     */
    private String threadUniqueIdentifier;
    /**
     * 等待时间
     */
    private long waitTime;
    /**
     * 锁过期自动释放时间
     */
    private long leaseTime;
    /**
     * 锁等待的时间单位（默认秒）
     */
    private TimeUnit timeUnit = TimeUnit.SECONDS;
    /**
     * key集合，用于红锁和联锁
     */
    private List<String> keyList;

    public LockInfo(String name, long waitTime, long leaseTime, TimeUnit timeUnit) {
        this.name = name;
        this.waitTime = waitTime;
        this.leaseTime = leaseTime;
        this.timeUnit = timeUnit;
    }

    public LockInfo(String name, List<String> keyList, long waitTime, long leaseTime, TimeUnit timeUnit) {
        this.name = name;
        this.keyList = keyList;
        this.waitTime = waitTime;
        this.leaseTime = leaseTime;
        this.timeUnit = timeUnit;
    }

}

2、定义锁服务接口

定义一个锁服务接口，该接口提供获取锁和释放锁以及设置锁信息三个方法，用户需要获取分布式锁的时候只需要构建锁信息（LockInfo）对象，然后调用锁服务接口的lock方法即可。

public interface LockService {

    /**
     * 添加锁信息
     *
     * @param lockInfo
     */
    void setLockInfo(LockInfo lockInfo);

    /**
     * 加锁
     *
     * @return boolean
     */
    boolean lock();

    /**
     * 释放锁
     *
     * @return void
     */
    void releaseLock();
}

3、公平锁实现

只需要通过我们前面构建redisson配置时注入的redissonClient对象，调用redissonClient.getFairLock()方法边可以获取到redisson实现的公平锁对象，然后调用tryLock即可。

// 自己封装基于redisson的公平锁实现
@Slf4j
public class FairLockServiceImpl implements LockService {

    @Autowired
    @Qualifier("lockRedissonClient")
    private RedissonClient redissonClient;

    private RLock rLock;

    private LockInfo lockInfo;

    @Override
    public void setLockInfo(LockInfo lockInfo) {
        this.lockInfo = lockInfo;
    }

    @Override
    public boolean lock() {
        try {
            // 获取锁对象，通过RedissonLock来获取锁（RedissonLock实现了RLock接口）
            rLock = redissonClient.getFairLock(lockInfo.getName());
            // 加锁
            return rLock.tryLock(lockInfo.getWaitTime(), lockInfo.getLeaseTime(), lockInfo.getTimeUnit());
        } catch (InterruptedException e) {
            log.info("获取公平锁时线程被意外中断，锁名称：{}，异常信息：{}", lockInfo.getName(), e);
        }
        return false;
    }

    @Override
    public void releaseLock() {
        // 如果该锁是被自己持有的
        if (rLock.isHeldByCurrentThread()) {
            rLock.unlockAsync();
        }
    }

}

4、将自己实现的公平锁注入到容器

上面我们已经基于redisson封装好了自己的公平锁实现，但是如何提供给其他服务使用呢？这时候我们就需要使用到自动配置，将我们的公平锁实现类注入到容器中。

@Bean
@Scope("prototype")
public FairLockServiceImpl fairLockService() {
    return new FairLockServiceImpl();
}

特别注意：这里作用于@Scope需要使用prototype，为什么呢？因为我们的公平锁实现中包含了一个LockInfo对象用来接收锁参数，但是正是因为这个对象的原因使得FairLockServiceImpl从无状态变成了有状态。在有状态的情况下便会导致多线程安全问题。所以我们要使用prototype来保证在每次使用公平锁时都是新产生的公平锁对象，而不是单例。（当然这里在上面的设计上也可以不需要LockInfo对象，在调用tryLock()方法的时候直接将锁需要的参数从外部传进来即可。）

到这里我们基于redisson自己去实现一个公平锁就实现完成了！！！

五、如何使用自己封装的分布式锁

1、引入自己封装的starter依赖

<dependency>
  <groupId>com.stone.monster</groupId>
  <artifactId>stone-starter-redis-lock</artifactId>
</dependency>

2、配置application.yml

spring:
  application:
    name: stone-redis-lock-test
  # 配置redis服务器相关信息
  redis:
    host: ${SPRING_REDIS_HOST:wenpan-host}
    port: ${SPRING_REDIS_PORT:6379}
    password: ${SPRING_REDIS_PASSWORD:xxxx}
    database: ${SPRING_REDIS_DATABASE:1}
    jedis:
      pool:
        max-active: ${SPRING_REDIS_POOL_MAX_ACTIVE:16}
        max-idle: ${SPRING_REDIS_POOL_MAX_IDLE:16}
        max-wait: ${SPRING_REDIS_POOL_MAX_WAIT:5000}
# 配置分布式锁所需要的信息
stone:
  redis:
    lock:
      host: ${SPRING_REDIS_HOST:wenpan-host}
      port: ${SPRING_REDIS_PORT:6379}
      password: ${SPRING_REDIS_PASSWORD:xxxx}
      cluster-mode: false
      timeout-millis: ${STONE_REDIS_COMMAND_TIMEOUT_MILLIS:3000}
      pool-min-idle: ${STONE_REDIS_POOL_MIN_IDLE:4}
      pool-max-idle: ${STONE_REDIS_POOL_MAX_IDLE:8}
      pool-max-wait-millis: ${STONE_REDIS_POOL_MAX_WAIT_MILLIS:30000}

3、编写测试类

@Slf4j
@Service
public class RedisFairLockTestService {

    @Autowired
    private FairLockServiceImpl fairLockService;

    public void test() {
        LockInfo lockInfo = LockInfo.builder()
          .name("stone:test:redis:test-lock")
          .waitTime(60L)
          .leaseTime(30L).build();
        fairLockService.setLockInfo(lockInfo);
      	// 获取锁
        boolean isLocked = fairLockService.lock();
        if (isLocked) {
            // do something xxx
        } else {
            log.warn("等待 {} 秒后，获取锁 {} 仍然失败！", lockInfo.getWaitTime(), lockInfo.getName());
        }
    }
}