分布式锁的原理以及实现方式

超、凢脫俗 2021-10-24 00:01 350阅读 0赞

## **一、为什么要使用分布式锁** ##

由于业务的发展以及复杂度，需要用到集群，一个业务被部署都多态服务器上，然后做负载均衡。为了保证一个方法或属性在高并发情况下的同一时间只能被同一个线程执行，在传统单体应用单机部署的情况下，可以使用Java并发处理相关的API(如ReentrantLock或Synchronized)进行互斥控制。在单机环境中，Java中提供了很多并发处理相关的API。但是，随着业务发展的需要，原单体单机部署的系统被演化成分布式集群系统后，由于分布式系统多线程、多进程并且分布在不同机器上，这将使原单机部署情况下的并发控制锁策略失效，单纯的Java API并不能提供分布式锁的能力。为了解决这个问题就需要一种跨JVM的互斥机制来控制共享资源的访问，这就是分布式锁要解决的问题！

## **二、分布式锁应该具备哪些条件** ##

1、在分布式系统环境下，一个方法在同一时间只能被一个机器的一个线程执行； 
    2、高可用的获取锁与释放锁； 
    3、高性能的获取锁与释放锁； 
    4、具备可重入特性； 
    5、具备锁失效机制，防止死锁； 
    6、具备非阻塞锁特性，即没有获取到锁将直接返回获取锁失败。

## **三、分布式锁的三种实现方式** ##

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

## **四、基于数据库的实现方式** ##

基于数据库的实现方式的核心思想是：在数据库中创建一个表，表中包含**方法名**等字段，并在**方法名字段上创建唯一索引**，想要执行某个方法，就使用这个方法名向表中插入数据，成功插入则获取锁，执行完成后删除对应的行数据释放锁。

## **五、基于Redis的实现方式** ##

**  ****[setnx+lua 脚本实现分布式锁][setnx_lua]**

**1、选用Redis实现分布式锁原因：**

（1）Redis有很高的性能；   
（2）Redis命令对此支持较好，实现起来比较方便

**2、使用命令介绍：**

（1）SETNX：SETNX key val：当且仅当key不存在时，set一个key为val的字符串，返回1；若key存在，则什么都不做，返回0。

（2）expire：expire key timeout：为key设置一个超时时间，单位为second，超过这个时间锁会自动释放，避免死锁。

（3）delete：delete key：删除key

在使用Redis实现分布式锁的时候，主要就会使用到这三个命令。

**3、实现思想：**

（1）获取锁的时候，使用setnx加锁，并使用expire命令为锁添加一个超时时间，超过该时间则自动释放锁，锁的value值为一个随机生成的UUID，通过此在释放锁的时候进行判断。

（2）获取锁的时候还设置一个获取的超时时间，若超过这个时间则放弃获取锁。

（3）释放锁的时候，通过UUID判断是不是该锁，若是该锁，则执行delete进行锁释放。

**4、 分布式锁的简单实现代码：**

package com.fs.fscloudadmin.redis;
    
    import redis.clients.jedis.Jedis;
    import redis.clients.jedis.JedisPool;
    import redis.clients.jedis.Transaction;
    import redis.clients.jedis.exceptions.JedisException;
    
    import java.util.List;
    import java.util.UUID;
    
    /**
     * 分布式锁 redis 实现的简单代码
     * @author huyunqiang
     * @Date 2019/8/14 21:10
     */
    public class RedisLock {
        private final JedisPool jedisPool;
    
        //初始化
        public RedisLock(JedisPool jedisPool) {
            this.jedisPool = jedisPool;
        }
    
        /**
         *
         * @param lockName   锁 key
         * @param acquireTimeout  获取超时时间
         * @param timeout         锁的超时时间
         * @return                锁标识
         */
        public String lockWithTimeOut(String lockName,long acquireTimeout,long timeout){
           Jedis conn = null;
           String retIdentifier = null;
           try {
               //获取连接
               conn = jedisPool.getResource();
               //随机生成一个value
               String identifier = UUID.randomUUID().toString();
               //锁名 key 值
               String lockKey = "lock:"+lockName;
               //设置超时时间，上锁后超时时间自动释放锁
               int lockExpire = (int)(timeout/1000);
               //获取缩的超时时间，超过这个时间放弃获取锁
               long end = System.currentTimeMillis()+acquireTimeout;
               while (System.currentTimeMillis()<end){
                   // key 存在什么都不做返回0
                   if(conn.setnx(lockKey,identifier) == 1){
                       //设置过期时间
                       conn.expire(lockKey,lockExpire);
                       //返回value用于释放锁时间确认
                       retIdentifier = identifier;
                       return  retIdentifier;
                   }
    
                   //返回-1 代表么有设置超时间，为key 设置一个超时时间
                   if(conn.ttl(lockKey) == -1){
                        conn.expire(lockKey,lockExpire);
                   }
                   try {
                       Thread.sleep(10);
                   }catch (InterruptedException e){
                        Thread.currentThread().interrupt();
                   }
    
               }
    
    
           }catch (JedisException e){
               e.printStackTrace();
           }finally {
               if(conn != null){
    //               conn.close();
               }
           }
           return retIdentifier;
        }
    
        /**
         * 释放锁
         * @param lockName
         * @param identifier
         * @return
         */
        public boolean releaseLock(String lockName,String identifier){
            Jedis conn = null;
            String lockKey = "lock:"+lockName;
            boolean retFlag = false;
            try {
                 conn = jedisPool.getResource();
                 while (true){
                     //监视lock, 准备开始事务
                     conn.watch(lockKey);
                     //通过前面返回的value判断是不是该锁，是则删除，释放锁
                     if(identifier.equals(conn.get(lockKey))){
                         //开启redis 事务删除锁
                         Transaction transaction = conn.multi();
                         transaction.del(lockKey);
                         List<Object> result = transaction.exec();
                         if(result == null){
                             continue;
                         }
    
                         retFlag = true;
                     }
                     conn.unwatch();
                     break;
                 }
            }catch (JedisException e){
                e.printStackTrace();
            }finally {
                if(conn != null){
    //                conn.close();
                }
            }
    
            return retFlag;
        }
    
    }

**模拟线程进行秒杀服务**

package com.fs.fscloudadmin.redis;
    /**
     * @author huyunqiang
     * @Date 2019/8/14 22:11
     */
    public class RedisLockThread extends Thread{
    
        private RedisService redisService;
    
        public RedisLockThread(RedisService redisService){
            this.redisService = redisService;
        }
    
        @Override
        public void run() {
            // 获取锁和释放锁
            redisService.seckill();
        }
    
    
        public static void main(String[] args) {
            RedisService redisService = new RedisService();
            for(int i = 0;i<50;i++){
                RedisLockThread thread = new RedisLockThread(redisService);
                thread.start();
            }
        }
    
    }

public void seckill(){
           //返回value, 供释放锁时候进行判断
           String identifier = lock.lockWithTimeOut("resourse",5000,1000);
           System.out.println(Thread.currentThread().getName()+"获得了锁");
           System.out.println(--n);
           //如果注释掉 释放锁一部分是异步进行的  加载比较慢
           lock.releaseLock("resourse",identifier);
       }

可以使用自定义注解的方式将以上代码封装在需要使用分布式锁的方法上加入该注解

@Documented
    @Inherited
    @Retention(RetentionPolicy.RUNTIME)
    @Target({ElementType.TYPE, ElementType.METHOD})
    public @interface RedisLock {
    
        String value() default "redis-lock";
    }

@Aspect
    @Component
    public class  RedisLockHandler {
    
        private static final Logger LOGGER = LoggerFactory.getLogger(RedisLockHandler.class);
    
        @Pointcut("@annotation(com.snowalker.annotation.RedisLock)")
        public void redisLock() {}
    
        @Around("@annotation(redisLock)")
        public void around(ProceedingJoinPoint joinPoint, RedisLock redisLock) {
            LOGGER.info("[开始]执行RedisLock环绕通知,获取Redis分布式锁开始");
            String lockName = redisLock.value();
            RedisDistributedLock redisDistributedLock = RedisDistributedLock.getInstance();
            if (redisDistributedLock.lock(lockName)) {
                try {
                    LOGGER.info("获取Redis分布式锁[成功]");
                    joinPoint.proceed();
                } catch (Throwable throwable) {
                    throwable.printStackTrace();
                }
                LOGGER.info("释放Redis分布式锁[成功]");
                redisDistributedLock.release(lockName);
            } else {
                LOGGER.error("获取Redis分布式锁[失败]");
            }
            LOGGER.error("[结束]执行RedisLock环绕通知");
        }
    }

/**
     * @author huyunqiang
     * @date 2018-7-9
     * @desc redis分布式锁核心实现
     */
    public class RedisDistributedLock implements DistributedLock {
    
        /**默认锁超时时间为10S*/
        private static final int EXPIRE_SECONDS = 50;
        private static final Logger log = LoggerFactory.getLogger(RedisDistributedLock.class);
    
        public RedisDistributedLock() {
        }
    
        private volatile static RedisDistributedLock redisDistributedLock;
    
        public static RedisDistributedLock getInstance() {
            if (redisDistributedLock == null) {
                synchronized (RedisDistributedLock.class) {
                    redisDistributedLock = new RedisDistributedLock();
                }
            }
            return redisDistributedLock;
        }
    
        /**
         * 加锁
         *
         * @param lockName
         * @return 返回true表示加锁成功，执行业务逻辑，执行完毕需要主动释放锁，否则就需要等待锁超时重新争抢
         * 返回false标识加锁失败，阻塞并继续尝试获取锁
         */
        @Override
        public boolean lock(String lockName) {
            /**1.使用setNx开始加锁*/
            log.info("开始获取Redis分布式锁流程,lockName={},CurrentThreadName={}", lockName, Thread.currentThread().getName());
            long lockTimeout = Long.parseLong(PropertiesUtil.getProperty("redis.lock.timeout", "5"));
            /**redis中锁的值为:当前时间+超时时间*/
            Long lockResult = RedisPoolUtil.setnx(lockName, String.valueOf(System.currentTimeMillis() + lockTimeout));
             // key 存在什么都不做返回0  等于1 获取到锁 设置超时时间
            if (lockResult != null && lockResult.intValue() == 1) {
                log.info("setNx获取分布式锁[成功],threadName={}", Thread.currentThread().getName());
                RedisPoolUtil.expire(lockName, EXPIRE_SECONDS);
                return true;
            } else {
                log.info("setNx获取分布式锁[失败],threadName={}", Thread.currentThread().getName());
    //            return tryLock(lockName, lockTimeout);
                return false;
            }
        }
    
        private boolean tryLock(String lockName, long lockTimeout) {
            /**
             * 2.加锁失败后再次尝试
             * 2.1获取锁失败，继续判断，判断时间戳，看是否可以重置并获取到锁
             *    setNx结果小于当前时间，表明锁已过期，可以再次尝试加锁
             */
            String lockValueStr = RedisPoolUtil.get(lockName);
            Long lockValueATime = Long.parseLong(lockValueStr);
            log.info("lockValueATime为:" + lockValueATime);
            if (lockValueStr != null && lockValueATime < System.currentTimeMillis()) {
    
                /**2.2再次用当前时间戳getset--->将给定 key 的值设为 value，并返回 key 的旧值(old value)
                 * 通过getset重设锁对应的值: 新的当前时间+超时时间，并返回旧的锁对应值
                 */
                String getSetResult = RedisPoolUtil.getSet(lockName, String.valueOf(System.currentTimeMillis() + lockTimeout));
                log.info("lockValueBTime为:" + Long.parseLong(getSetResult));
                if (getSetResult == null || (getSetResult != null && StringUtils.equals(lockValueStr, getSetResult))) {
                    /**
                     *2.3旧值判断，是否可以获取锁
                     *当key没有旧值时，即key不存在时，返回nil ->获取锁，设置锁过期时间
                     */
                    log.info("获取Redis分布式锁[成功],lockName={},CurrentThreadName={}",
                            lockName, Thread.currentThread().getName());
                    RedisPoolUtil.expire(lockName, EXPIRE_SECONDS);
                    return true;
                } else {
                    log.info("获取锁失败,lockName={},CurrentThreadName={}",
                            lockName, Thread.currentThread().getName());
                    return false;
                }
            } else {
                /**3.锁未超时，获取锁失败*/
                log.info("当前锁未失效！！！！，竞争失败，继续持有之前的锁,lockName={},CurrentThreadName={}",
                        lockName, Thread.currentThread().getName());
                return false;
            }
        }
    
        /**
         * 解锁
         *
         * @param lockName
         */
        @Override
        public boolean release(String lockName) {
            Long result = RedisPoolUtil.del(lockName);
            if (result != null && result.intValue() == 1) {
                log.info("删除Redis分布式锁成功，锁已释放, key= :{}", lockName);
                return true;
            }
            log.info("删除Redis分布式锁失败，锁未释放, key= :{}", lockName);
            return false;
        }
    }

[setnx_lua]: http://www.iocoder.cn/Fight/Correct-implementation-of-Redis-distributed-locks-by-Java/?self