Redisson分布式锁

Redisson分布式锁

问题背景描述

在最近的一个项目中，本地开发、测试环境下，都是好好的，没出什么差错！！！^_^

可是一上线后，有些业务数据，重复记录了多遍！！！=_=

项目总监：谁谁谁呀，这咋回事，为什么业务数据会重复出现多条呢？

我：这.... 啊吧啊吧！！！

然后，经过一段分析操作，发现：

这些重复的业务数据，是由定时器任务执行产生，然后正式环境下，这个微服务集群部署了3个节点，导致每次定时器任务执行后，3个节点都执行一次，那就每次都产生了3条重复的业务数据！！！

还是太年轻了呀，没考虑分布式部署的问题！！！

可这，谁能想到他正式环境，部署了多个节点？=_=

这，你还怪部署环境的人了？看我反手就给你个...

好了，出了问题，咋们得想办法解决了呀。

怎么说呢，这问题也比较好解决啦，就是加锁，保证每次定时器任务执行，只有一个节点运行，那不就行了嘛！！！

加锁，一时间，很多小盆友，就说：这还不简单，看哥们直接加上一个synchronized，不就完事了？

结果一部署上正式环境，还是一样的结果。老大，又给你一顿...

小盆友呀，你要知道这个，还不是一个synchronized关键字能解决的，这个是分布式的问题了。

有3个节点，那就说明有3个jvm环境了。synchronized关键字，只能保证1个jvm环境下是同步的，所以在这个环境下，是不管用的。

解决这个问题，还得用分布式锁出马才行！！！

分布式锁描述

实现一个分布式锁，一般来说，有3种方式啦！！！

• 基于数据库实现分布式锁；
• 基于缓存（Redis等）实现分布式锁
• 基于Zookeeper实现分布式锁

基于数据库，也可以实现分布式锁，当然啦，高并发情况下，数据库分布式锁，很影响效率，一般不会怎么使用这种方式了。

我们一般会使用后面2种方式，实现分布式锁。也是目前比较主流的实现方式了。

• 分布式锁，我们要考虑什么问题呢？

（1）能加锁
（2）锁互斥
（3）可重⼊加锁机制
（4）锁释放机制
 ...

一般来说，redis是没有分布式锁的实现的，需要我们自己用redis实现，锁应该要包含可重⼊加锁机制、锁释放机制等等实现！！！

怎么说呢，就是你要实现一把分布式锁，就得考虑很多的问题啦。可能对于咋们这些加班狗来说，问题有点大了。

那有无已经写好的框架，可以开箱即用的呢？

那就是今天要讲的主题了：Redisson分布式锁

Redisson分布式锁，已经帮我们封装了，一把分布式锁，应该要有的机制，可重⼊加锁机制、锁释放机制等等，都已经有相关的实现。

真的对于我们这些使用人员来说，可以说是开箱即用了呀！！！

确实是牛！！！

分布式锁使用

• pom依赖

<!--redisson依赖-->
<dependency>
    <groupId>org.redisson</groupId>
    <artifactId>redisson</artifactId>
</dependency>

• redis分布式锁配置类

/**
 * redis分布式锁配置类
 */
@Configuration
public class RedissonConfig {

    @Value("${redis.database}")
    private int database;
    @Value("${redis.host}")
    private String host;
    @Value("${redis.port}")
    private String port;
    @Value("${redis.password}")
    private String password;
    @Value("${redis.timeout}")
    private int timeout;

    /**
     * RedissonClient,单机模式
     */
    @Bean
    public RedissonClient redisson() {
        Config config = new Config();
        SingleServerConfig singleServerConfig = config.useSingleServer();
        singleServerConfig.setAddress("redis://" + host + ":" + port);
        singleServerConfig.setTimeout(timeout);
        singleServerConfig.setDatabase(database);
        if (!StringUtils.isEmpty(password)) {
            singleServerConfig.setPassword(password);
        }
        return Redisson.create(config);
    }
}

该配置的作用，是帮我们注入RedissonClient

• redis的配置如下

# Redis配置

redis:
    database: 1 #Redis索引0~15，默认为0
    host: 127.0.0.1
    port: 6379
    password:  #密码（默认为空）
    ssl: true
    timeout: 30000 #连接超时时间（毫秒）

• redis分布式锁工具类

/**
 * redis分布式锁工具类
 */
@Component
public class RedissonUtils {

    @Autowired
    private RedissonClient redissonClient;

    /**
     * 获取锁，如果锁不可用，则当前线程处于休眠状态，直到获得锁为止。
     *
     * @param lockKey
     */
    public void lock(String lockKey) {
        RLock lock = redissonClient.getLock(Constants.REDISSON_KEY + lockKey);
        lock.lock();
    }

    /**
     * 释放锁
     */
    public void unlock(String lockKey) {
        RLock lock = redissonClient.getLock(Constants.REDISSON_KEY + lockKey);
        lock.unlock();
    }

    /**
     * 获取锁,如果锁不可用，则当前线程处于休眠状态，直到获得锁为止。如果获取到锁后，执行结束后解锁或达到超时时间后会自动释放锁
     *
     * @param lockKey
     * @param timeout
     */
    public void lock(String lockKey, int timeout) {
        RLock lock = redissonClient.getLock(Constants.REDISSON_KEY + lockKey);
        lock.lock(timeout, TimeUnit.SECONDS);
    }

    /**
     * 获取锁,如果锁不可用，则当前线程处于休眠状态，直到获得锁为止。如果获取到锁后，执行结束后解锁或达到超时时间后会自动释放锁
     *
     * @param lockKey
     * @param unit
     * @param timeout
     */
    public void lock(String lockKey, TimeUnit unit, int timeout) {
        RLock lock = redissonClient.getLock(Constants.REDISSON_KEY + lockKey);
        lock.lock(timeout, unit);
    }

    /**
     * 尝试获取锁，获取到立即返回true,未获取到立即返回false
     *
     * @param lockKey
     * @return
     */
    public boolean tryLock(String lockKey) {
        RLock lock = redissonClient.getLock(Constants.REDISSON_KEY + lockKey);
        return lock.tryLock();
    }

    /**
     * 尝试获取锁，在等待时间内获取到锁则返回true,否则返回false,如果获取到锁，则要么执行完后程序释放锁，
     * 要么在给定的超时时间leaseTime后释放锁
     *
     * @param lockKey
     * @param waitTime
     * @param leaseTime
     * @param unit
     * @return
     */
    public boolean tryLock(String lockKey, long waitTime, long leaseTime, TimeUnit unit) throws InterruptedException {
        RLock lock = redissonClient.getLock(Constants.REDISSON_KEY + lockKey);
        return lock.tryLock(waitTime, leaseTime, unit);
    }

    /**
     * 检查锁是否被任何线程锁定，如果锁定，则返回true，否则返回false
     *
     * @param lockKey
     * @return
     */
    public boolean isLocked(String lockKey) {
        RLock lock = redissonClient.getLock(Constants.REDISSON_KEY + lockKey);
        return lock.isLocked();
    }


    /**
     * 检查此锁是否由当前线程持有，持有返回true，否则为false
     *
     * @param lockKey
     * @return
     */
    public boolean isHeldByCurrentThread(String lockKey) {
        RLock lock = redissonClient.getLock(Constants.REDISSON_KEY + lockKey);
        return lock.isHeldByCurrentThread();
    }
}

• 定时器任务中使用

private void testLockJobTask() {
    String lockKey = "jobTask";
    try {
        // 尝试获取锁，获取到锁返回true，获取不到返回false
        boolean falg = RedissonUtils.tryLock(lockKey);
        
        //判断是否获取到锁，获取到，即执行相应的逻辑
        if(falg){
            //处理业务数据
            ...
        }
    } catch (Exception e) {
        e.printStackTrace();
    } finally {
        redisLockUtil.unlock(lockKey);
    }
}

好了，Redisson分布式锁的使用，就是这样了！！！^_^

牛牛牛！！！

后面，咋们来谈谈：Redis 分布式锁的缺陷：

redis主从复制环境下，会有这样的问题：

就是如果你对某个 master 实例，客户端1 写⼊了 myLock 这种锁key 的 value，此时会异步复制给对应的 slave 实例。

但是这个过程中⼀旦发⽣ master 宕机，主备切换， slave 变为了 master。

接着就会导致，客户端2 来尝试加锁的时候，在新的 master 上完成了加锁，⽽客户端1之前已成功加锁，那么它也以为⾃⼰成功加了锁。

此时就会导致多个客户端对⼀个分布式锁完成了加锁。

这时系统在业务语义上⼀定会出现问题，导致各种脏数据的产⽣。 所以这个就是 redis cluster，或者是 redis master-slave 架构的主从异步复制导致的

redis 分布式锁的最⼤缺陷：在 master 实例宕机的时候，可能导致多个客户端同时完成加锁

哈哈，既然会有这样的问题，那我们不妨试试分布式锁的第三种方式：Zookeeper

这个以后再说了，今天就先到这里了，溜了溜了溜了！！！^_^