devbox
人工智能uu
这个屌丝很懒，什么也没留下！
关注作者
热门标签
热门文章
当前位置:   article > 正文

【Redis深入】Redis分布式锁的实现_cacheutils.setexnx

作者：人工智能uu | 2024-06-22 03:13:31

赞

踩

cacheutils.setexnx

1. 引入

  • 在实际的应用中，有很多情况需要加锁处理，最典型的情况就是秒杀了，一般的解决方案都是基于DB实现，但如果使用Redis，该如何实现呢。
  • 我们都知道Redis为单线程模式，采用队列模式将并发访问变成串行访问，且多客户端对redis的连接并不存在竞争关系。而且Redis提供一些命令，如SETNX，GETSET，来实现分布式锁机制。
  • 实现分布式锁需要用到的几个命令，请参考：Redis的常用命令
  • 为什么不能使用传统的锁：因为传统的锁（Synchronized 和 lock）是解决在一个JVM下的多线程竞争问题，而分布式是具有多个JVM的。传统的锁跨不了JVM。

2. 分布式锁的实现

2.1 分布式锁的实现方式

  • 基于Redis的分布式锁
  • MySQL数据库乐观锁
  • 基于ZooKeeper的分布式锁
  • 自研分布式锁：如谷歌的 Chubby。

2.2 Redis分布式锁的实现原理

  • Redis分布式锁可基本原理：用一个状态值表示锁，对锁的占用和释放通过状态值来标识。
  • Redis为单进程单线程模式，采用队列模式将并发访问变成串行访问，且多客户端对Redis的连接并不存在竞争关系。redis的SETNX命令可以方便的实现分布式锁。
  • 安全属性（互斥）：不管任何时候，只有一个客户端能持有同一个锁。
  • 效率属性
    • 不会死锁，最终一定会得到锁，就算一个持有锁的客户端宕掉或者发生网络分区。
    • 容错，只要大多数Redis节点正常工作，客户端应该都能获取和释放锁。
  • 多个进程执行以下Redis命令
SETNX lock.foo < current Unix time + lock timeout + 1>

  • 1
  • 如果 SETNX 返回1，说明该进程获得锁，SETNX将键 lock.foo 的值设置为锁的超时时间（当前时间 + 锁的有效时间）。
  • 如果 SETNX 返回0，说明其他进程已经获得了锁，进程不能进入临界区。进程可以在一个循环中不断地尝试 SETNX 操作，以获得锁。
  • Redis分布式锁实现的关键
    • 原子命令加锁：SET key random_value NX PX 30000
    • 设置值random_value是随机的，为了更安全的释放锁。例如：使用uuid
    • 释放锁的时候需要检查 key 是否存在，且 key 对应的值是否和指定的值相等，相等才能释放锁。为了保障原子性，需要用 lua 脚本。

2.3 死锁问题

（1）出现死锁的原因

  • 获取锁的客户端端执行时间过长，进程被kill掉，无法释放锁
  • 或是因为其他异常崩溃（如网络中断），那么其他进程都会处于一直等待的状态，即出现“死锁”。

（2）死锁的解决方法

  • 通过DEL命令删除锁
    通过DEL命令删除锁，然后再SETNX一次，这会出现问题，当多个客户端检测到锁超时后都会尝试去释放它，可能会出现一个竞态条件。
  • 对加锁要做时效性检测（GETSET命令）
    在加锁时，把当前时间戳作为value存入此锁中，通过当前时间戳和Redis中的时间戳进行对比，如果超过一定差值，认为锁已经时效，防止锁无限期的锁下去，
  • 注：为了让分布式锁的算法更稳键，持有锁的客户端在解锁之前应该再检查一次自己的锁是否已经超时，再去做DEL操作，因为可能客户端因为某个耗时的操作而挂起，操作完的时候锁因为超时已经被别人获得，这时就不必解锁了。

2.4 分布式锁可用的条件：

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

3. Jedis代码实现

3.1 利用SETNX和SETEX实现

public class Setexnx {

    private static Logger logger = Logger.getLogger(Setexnx.class.getName());

    //最长时间锁为1小时
    private final static int maxExpireTime = 1 * 60 * 60;

    //系统时间偏移量15秒，服务器间的系统时间差不可以超过15秒,避免由于时间差造成错误的解锁
    private final static int offsetTime = 15;

    //分布式锁的实现
    public static boolean Lock(String key, String value, int waitTime, int expire) {

        long start = System.currentTimeMillis();
        String lock_key = key + "_lock";
        logger.info("开始获取分布式锁 key：" + key + " lock_key:" + lock_key + " value:" + value);

        do {
            try {
                Thread.sleep(1);
                long ret = CacheUtils.Setnx(CacheSpacePrefixEnum.TOOLBAR_SYS.name(), lock_key, System.currentTimeMillis() + "$T$" + value,
                        (expire > maxExpireTime) ? maxExpireTime : expire);
                if (ret == 1) {
                    logger.info("成功获得分布式锁 key：" + key + " value:" + value);
                    return Boolean.TRUE;
                } else { // 存在锁,并对死锁进行修复
                    String desc = CacheUtils.GSetnx(CacheSpacePrefixEnum.TOOLBAR_SYS.name(), lock_key);

                    // 首次锁检测
                    if (desc.indexOf("$T$") > 0) {
                        // 上次锁时间
                        long lastLockTime = NumberUtils.toLong(desc.split("[$T$]")[0]);
                        // 明确死锁,利用Setex复写，再次设定一个合理的解锁时间让系统正常解锁
                        if (System.currentTimeMillis() - lastLockTime > (expire + offsetTime) * 1000) {
                            // 原子操作，只需要一次,会发生小概率事件，多个服务同时发现死锁同时执行此行代码(并发),
                            // 为什么设置解锁时间为expire（而不是更小的时间），防止在解锁发送错乱造成新锁解锁
                            CacheUtils.Setex(CacheSpacePrefixEnum.TOOLBAR_SYS.name(), lock_key, value, expire);
                            logger.warn("发现死锁【" + expire + "秒后解锁】key：" + key + " desc:" + desc);
                        } else {
                            logger.info("当前锁key：" + key + " desc:" + desc);
                        }
                    } else {
                        logger.warn("死锁解锁中key：" + key + " desc:" + desc);
                    }

                }
                if (waitTime == 0) {
                    break;
                }
                Thread.sleep(500);
            }
            catch (Exception ex) {
                logger.error(Trace.GetTraceStackDetails("获取锁失败", ex));
            }
        }
        while ((System.currentTimeMillis() - start) < waitTime * 1000);
        logger.warn("获取分布式锁失败 key：" + key + " value:" + value);
        return Boolean.FALSE;
    }

    //解锁
    public static boolean UnLock(String key) {
        String lock_key = key + "_lock";
        try {
            CacheUtils.Del(CacheSpacePrefixEnum.TOOLBAR_SYS.name(), lock_key);
        }
        catch (Exception ex) {
            logger.error(Trace.GetTraceStackDetails("解锁锁失败key：" + key + " lock_key:" + lock_key, ex));
        }
        return Boolean.FALSE;
    }


    public Long Setnx(String key, String value, int expireTime) throws Exception {
        ShardedJedis jedis = null;

        try {
            jedis = pool.getResource();

            Long ret = jedis.setnx(key, value);
            if (ret == 1 && expireTime > 0) {
                jedis.expire(key, expireTime);
            }
            return ret;
        }
        catch (Exception e) {
            throw e;
        }
        finally {
            if (pool != null && jedis != null) {
                pool.returnResourceObject(jedis);
            }
        }
    }

    public String GSetnx(String key) throws Exception {
        ShardedJedis jedis = null;
        try {
            jedis = pool.getResource();
            return jedis.get(key);
        }
        catch (Exception e) {
            throw e;
        }
        finally {
            if (pool != null && jedis != null) {
                pool.returnResourceObject(jedis);
            }
        }
    }
}


  • 1
  • 2
  • 3
  • 4
  • 5
  • 6
  • 7
  • 8
  • 9
  • 10
  • 11
  • 12
  • 13
  • 14
  • 15
  • 16
  • 17
  • 18
  • 19
  • 20
  • 21
  • 22
  • 23
  • 24
  • 25
  • 26
  • 27
  • 28
  • 29
  • 30
  • 31
  • 32
  • 33
  • 34
  • 35
  • 36
  • 37
  • 38
  • 39
  • 40
  • 41
  • 42
  • 43
  • 44
  • 45
  • 46
  • 47
  • 48
  • 49
  • 50
  • 51
  • 52
  • 53
  • 54
  • 55
  • 56
  • 57
  • 58
  • 59
  • 60
  • 61
  • 62
  • 63
  • 64
  • 65
  • 66
  • 67
  • 68
  • 69
  • 70
  • 71
  • 72
  • 73
  • 74
  • 75
  • 76
  • 77
  • 78
  • 79
  • 80
  • 81
  • 82
  • 83
  • 84
  • 85
  • 86
  • 87
  • 88
  • 89
  • 90
  • 91
  • 92
  • 93
  • 94
  • 95
  • 96
  • 97
  • 98
  • 99
  • 100
  • 101
  • 102
  • 103
  • 104
  • 105
  • 106
  • 107
  • 108
  • 109
  • 110
  • 111
  • 112

存在的问题

  1. 需要设置睡眠时间
  • 为了减少对Redis的压力，获取锁尝试时，循环之间一定要做sleep操作。但是sleep时间是多少是门学问。需要根据自己的Redis的QPS，加上持锁处理时间等进行合理计算。
  • 最好使用随机时间,可以防止饥饿进程的出现,
  • 当同时到达多个进程，只会有一个进程获得锁,其他的进程都用同样的频率进行尝试申请锁,这将可能导致前面来的锁得不到满足.
  • 使用随机的等待时间可以一定程度上保证公平性
  1. setnx和expire是两条Redis命令，不具有原子性，如果程序在执行完setnx()之后突然崩溃，导致锁没有设置过期时间。那么将会发生死锁。之所以这样实现，是因为低版本的jedis并不支持多参数的set()方法。
  2. 锁不具备拥有者标识，解锁时没有预先判断锁的拥有者而直接解锁，会导致任何客户端都可以随时进行解锁，不管这把锁是不是它的。

3.2 只用SETNX命令实现分布式锁

public class RedisLockSetNx {

    private static Logger logger = Logger.getLogger(RedisLockSetNx.class.getName());
    private RedisTemplate redisTemplate;
    private static final int DEFAULT_ACQUIRY_RESOLUTION_MILLIS = 100;
    private String lockKey;
    //锁超时时间，防止线程在入锁以后，无限的执行等待
    private int expireMsecs = 60 * 1000;
    //锁等待时间，防止线程饥饿
    private int timeoutMsecs = 10 * 1000;
    private volatile boolean locked = false;

    //Detailed constructor with default acquire timeout 10000 msecs and lock expiration of 60000 msecs.
    public RedisLock(RedisTemplate redisTemplate, String lockKey) {
        this.redisTemplate = redisTemplate;
        this.lockKey = lockKey + "_lock";
    }

    //Detailed constructor with default lock expiration of 60000 msecs.
    public RedisLock(RedisTemplate redisTemplate, String lockKey, int timeoutMsecs) {
        this(redisTemplate, lockKey);
        this.timeoutMsecs = timeoutMsecs;
    }

    //Detailed constructor.
    public RedisLock(RedisTemplate redisTemplate, String lockKey, int timeoutMsecs, int expireMsecs) {
        this(redisTemplate, lockKey, timeoutMsecs);
        this.expireMsecs = expireMsecs;
    }

    //获取lockkey
    public String getLockKey() {
        return lockKey;
    }

    //获取键的值
    private String get(final String key) {
        Object obj = null;
        try {
            obj = redisTemplate.execute(new RedisCallback<Object>() {
                @Override
                public Object doInRedis(RedisConnection connection) throws DataAccessException {
                    StringRedisSerializer serializer = new StringRedisSerializer();
                    byte[] data = connection.get(serializer.serialize(key));
                    connection.close();
                    if (data == null) {
                        return null;
                    }
                    return serializer.deserialize(data);
                }
            });
        } catch (Exception e) {
            logger.error("get redis error, key : {}", key);
        }
        return obj != null ? obj.toString() : null;
    }

    //设置键的值
    private boolean setNX(final String key, final String value) {
        Object obj = null;
        try {
            obj = redisTemplate.execute(new RedisCallback<Object>() {
                @Override
                public Object doInRedis(RedisConnection connection) throws DataAccessException {
                    StringRedisSerializer serializer = new StringRedisSerializer();
                    Boolean success = connection.setNX(serializer.serialize(key), serializer.serialize(value));
                    connection.close();
                    return success;
                }
            });
        } catch (Exception e) {
            logger.error("setNX redis error, key : {}", key);
        }
        return obj != null ? (Boolean) obj : false;
    }

    //获取并设置键的值，并返回旧值
    private String getSet(final String key, final String value) {
        Object obj = null;
        try {
            obj = redisTemplate.execute(new RedisCallback<Object>() {
                @Override
                public Object doInRedis(RedisConnection connection) throws DataAccessException {
                    StringRedisSerializer serializer = new StringRedisSerializer();
                    byte[] ret = connection.getSet(serializer.serialize(key), serializer.serialize(value));
                    connection.close();
                    return serializer.deserialize(ret);
                }
            });
        } catch (Exception e) {
            logger.error("setNX redis error, key : {}", key);
        }
        return obj != null ? (String) obj : null;
    }

    //实现分布式锁
    public synchronized boolean lock() throws InterruptedException {
        int timeout = timeoutMsecs;
        while (timeout >= 0) {
            long expires = System.currentTimeMillis() + expireMsecs + 1;
            String expiresStr = String.valueOf(expires); //锁到期时间
            if (this.setNX(lockKey, expiresStr)) {
                // 获取锁
                locked = true;
                return true;
            }
            //redis系统的时间
            String currentValueStr = this.get(lockKey);
            //判断是否为空，不为空的情况下，如果被其他线程设置了值，则第二个条件判断是过不去的
            if (currentValueStr != null && Long.parseLong(currentValueStr) < System.currentTimeMillis()) {

                //获取上一个锁到期时间，并设置现在的锁到期时间，
                //只有一个线程才能获取上一个线上的设置时间，因为jedis.getSet是同步的
                String oldValueStr = this.getSet(lockKey, expiresStr);

                //防止误删（覆盖，因为key是相同的）了他人的锁——这里达不到效果，这里值会被覆盖，但是因为什么相差了很少的时间，所以可以接受
                if (oldValueStr != null && oldValueStr.equals(currentValueStr)) {
                    //分布式的情况下:如过这个时候，多个线程恰好都到了这里，但是只有一个线程的设置值和当前值相同，他才有权利获取锁
                    // 获取锁
                    locked = true;
                    return true;
                }
            }
            timeout -= DEFAULT_ACQUIRY_RESOLUTION_MILLIS;
            //延迟100 毫秒,
            Thread.sleep(DEFAULT_ACQUIRY_RESOLUTION_MILLIS);
        }
        return false;
    }

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

  • 1
  • 2
  • 3
  • 4
  • 5
  • 6
  • 7
  • 8
  • 9
  • 10
  • 11
  • 12
  • 13
  • 14
  • 15
  • 16
  • 17
  • 18
  • 19
  • 20
  • 21
  • 22
  • 23
  • 24
  • 25
  • 26
  • 27
  • 28
  • 29
  • 30
  • 31
  • 32
  • 33
  • 34
  • 35
  • 36
  • 37
  • 38
  • 39
  • 40
  • 41
  • 42
  • 43
  • 44
  • 45
  • 46
  • 47
  • 48
  • 49
  • 50
  • 51
  • 52
  • 53
  • 54
  • 55
  • 56
  • 57
  • 58
  • 59
  • 60
  • 61
  • 62
  • 63
  • 64
  • 65
  • 66
  • 67
  • 68
  • 69
  • 70
  • 71
  • 72
  • 73
  • 74
  • 75
  • 76
  • 77
  • 78
  • 79
  • 80
  • 81
  • 82
  • 83
  • 84
  • 85
  • 86
  • 87
  • 88
  • 89
  • 90
  • 91
  • 92
  • 93
  • 94
  • 95
  • 96
  • 97
  • 98
  • 99
  • 100
  • 101
  • 102
  • 103
  • 104
  • 105
  • 106
  • 107
  • 108
  • 109
  • 110
  • 111
  • 112
  • 113
  • 114
  • 115
  • 116
  • 117
  • 118
  • 119
  • 120
  • 121
  • 122
  • 123
  • 124
  • 125
  • 126
  • 127
  • 128
  • 129
  • 130
  • 131
  • 132
  • 133
  • 134
  • 135
  • 136
  • 137
  • 138
  • 139

存在的问题

  1. 需要保证时间同步
    由于是客户端自己生成过期时间，所以需要强制要求分布式下每个客户端的时间必须同步。
  2. 过期时间被覆盖
    当锁过期的时候，如果多个客户端同时执行jedis.getSet()方法，那么虽然最终只有一个客户端可以加锁，但是这个客户端的锁的过期时间可能被其他客户端覆盖。
  3. 解锁时误删锁
    若调用jedis.del()方法的时候，这把锁已经不属于当前客户端的时候会解除他人加的锁。例如客户端A加锁，一段时间之后客户端A解锁，在执行jedis.del()之前，锁突然过期了，此时客户端B尝试加锁成功，然后客户端A再执行del()方法，则将客户端B的锁给解除了。
    锁误解除

3.3 使用多参数的set方法实现单机的分布式锁

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";
    private static final Long RELEASE_SUCCESS = 1L;

    /**
     * 尝试获取分布式锁
     * @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);
        return LOCK_SUCCESS.equals(result);
    }

    /**
     * 释放分布式锁
     * @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));
        return RELEASE_SUCCESS.equals(result);
    }

}

  • 1
  • 2
  • 3
  • 4
  • 5
  • 6
  • 7
  • 8
  • 9
  • 10
  • 11
  • 12
  • 13
  • 14
  • 15
  • 16
  • 17
  • 18
  • 19
  • 20
  • 21
  • 22
  • 23
  • 24
  • 25
  • 26
  • 27
  • 28
  • 29
  • 30
  • 31
  • 32
  • 33
  • 34

分析：

  1. 加锁set()方法的五个形参解释
  • key：使用key来当锁，因为key是唯一的。
  • value：传入requestId，因为分布式锁要满足第四个条件解铃还须系铃人，通过给value赋值为requestId，就能知道这把锁是哪个请求加的，在解锁的时候就可以有依据。requestId可以使用UUID.randomUUID().toString()方法生成。
  • nxxx：传入NX，即SET IF NOT EXIST，即当key不存在时，我们进行set操作；若key已经存在，则不做任何操作；
  • expx：传入PX，给key加一个过期的设置，具体时间由参数time决定。
  • time：与expx参数对应，代表key的过期时间。
  1. 解锁使用lua脚本
  • 使用lua脚本获取锁对应的value值，检查是否与requestId相等，如果相等则删除锁（解锁）。
  • 解锁lua脚本以确保该操作是原子性的，在eval命令执行Lua代码的时候，Lua代码将被当成一个命令去执行，并且直到eval命令执行完成，Redis才会执行其他命令。

存在的问题：

  • 单机部署的情况，宕机后不可用。使用Redlock解决。
  • 客户端持有锁超时，使用Redssion 解决。
    超时解锁导致并发

4. Redis分布式锁的实现

4.1 Redlock

  • 在 Redis 的分布式环境中，我们假设有 N 个 Redis Master。这些节点完全互相独立，不存在主从复制或者其他集群协调机制。
  • 假设有 5 个完全独立的 Redis Master 节点，他们分别运行在 5 台服务器中，可以保证他们不会同时宕机。客户端获取锁的5个步骤（ 来自redis官网）
  1. 获取当前 Unix 时间，以毫秒为单位。
  2. 依次尝试从 N 个实例，使用相同的 key 和随机值获取锁。
  3. 客户端使用当前时间减去开始获取锁时间（步骤 1 记录的时间）就得到获取锁使用的时间。当且仅当从大多数（这里是 3 个节点）的 Redis 节点都取到锁，并且使用的时间小于锁失效时间时，锁才算获取成功。
  4. 如果取到了锁，key 的真正有效时间等于有效时间减去获取锁所使用的时间（步骤 3 计算的结果）。
  5. 如果因为某些原因，获取锁失败（没有在至少 N/2+1 个Redis实例取到锁或者取锁时间已经超过了有效时间），客户端应该在所有的 Redis 实例上进行解锁（即便某些 Redis 实例根本就没有加锁成功）。
  • 只要大多数的节点可以正常工作，就可以保证 Redlock 的正常工作。就解决了单点 Redis 宕机，由于集群异步通信，导致锁失效的问题。
  • Redlock存在的问题：
    • 对时钟依赖性太强， 若Ｎ个节点中的某个节点发生 时间跳跃 ，也可能会引此而引发锁安全性问题。
    • 某个节点故障重启后可能一把锁被多个客户端持有。延迟重启解决方法会导致系统在TTL时间内任何锁都将无法加锁成功。所以不能用

4.2 Redisson

  • 使用 Redssion 做分布式锁，不需要明确指定 value ，框架会生成一个由 UUID 和 加锁操作的线程的 threadId 用冒号拼接起来的字符串。
  • 锁名称用的是hash类型，而不是string。因为加锁使用的hincrby 命令，支持可重入锁
  • 源码（版本3.15.0）剖析：
    • 加锁的过期时间默认是 30s。当一个 key 加锁成功或者当一个锁重入成功后都会返回空，只有加锁失败的情况下会返回当前锁剩余的时间。
    private <T> RFuture<Long> tryAcquireAsync(long waitTime, long leaseTime, TimeUnit unit, long threadId) {
        if (leaseTime != -1) {
            return tryLockInnerAsync(waitTime, leaseTime, unit, threadId, RedisCommands.EVAL_LONG);
        }
        RFuture<Long> ttlRemainingFuture = tryLockInnerAsync(waitTime, internalLockLeaseTime,
                                                                TimeUnit.MILLISECONDS, threadId, RedisCommands.EVAL_LONG);
        ttlRemainingFuture.onComplete((ttlRemaining, e) -> {
            if (e != null) {
                return;
            }

            // lock acquired
            if (ttlRemaining == null) {
                scheduleExpirationRenewal(threadId);
            }
        });
        return ttlRemainingFuture;
    }

	// 加锁成功进入该方法，即将进入看门狗逻辑
    protected void scheduleExpirationRenewal(long threadId) {
        ExpirationEntry entry = new ExpirationEntry();
        ExpirationEntry oldEntry = EXPIRATION_RENEWAL_MAP.putIfAbsent(getEntryName(), entry);
        if (oldEntry != null) {
            oldEntry.addThreadId(threadId);
        } else {
            entry.addThreadId(threadId);
            renewExpiration();
        }
    }

  • 1
  • 2
  • 3
  • 4
  • 5
  • 6
  • 7
  • 8
  • 9
  • 10
  • 11
  • 12
  • 13
  • 14
  • 15
  • 16
  • 17
  • 18
  • 19
  • 20
  • 21
  • 22
  • 23
  • 24
  • 25
  • 26
  • 27
  • 28
  • 29
  • 30
  • 当加锁成功或者重入成功后进入看门狗逻辑：定时任务每 internalLockLeaseTime/3ms 后执行一次。而 internalLockLeaseTime 默认为 30000。所以该任务每 10s 执行一次。定时任务（Timeout）基于 netty 的时间轮HashedWheelTimer
    private void renewExpiration() {
        ExpirationEntry ee = EXPIRATION_RENEWAL_MAP.get(getEntryName());
        if (ee == null) {
            return;
        }
        
        Timeout task = commandExecutor.getConnectionManager().newTimeout(new TimerTask() {
            @Override
            public void run(Timeout timeout) throws Exception {
                ExpirationEntry ent = EXPIRATION_RENEWAL_MAP.get(getEntryName());
                if (ent == null) {
                    return;
                }
                Long threadId = ent.getFirstThreadId();
                if (threadId == null) {
                    return;
                }
                
                RFuture<Boolean> future = renewExpirationAsync(threadId);
                future.onComplete((res, e) -> {
                    if (e != null) {
                        log.error("Can't update lock " + getName() + " expiration", e);
                        EXPIRATION_RENEWAL_MAP.remove(getEntryName());
                        return;
                    }
                    
                    if (res) {
                        // reschedule itself
                        renewExpiration();
                    }
                });
            }
        }, internalLockLeaseTime / 3, TimeUnit.MILLISECONDS);
        
        ee.setTimeout(task);
    }

  • 1
  • 2
  • 3
  • 4
  • 5
  • 6
  • 7
  • 8
  • 9
  • 10
  • 11
  • 12
  • 13
  • 14
  • 15
  • 16
  • 17
  • 18
  • 19
  • 20
  • 21
  • 22
  • 23
  • 24
  • 25
  • 26
  • 27
  • 28
  • 29
  • 30
  • 31
  • 32
  • 33
  • 34
  • 35
  • 36
  • 释放锁： 有个counter 的判断，如果减一后小于等于 0。就执行 del key 的操作，之后publish发布事件（tryAcquire 加锁失败订阅的事件）
  • 解锁完成后通过cancelExpirationRenewal方法取消看门狗定时任务
    public RFuture<Void> unlockAsync(long threadId) {
        RPromise<Void> result = new RedissonPromise<>();
        RFuture<Boolean> future = unlockInnerAsync(threadId);

        future.onComplete((opStatus, e) -> {
        	// 取消看门狗定时任务（响应式编程）
            cancelExpirationRenewal(threadId);

            if (e != null) {
                result.tryFailure(e);
                return;
            }

            if (opStatus == null) {
                IllegalMonitorStateException cause = new IllegalMonitorStateException("attempt to unlock lock, not locked by current thread by node id: "
                        + id + " thread-id: " + threadId);
                result.tryFailure(cause);
                return;
            }

            result.trySuccess(null);
        });

        return result;
    }

  • 1
  • 2
  • 3
  • 4
  • 5
  • 6
  • 7
  • 8
  • 9
  • 10
  • 11
  • 12
  • 13
  • 14
  • 15
  • 16
  • 17
  • 18
  • 19
  • 20
  • 21
  • 22
  • 23
  • 24
  • 25

订阅消息等待锁释放

  • 通过方法可以设置过期时间，就不用启动看门狗了
  • Redisson加锁解锁流程图
    Redisson加锁解锁流程图

4.3 分布式锁存在的问题

4.3.1 主备切换
  • 为了保证 Redis 的可用性，一般采用主从方式部署。主从数据同步有异步和同步两种方式，Redis 将指令记录在本地内存 buffer 中，然后异步将 buffer 中的指令同步到从节点，从节点一边执行同步的指令流来达到和主节点一致的状态，一边向主节点反馈同步情况。
  • 在包含主从模式的集群部署方式中，当主节点挂掉时，从节点会取而代之，但客户端无明显感知。当客户端 A 成功加锁，但是数据还没被同步到 Salve，此时主节点挂掉，从节点提升为主节点，新的主节点没有锁的数据，当客户端 B 加锁时就会成功，出现一把锁被拿到了两次的场景。
    主备切换的问题
4.3.2 集群脑裂
  • 集群脑裂指因为网络问题，导致 Redis master 节点跟 slave 节点和 sentinel 集群处于不同的网络分区，因为 sentinel 集群无法感知到 master 的存在，所以将 slave 节点提升为 master 节点，此时存在两个不同的 master 节点。Redis Cluster 集群部署方式同理。
  • 当不同的客户端连接不同的 master 节点时，两个客户端可以同时拥有同一把锁。如下：
    集群脑裂
4.3.3 小结
  • 分布式锁的一致性要求 CP，但是Redis 集群架构之间的异步通信满足的是 AP ，因为大多场景中能容忍，BASE保持最终一致性行就可以。

参考链接：
1.分布式系统互斥性与幂等性问题的分析与解决-美团技术博客
2. 分布式锁的几种实现方式
3. Redis锁从面试连环炮聊到神仙打架
4. Redisson看门狗+时间轮
5. 分布式锁的实现之 redis 篇

声明：本文内容由网友自发贡献，版权归原作者所有，本站不承担相应法律责任。如您发现有侵权的内容，请联系我们。转载请注明出处：【wpsshop博客】
推荐阅读
相关标签

Copyright © 2003-2013 www.wpsshop.cn 版权所有，并保留所有权利。

  

闽ICP备14008679号