锁的使用

一、业务场景分析

　　在实际开发中，我们需要使用到锁，来防止并发问题，以秒杀为例，如果没有锁的处理，就会发生超卖的问题，超卖问题的出现，是由于多线程并发处理，一个事务未提交，但是另外的线程来查询时，可以查询到仍有库存，就会发生超卖。

　　解决这类问题，一般有三种方案：

　　　　1、使用锁处理

　　　　　　也就是让所有的操作都串行的执行，

　　　　2、让所有的操作原子性执行

　　　　　　因为redis是天然的单线程的操作，因此可以使用redis进行处理，也就是将库存数据存储在redis中，使用 res = hincrent(“seckill_goods_stock_1”,-1) 进行库存的扣减，不过这里还需要判断一下库存是否需要满足。

　　　　3、使用队列

　　　　　　队列的长度等于库存的数量，然后队列中的存储的数据是商品ID，然后使用POP操作，进行扣减库存，同时每一种商品对应一个队列。

　　下面就从锁的角度来解决这种问题。

　　针对项目部署的情况，可以使用单机锁和分布式锁，单机锁和分布式锁，分布式锁又可以使用数据库、Redis、ZK、etcd来实现

二、单机锁

（一）单机锁与事务

 　　JDK中提供了ReentranLock，胆码逻辑如下。

    //程序锁
    //互斥锁 参数默认false，不公平锁
    private Lock lock = new ReentrantLock(true);
public HttpResult startKilled(Long killId, String userId){
        // 加锁
        lock.lock();
        try {
            // 执行业务
        } catch (Exception e) {
            //异常处理
        } finally {
            // 释放锁
            lock.unlock();
        }
        return null;
    }

　　其实这样是有个问题的，就是会存在超卖问题，这是因为事务是在该方法执行完之后才会提交，但是在事务提交前，就是释放了锁，导致超卖。

　　对于这种事务在释放锁之后的问题，只需要将加锁和释放锁的操作往上提取，在事务提交之后处理即可，但是所有的操作加锁和解锁操作都放在service上层，就会使Controller层太臃肿，因此可以使用自定义注解来进行处理。

（二）AOP方式解决超卖问题

　　1、定义一个注解

@Target({ElementType.PARAMETER,ElementType.METHOD})
@Retention(RetentionPolicy.RUNTIME)
@Documented
public @interface ServiceLock {

    String description() default "";

}

　　2、编写AOP切面

　　在切面中使用自定义的注解，然后使用环绕通知，在业务处理的前后进行加锁和解锁操作。

　　这里需要特殊说明一点，由于该注解需要在事务注解前执行，因此需要@Order注解来注明注解的执行顺序，oder为int类型，越小越早执行，由于事务注解没有Order，那么就说明其Order的默认值为int的最大值（2147483647），因此只要设置order大于该值，那么该注解就会在事务注解前执行。

@Component
@Scope
@Aspect
@Order(1)
public class LockAspect {
    // 定义锁对象
    private static Lock lock = new ReentrantLock(true);
    @Pointcut("@annotation(com.sugo.seckill.aop.lock.ServiceLock)")
    public void lockAspect(){

    }
    // 增强方法
    @Around("lockAspect()")
    public Object around(ProceedingJoinPoint joinPoint){
        Object obj = null;
        // 加锁
        lock.lock();
        // 执行业务
        try {
            obj = joinPoint.proceed();
        } catch (Throwable throwable) {
            throwable.printStackTrace();
        } finally {
            // 释放锁
            lock.unlock();
        }
        return obj;
    }
}

　　3、业务代码使用单机锁注解

    @Transactional
    @ServiceLock
    @Override
    public HttpResult startKilledByLocked(Long killId, String userId) {
            //业务处理
            //下单
            return HttpResult.ok("秒杀成功");
        } catch (Exception e) {
            e.printStackTrace();
        }
        return null;
    }

三、分布式锁--Mysql

　　对于分布式部署的项目来说，单机锁肯定不行，因此就需要分布式锁。所谓分布式锁，就是依赖于第三方的组件，完成锁的处理，目前比较常见的，可以使用数据库、Redis、ZK、etcd等。

　　对于单机锁，其锁的是线程，也就是防止多线程操作同一资源，而分布式锁，其锁的是进程，也就是防止多个进程间操作同一资源。

　　锁分为悲观锁和乐观锁。

　　1、悲观锁的实现方式

　　　　悲观锁的实现方式，其实就是在查询的时候使用for update，这种处理的好处是非常简单，缺点是非常影响性能。

    @Select(value = "select * from tb_seckill_goods where id = #{seckillId} for update")
    TbSeckillGoods selectByPrimaryKeyBySQLLock(Long seckillId);

　　2、乐观锁的实现方式

　　　　乐观锁的实现方式，其实就是加上一个版本号，在提交事务的时候保证提交时的版本和查询时的版本号一致。

    @Update(value = "UPDATE tb_seckill_goods SET stock_count=stock_count-1 WHERE id=#{seckillId} AND stock_count>0")
    int updateSeckillGoodsByPrimaryKeyByLock(@Param("seckillId") Long seckillId);

四、分布式锁--redis

（一）Reddssion客户端

　　如果使用Redis锁，强烈建议使用Redssion客户端，因为Redssion客户端已经对分布式锁做的非常完善。对于可重入锁、异步锁、公平锁、联锁、红锁都做了实现，具体的demo如下所示。

/**
 * redis分布式锁Demo
 * @author hubin
 */
public class RedissLockDemo {
    /**
     * 可重入锁（Reentrant Lock） 
     * Redisson的分布式可重入锁RLock Java对象实现了java.util.concurrent.locks.Lock接口，同时还支持自动过期解锁
     * @param redisson
     */
    public void testReentrantLock(RedissonClient redisson) {
        RLock lock = redisson.getLock("anyLock");
        try {
            // 1. 最常见的使用方法
            // lock.lock();
            // 2. 支持过期解锁功能,10秒钟以后自动解锁, 无需调用unlock方法手动解锁
            // lock.lock(10, TimeUnit.SECONDS);
            // 3. 尝试加锁，最多等待3秒，上锁以后10秒自动解锁
            boolean res = lock.tryLock(3, 10, TimeUnit.SECONDS);
            if (res) { // 成功
                // do your business
            }
        } catch (InterruptedException e) {
            e.printStackTrace();
        } finally {
            lock.unlock();
        }
    }
    /**
     * Redisson同时还为分布式锁提供了异步执行的相关方法
     * @param redisson
     */
    public void testAsyncReentrantLock(RedissonClient redisson) {
        RLock lock = redisson.getLock("anyLock");
        try {
            lock.lockAsync();
            lock.lockAsync(10, TimeUnit.SECONDS);
            Future<Boolean> res = lock.tryLockAsync(3, 10, TimeUnit.SECONDS);
            if (res.get()) {
                // do your business
            }
        } catch (InterruptedException e) {
            e.printStackTrace();
        } catch (ExecutionException e) {
            e.printStackTrace();
        } finally {
            lock.unlock();
        }
    }
    /**
     * 公平锁（Fair Lock）
     * Redisson分布式可重入公平锁也是实现了java.util.concurrent.locks.Lock接口的一种RLock对象。
     * 在提供了自动过期解锁功能的同时，保证了当多个Redisson客户端线程同时请求加锁时，优先分配给先发出请求的线程。
     * @param redisson
     */
    public void testFairLock(RedissonClient redisson){  
        RLock fairLock = redisson.getFairLock("anyLock");  
        try{  
            // 最常见的使用方法  
            fairLock.lock();  
            // 支持过期解锁功能, 10秒钟以后自动解锁,无需调用unlock方法手动解锁  
            fairLock.lock(10, TimeUnit.SECONDS);  
            // 尝试加锁，最多等待100秒，上锁以后10秒自动解锁  
            boolean res = fairLock.tryLock(100, 10, TimeUnit.SECONDS);  
            if (res) {
                // do your business
            }
        } catch (InterruptedException e) {  
            e.printStackTrace();  
        } finally {  
            fairLock.unlock();  
        }  
//      Redisson同时还为分布式可重入公平锁提供了异步执行的相关方法：
//        RLock fairLock = redisson.getFairLock("anyLock");  
//        fairLock.lockAsync();  
//        fairLock.lockAsync(10, TimeUnit.SECONDS);  
//        Future<Boolean> res = fairLock.tryLockAsync(100, 10, TimeUnit.SECONDS);  
    } 
    /**
     * 联锁（MultiLock）
     * Redisson的RedissonMultiLock对象可以将多个RLock对象关联为一个联锁，每个RLock对象实例可以来自于不同的Redisson实例
     * @param redisson1
     * @param redisson2
     * @param redisson3
     */
    public void testMultiLock(RedissonClient redisson1,RedissonClient redisson2, RedissonClient redisson3){  
        RLock lock1 = redisson1.getLock("lock1");  
        RLock lock2 = redisson2.getLock("lock2");  
        RLock lock3 = redisson3.getLock("lock3");  
        RedissonMultiLock lock = new RedissonMultiLock(lock1, lock2, lock3);  
        try {  
            // 同时加锁：lock1 lock2 lock3, 所有的锁都上锁成功才算成功。  
            lock.lock();  
            // 尝试加锁，最多等待100秒，上锁以后10秒自动解锁  
            boolean res = lock.tryLock(100, 10, TimeUnit.SECONDS);  
            if (res) {
                // do your business
            }
        } catch (InterruptedException e) {  
            e.printStackTrace();  
        } finally {  
            lock.unlock();  
        }  
    }  
    /**
     *  红锁（RedLock）
     *  Redisson的RedissonRedLock对象实现了Redlock介绍的加锁算法。该对象也可以用来将多个RLock对象关联为一个红锁，每个RLock对象实例可以来自于不同的Redisson实例
     * @param redisson1
     * @param redisson2
     * @param redisson3
     */
    public void testRedLock(RedissonClient redisson1,RedissonClient redisson2, RedissonClient redisson3){  
        RLock lock1 = redisson1.getLock("lock1");  
        RLock lock2 = redisson2.getLock("lock2");  
        RLock lock3 = redisson3.getLock("lock3");  
        RedissonRedLock lock = new RedissonRedLock(lock1, lock2, lock3);  
        try {  
            // 同时加锁：lock1 lock2 lock3, 红锁在大部分节点上加锁成功就算成功。  
            lock.lock();  
            // 尝试加锁，最多等待100秒，上锁以后10秒自动解锁  
            boolean res = lock.tryLock(100, 10, TimeUnit.SECONDS);  
            if (res) {
                // do your business
            }
        } catch (InterruptedException e) {  
            e.printStackTrace();  
        } finally {  
            lock.unlock();  
        }  
    }  
    //读写锁（ReadWriteLock）、信号量（Semaphore）、可过期性信号量（PermitExpirableSemaphore）、闭锁（CountDownLatch）
}

（二）对于超卖问题使用redis

　　无论对于Redis锁还是ZK锁，其实都和单机锁存在同样的问题，就是事务提交时间的问题，所以使用Redis和ZK都可以使用AOP将锁的处理前移。

　　1、添加Redisson工具类

public class RedissLockUtil {
    private static RedissonClient redissonClient;
    
    public void setRedissonClient(RedissonClient locker) {
        redissonClient = locker;
    }
    
    /**
     * 加锁
     * @param lockKey
     * @return
     */
    public static RLock lock(String lockKey) {
        RLock lock = redissonClient.getLock(lockKey);
        lock.lock();
        return lock;
    }

    /**
     * 释放锁
     * @param lockKey
     */
    public static void unlock(String lockKey) {
        RLock lock = redissonClient.getLock(lockKey);
        lock.unlock();
    }
    
    /**
     * 释放锁
     * @param lock
     */
    public static void unlock(RLock lock) {
        lock.unlock();
    }

    /**
     * 带超时的锁
     * @param lockKey
     * @param timeout 超时时间   单位：秒
     */
    public static RLock lock(String lockKey, int timeout) {
        RLock lock = redissonClient.getLock(lockKey);
        lock.lock(timeout, TimeUnit.SECONDS);
        return lock;
    }
    
    /**
     * 带超时的锁
     * @param lockKey
     * @param unit 时间单位
     * @param timeout 超时时间
     */
    public static RLock lock(String lockKey, TimeUnit unit ,int timeout) {
        RLock lock = redissonClient.getLock(lockKey);
        lock.lock(timeout, unit);
        return lock;
    }
    
    /**
     * 尝试获取锁
     * @param lockKey
     * @param waitTime 最多等待时间
     * @param leaseTime 上锁后自动释放锁时间
     * @return
     */
    public static boolean tryLock(String lockKey, int waitTime, int leaseTime) {
        RLock lock = redissonClient.getLock(lockKey);
        try {
            return lock.tryLock(waitTime, leaseTime, TimeUnit.SECONDS);
        } catch (InterruptedException e) {
            return false;
        }
    }
    
    /**
     * 尝试获取锁
     * @param lockKey
     * @param unit 时间单位
     * @param waitTime 最多等待时间
     * @param leaseTime 上锁后自动释放锁时间
     * @return
     */
    public static boolean tryLock(String lockKey, TimeUnit unit, int waitTime, int leaseTime) {
        RLock lock = redissonClient.getLock(lockKey);
        try {
            return lock.tryLock(waitTime, leaseTime, unit);
        } catch (InterruptedException e) {
            return false;
        }
    }

    /**
     * 初始红包数量
     * @param key
     * @param count
     */
    public void initCount(String key,int count) {
        RMapCache<String, Integer> mapCache = redissonClient.getMapCache("skill");
        mapCache.putIfAbsent(key,count,3,TimeUnit.DAYS);
    }
    /**
     * 递增
     * @param key
     * @param delta 要增加几(大于0)
     * @return
     */
    public int incr(String key, int delta) {
        RMapCache<String, Integer> mapCache = redissonClient.getMapCache("skill");
        if (delta < 0) {
            throw new RuntimeException("递增因子必须大于0");
        }
        return  mapCache.addAndGet(key, 1);//加1并获取计算后的值
    }

    /**
     * 递减
     * @param key 键
     * @param delta 要减少几(小于0)
     * @return
     */
    public int decr(String key, int delta) {
        RMapCache<String, Integer> mapCache = redissonClient.getMapCache("skill");
        if (delta < 0) {
            throw new RuntimeException("递减因子必须大于0");
        }
        return mapCache.addAndGet(key, -delta);//加1并获取计算后的值
    }
}

　　2、添加自定义注解

@Target({ElementType.PARAMETER,ElementType.METHOD})
@Retention(RetentionPolicy.RUNTIME)
@Documented
public @interface ServiceRedisLock {
    String description() default "";
}

　　2、AOP

@Component
@Scope
@Aspect
@Order(1)
public class LockRedisAspect {
    @Pointcut("@annotation(com.sugo.seckill.aop.redis.ServiceRedisLock)")
    public void lockAspect(){
    }
    // 增强方法
    @Around("lockAspect()")
    public Object around(ProceedingJoinPoint joinPoint){
        Object obj = null;
        // 加锁
        boolean res = RedissLockUtil.tryLock(Constants.DISTRIBUTED_REDIS_LOCK_KEY,
                TimeUnit.SECONDS, 3,
                10);
        // 执行业务
        try {
            if(res){

                obj = joinPoint.proceed();
            }
        } catch (Throwable throwable) {
            throwable.printStackTrace();
        } finally {
            // 释放锁
            if(res){

                RedissLockUtil.unlock(Constants.DISTRIBUTED_REDIS_LOCK_KEY);
            }
        }
        return obj;
    }
}

　　3、业务代码

　　　　业务代码和使用单机锁的业务代码一致，只需要替换锁的注解即可。

　　4、总结

　　　　Redis 分布式设置时候的时候，为了防止线程阻塞，设置了锁的等待时候，锁的等待时间一旦设置，意味着一旦网络延迟，加锁时间超时，导致加锁失败；

五、分布式锁--zk

（一）zk锁demo

　　zk客户端有ZkClient和Curator ，其中ZkClient 是一个开源客户端，在 Zookeeper 原生 API 接口的基础上进行了包装，更便于开发人员使用。内部实现了 Session 超时重连，Watcher 反复注册等功能。像 dubbo 等框架 对其也进行了集成使用。Curator 是 Netflix 公司开源的一套 zk 客户端框架，与 ZkClient 一样，其也封装了 zk 原生 API。其目前已经成为 Apache 的顶级项目。同时，Curator 还提供了一套易用性、可读性更 强的 Fluent 风格的客户端 API 框架。

 　　这里使用Curator来进行实现分布式锁。

/**
 * 基于curator的zookeeper分布式锁
 * 这里我们开启5个线程，每个线程获取锁的最大等待时间为5秒，为了模拟具体业务场景，方法中设置4秒等待时间。
 * 开始执行main方法，通过ZooInspector监控/curator/lock下的节点如下图：
 */
public class CuratorUtil {
    private static String address = "192.168.1.180:2181";
    
    public static void main(String[] args) {
        //1、重试策略：初试时间为1s 重试3次
        RetryPolicy retryPolicy = new ExponentialBackoffRetry(1000, 3); 
        //2、通过工厂创建连接
        CuratorFramework client = CuratorFrameworkFactory.newClient(address, retryPolicy);
        //3、开启连接
        client.start();
        //4 分布式锁
        final InterProcessMutex mutex = new InterProcessMutex(client, "/curator/lock"); 
        //读写锁
        //InterProcessReadWriteLock readWriteLock = new InterProcessReadWriteLock(client, "/readwriter");
        
        ExecutorService fixedThreadPool = Executors.newFixedThreadPool(5);
        
        for (int i = 0; i < 5; i++) {
            fixedThreadPool.submit(new Runnable() {
                @Override
                public void run() {
                    boolean flag = false;
                    try {
                        //尝试获取锁，最多等待5秒
                        flag = mutex.acquire(5, TimeUnit.SECONDS);
                        Thread currentThread = Thread.currentThread();
                        if(flag){
                            System.out.println("线程"+currentThread.getId()+"获取锁成功");
                        }else{
                            System.out.println("线程"+currentThread.getId()+"获取锁失败");
                        }
                        //模拟业务逻辑，延时4秒
                        Thread.sleep(4000);
                    } catch (Exception e) {
                        e.printStackTrace();
                    } finally{
                        if(flag){
                            try {
                                mutex.release();
                            } catch (Exception e) {
                                e.printStackTrace();
                            }
                        }
                    }
                }
            });
        }
    }
}

（二）zk锁实现业务

　　1、自定义注解

@Target({ElementType.PARAMETER,ElementType.METHOD})
@Retention(RetentionPolicy.RUNTIME)
@Documented
public @interface ServiceZkLock {

    String description() default "";

}

　　2、AOP

@Component
@Scope
@Aspect
@Order(1)
public class LockZkAspect {
    @Pointcut("@annotation(com.sugo.seckill.aop.zk.ServiceZkLock)")
    public void lockAspect(){
    }
    // 增强方法
    @Around("lockAspect()")
    public Object around(ProceedingJoinPoint joinPoint){
        Object obj = null;
        // 加锁
        boolean acquire = ZkLockUtil.acquire(10, TimeUnit.SECONDS);
        // 执行业务
        try {
            if(acquire){

                obj = joinPoint.proceed();
            }
        } catch (Throwable throwable) {
            throwable.printStackTrace();
        } finally {
            // 释放锁
            if(acquire){

                ZkLockUtil.release();
            }
        }
        return obj;
    }
}

 　　3、业务实现

　　　　业务实现只需要替换注解即可。