Redis分布式锁以及一种简单的锁时间自行延长的实现讨论

秒速五厘米 2023-07-03 07:09 3阅读 0赞

在分布式环境中普通的java的锁是无法很好的控制的并发的在分布式环境中分布式锁有主要有两种实现方式，一种是利用Zookeeper的可以作为注册中心的特性进行实现，另外一种就是利用Redis实现的分布式锁，两者各有千秋利弊，因为redis和zookeeper的特性是不一样的，有一点比较明显的如zookeeper保证的CAP原则中CP（一致性和分区容错性），而Redis保证的是AP（可用性和分区容错性），就但从这个原则上讲两者就是不一样的需要保证什么还是要根据业务需求。因为Redis在各种程序中基本是必不可少的一种缓存中间件，所以今天就介绍下利用Redis实现的分布式锁。在市面上也有不少实现了redis分布式锁的第三方的jar包（如Redisson），但是相对于简单的应用而言没有必要引入jar包。  
  在各种资料以及博客中可能实现了各种版本的redis锁的实现，但是很少提及的一点是如何让锁根据业务的执行时间自行延长（Redisson是实现了此功能的），本文介绍的方式没有实际应用到生产环境，可能存在很多的漏洞，当然也希望有大神能够进行指点。实现该功能的大体思路就是创建了一个守护线程进行业务的感知，在时间到达指定时间的三分之二的时候进行自行的延长，当业务执行退出以后守护线程感知到不再做任何操作，相对而言这种方式的实现我认为比较简单但是比较耗费性能，因为每次使用锁的时候就要开启守护线程，造成了资源的没必要的浪费。本文介绍的Redis实现的分布式锁主要是采用了Redis的setnx这个原子操作的api以及使用了Lua脚本在删除锁的时候保证实现的原子性。还有一点实现细节就是在集群模式下是不支持Lua脚本的（一个大坑），所以在使用是需要获取原生的Jedis对象。  
 接下来就是代码的实现过程：

import org.apache.commons.logging.Log;
    import org.apache.commons.logging.LogFactory;
    import org.springframework.dao.DataAccessException;
    import org.springframework.data.redis.connection.RedisConnection;
    import org.springframework.data.redis.core.RedisCallback;
    import org.springframework.data.redis.core.RedisTemplate;
    import org.springframework.util.StringUtils;
    import redis.clients.jedis.Jedis;
    import redis.clients.jedis.JedisCluster;
    import redis.clients.jedis.JedisCommands;
    
    import java.util.ArrayList;
    import java.util.List;
    import java.util.UUID;
    import java.util.concurrent.TimeUnit;
    
    
    public class RedisDistributedLock {
    
    private static Log log = LogFactory.getLog(RedisDistributedLock.class);
    /**
     * 锁对象
     */
    private  static  final Object object=new Object();
    
    /**
     * (ms)
     */
    public static final long EXPIRE = 20 * 1000;
    
    /**
     * 默认请求锁的超时时间(ms)
     */
    private static final long TIME_OUT = 10 * 1000;
    
    /**
     * 存储到redis中的锁标志
     */
    private ThreadLocal<String> lockFlag = new ThreadLocal<>();
    
    /**
     * redis管理模板
     */
    private RedisTemplate<Object, Object> redisTemplate;
    
    /**
     * 获取锁失败最大重试次数
     */
    private int retryTimes = 3;
    
    public static final String UNLOCK_LUA;
    
    private InheritableThreadLocal<Boolean> checkThredFlag = new InheritableThreadLocal<>();
    
    ThreadLocal<CheckLock> threadLocal = new ThreadLocal<>();
    
    private volatile boolean loopFlag = false;
    
    static {
        StringBuilder sb = new StringBuilder();
        sb.append("if redis.call(\"get\",KEYS[1]) == ARGV[1] ");
        sb.append("then ");
        sb.append("    return redis.call(\"del\",KEYS[1]) ");
        sb.append("else ");
        sb.append("    return 0 ");
        sb.append("end ");
        UNLOCK_LUA = sb.toString();
    }
    
    public RedisDistributedLock() {
        //此处根据自己的业务注入自己的redis模板
        redisTemplate = new RedisTemplate<>();
    }
    
    
    /**
     * 方法描述: 重载方法（锁标志对应的key）
     *
     * @param key
     * @return boolean
     */
    public boolean lock(String key) {
    
        return lock(key, EXPIRE, TIME_OUT);
    }
    
    /**
     * 方法描述: 重载方法（锁标志对应的key，过期时间）
     *
     * @param key
     * @param expire(ms)
     * @return boolean
     */
    public boolean lock(String key, long expire) {
    
        return lock(key, expire, TIME_OUT);
    
    }
    
    /**
     * 方法描述: 加锁（锁标志对应的key，过期时间，超时时间）
     *
     * @param key
     * @param expire(ms)
     * @param timeOut(ms)
     */
    public boolean lock(String key, long expire, long timeOut) {
    
        boolean result;
        // 系统当前时间，单位：纳秒
        long nowTime = System.nanoTime();
        timeOut = timeOut * 1000 * 1000;
    
        result = setRedis(key, expire);
        if (log.isDebugEnabled()) {
            log.debug("RedisDistributedLock---lock---result1 - " + result + "; key - " + key + " - value - " + getRedis(key));
        }
    
        // 获取锁失败自旋，超时或超过重试次数放弃请求锁
        while ((!result) && (System.nanoTime() - nowTime) < timeOut && retryTimes-- > 0) {
            result = setRedis(key, expire);
            if (log.isDebugEnabled()) {
                log.debug("RedisDistributedLock---lock---result2 - " + result + "; key - " + key + " - value - " + getRedis(key));
            }
        }
    
        return result;
    }
    
    /**
     * 设置锁-原子操作
     */
    private boolean setRedis(final String key, final long expire) {
    
        try {
    
            String result = redisTemplate.execute(new RedisCallback<String>() {
                @Override
                public String doInRedis(RedisConnection connection) throws DataAccessException {
                    JedisCommands commands = (JedisCommands) connection.getNativeConnection();
                    String uuid = UUID.randomUUID().toString();
                    lockFlag.set(uuid);
                    loopFlag = true;
                    return commands.set(key, uuid, "NX", "PX", expire);
                }
            });
            try {
                //开启守护线程
                if (!StringUtils.isEmpty(result)) {
    
                    CheckLock checkLock = new CheckLock(key, expire);
                    Thread thread = new Thread(checkLock);
                    //将线程放入threadLocal
                    threadLocal.set(checkLock);
                    thread.start();
                    return true;
    
                }
    
            } catch (Exception e) {
    
                //异常解锁返回失败不必解锁不要影响上锁
                return true;
    
            }
            return false;
    
        } catch (Exception e) {
            log.error("RedisDistributedLock---setRedis---set redis occured an exception", e);
            checkThredFlag.set(false);
        }
        return false;
    }
    
    private String getRedis(final String key) {
        String result = null;
        try {
            result = redisTemplate.execute(new RedisCallback<String>() {
                @Override
                public String doInRedis(RedisConnection connection) throws DataAccessException {
                    JedisCommands commands = (JedisCommands) connection.getNativeConnection();
                    return commands.get(key);
                }
            });
        } catch (Exception e) {
            log.error("RedisDistributedLock---getRedis---get redis occured an exception", e);
        }
        return result;
    }
    
    
    /**
     * 方法描述: 释放锁
     * 释放锁的时候，有可能因为持锁之后方法执行时间大于锁的有效期，此时有可能已经被另外一个线程持有锁，所以不能直接删除
     */
    public boolean releaseLock(String key) {
    
        try {
    
            final List<String> keys = new ArrayList<>();
            keys.add(key);
            final List<String> args = new ArrayList<>();
            args.add(lockFlag.get());
            //防止在同一个应用的逻辑不会同时去将值设置为true
            synchronized (this) {
                // 使用lua脚本删除redis中匹配value的key，可以避免由于方法执行时间过长而redis锁自动过期失效的时候误删其他线程的锁
                // spring自带的执行脚本方法中，集群模式直接抛出不支持执行脚本的异常，所以只能拿到原redis的connection来执行脚本
                Long result = redisTemplate.execute(new RedisCallback<Long>() {
                    @Override
                    public Long doInRedis(RedisConnection connection) throws DataAccessException {
                        Object nativeConnection = connection.getNativeConnection();
                        // 集群模式和单机模式虽然执行脚本的方法一样，但是没有共同的接口，所以只能分开执行
                        // 集群模式
                        if (nativeConnection instanceof JedisCluster) {
                            return (Long) ((JedisCluster) nativeConnection).eval(UNLOCK_LUA, keys, args);
                            // 单机模式
                        } else if (nativeConnection instanceof Jedis) {
                            return (Long) ((Jedis) nativeConnection).eval(UNLOCK_LUA, keys, args);
                        }
                        return 0L;
                    }
                });
    
                if (log.isDebugEnabled()) {
                    log.debug("RedisDistributedLock---releaseLock---result - " + result + "; key - " + keys + " - value - " + args);
                }
    
                if (result != null && result > 0) {
                    loopFlag = false;
                    return true;
                }
            }
            return false;
    
        } catch (Exception e) {
            log.error("RedisDistributedLock---releaseLock---release lock occured an exception", e);
        }
        return false;
    }
    
    /**
     * 开启守护子线程：进行自旋判断如果时间超过了超时时间的三分之二就会延长
     */
    class CheckLock implements Runnable {
    
        private String checkKey;
        private long exTime;
    
       private CheckLock(String key, long exprie) {
            checkKey = key;
            exTime = exprie;
        }
    
        @Override
        public void run() {
    
            long begin = System.currentTimeMillis();
    
            System.out.println(loopFlag);
            
            synchronized(object) {
    
                while (loopFlag) {
    
                    //自旋了过期时间的三分之二自动延长时间
                    if ((System.currentTimeMillis() - begin >= exTime * 2 / 3)) {
    
                        redisTemplate.expire(checkKey, exTime, TimeUnit.MILLISECONDS);
                        begin = System.currentTimeMillis();
                        try {
                            Thread.sleep(3000);
                        } catch (InterruptedException e) {
                            e.printStackTrace();
                        }
    
                    }
    
                }
            }
    
        }
    }
    }