关于java:Redis并发阻塞锁方案

因为用户同时拜访线上的下订单接口，导致在扣减库存时呈现了异样，这是一个很典型的并发问题，本篇文章为解决并发问题而生，采纳的技术为 Redis 锁机制 + 多线程的阻塞唤醒办法。

在实现 Redis 锁机制之前，咱们须要理解一下前置常识。

一、前置常识

1、多线程

将 wait()、notifyAll() 归为到多线程的办法中略有一些不失当，这两个办法是 Object 中的办法。

① 当调用了 wait() 办法后，让以后线程进入期待状态，并且让以后线程开释对象锁，期待既为阻塞状态，期待 notifyAll() 办法的唤醒。

wait() 办法和 sleep() 办法有一些相似之处，都是使以后线程阻塞，但他们理论是有一些区别的。

1. 执行 wait() 办法之前须要申请锁，wait() 办法执行的时候会开释锁，期待被唤醒的时候竞争锁。
2. sleep() 只是让以后线程休眠一段时间，忽视锁的存在。
3. wait() 是 Object 类的办法 sleep() 是 Thread 的静态方法

② notifyAll() 办法为唤醒 wait() 中的线程。

notifyAll() 和 notify() 办法都是能够唤醒调用了 wait() 办法，而陷入阻塞的线程。

然而 notify() 是随机唤醒这个阻塞队列中随机的一个线程，而 notifyAll() 是唤醒所用的调用了 wait() 办法而陷入阻塞的线程，让他们本人去抢占对象锁。

notifyAll() 和 notify() 也都是必须在加锁的同步代码块中被调用，它们起的是唤醒的作用，不是开释锁的作用，只用在以后同步代码块中的程序执行完，也就是对象锁天然开释了，notifyAll() 和 notify() 办法才会起作用，去唤醒线程。

wait() 办法个别是和 notify() 或者 notifyAll() 办法一起连用的。

以上为把握本篇博客必备的多线程常识，如果零碎学习多线程的相干常识可查阅博客 程序员田同学

2、Redis

加锁的过程实质上就是往 Redis 中 set 值，当别的过程也来 set 值时候，发现外面曾经有值了，就只能放弃获取稍后再试。

Redis 提供了一个人造实现锁机制的办法。

 在 Redis 客户端的命令为 setnx(set if not exists) 

在集成 Springboot 中采纳的办法为：

redisTemplate.opsForValue().setIfAbsent(key, value);

如果外面 set 值胜利会返回 True，如果外面曾经存在值就会返回 False。

在咱们理论应用的时候，setIfAbsent() 办法并不是总是返回 True 和 False。

如果咱们的业务中加了事务，该办法会返回 null，不晓得这是一个 bug 还是什么，这是 Redis 的一个巨坑，节约了很长时间才发现了这个问题，如果解决此问题能够跳转到第四章。

二、实现原理

分布式锁实质上要实现的指标就是在 Redis 外面占一个地位，当别的过程也要来占时，发现曾经有人占在那里了，就只好放弃或者稍后再试。占位个别是应用 setnx(set if not exists) 指令，只容许被一个客户端占位。先来先占，事办完了，再调用 del 指令开释茅坑。

其中，发现 Redis 中曾经有值了，以后线程是间接放弃还是稍后再试别离就代表着，非阻塞锁和阻塞锁。

在咱们的业务场景中必定是要稍后再试（阻塞锁），如果是间接放弃（非阻塞锁）在数据库层面就能够间接做，就不须要咱们在代码大费周章了。

非阻塞锁只能保留数据的正确性，在高并发的状况下会抛出大量的异样，当一百个并发申请到来时，只有一个申请胜利，其余均会抛出异样。

Redis 非阻塞锁和 MySQL 的乐观锁，最终达到的成果是一样的，乐观锁是采纳 CAS 的思维。

乐观锁办法：表字段 加一个版本号，或者别的字段也能够！加版本号，能够晓得管制程序而已！在 update 的时候能够 where 前面加上 version= oldVersion。数据库，在任何并发的状况下，update 胜利就是 1 失败就是 0 . 能够依据返回的 1，0 做相应的解决！

咱们更举荐大家应用阻塞锁的形式。

当获取不到锁时候，咱们让以后线程应用 wait() 办法唤醒，当持有锁的线程应用实现后，调用 notifyAll() 唤醒所有期待的办法。

三、具体实现

以下代码为阻塞锁的实现形式。

业务层：

    public String test() throws InterruptedException {lock("lockKey");
        System.out.println("11");
        System.out.println("22");
        System.out.println(Thread.currentThread().getName()+"***********");
        Thread.sleep(2000);
        System.out.println("33");
        System.out.println("44");
        System.out.println("55");
        unlock("lockKey");
        return "String";
    }

锁的工具类:

次要是加锁和解锁的两个办法。

 // 每一个 redis 的 key 对应一个阻塞对象
    private static HashMap<String, Object> blockers = new HashMap<>();

    // 以后取得锁的线程
    private static Thread curThread;

    public static RedisTemplate redisTemplate = (RedisTemplate) SpringUtils.getBean("redisTemplate") ;

    /**
     * 加锁
     * @param key
     * @throws InterruptedException
     */

    public static void lock(String key) {
        // 循环判断是否可能创立 key，不能则间接 wait 开释 CPU 执行权

        // 放不进指阐明锁正在被占用
        System.out.println(key+"**");

        while (!RedisUtil.setLock(key,"1",3)){synchronized (key) {blockers.put(key, key);
                //wait 开释 CPU 执行权
                try {key.wait();
                } catch (InterruptedException e) {e.printStackTrace();
                }
            }
        }
        blockers.put(key, key);
        // 可能胜利创立，获取锁胜利记录以后获取锁线程
        curThread = Thread.currentThread();}

    /**
     * 解锁
     * @param key
     */
    public static void unlock(String key) {
        // 判断是否为加锁的线程执行解锁，不是则间接疏忽
        if(curThread == Thread.currentThread()) {RedisUtil.delete(key);
            // 删除 key 之后须要 notifyAll 所有的利用，所以这里采纳发订阅音讯给所有的利用
          //  RedisUtil.publish("lock", key);

            //notifllall 其余线程
            Object lock = blockers.get(key);
            if(lock != null) {synchronized (lock) {lock.notifyAll();
                }
            }

        }
    }

当咱们在不加锁时候，应用接口测试工具测试时，12345 并不能都是程序执行的，会造成输入程序不统一，如果是在咱们的理论场景中，这是输出换成了数据库的 select 和 update，数据呈现错乱也是很失常的状况了。

当咱们加上锁当前，12345 都是程序输入，并发问题顺利解决了。

四、附录

1、Redis 存在的 bug

原本 lock() 办法是间接调用 “Redis.setIfAbsent()” 办法，然而在应用时候始终报空指针异样，最终定位问题为 Redis.setIfAbsent() 办法存在问题。

在我的理论业务中，下订单的办法应用了 @Transflastion 减少了事务，导致该办法返回 null，咱们手写一个实现 setIfAbsent() 的作用。

 /**
     * 只有 key 不存在时, 才设置值, 返回 true, 否则返回 false
     *
     * @param key     key 不能为 null
     * @param value   value 不能为 null
     * @param timeout 过期时长, 单位为妙
     * @return
     */
    public static Boolean setLock(String key,String value, long timeout) {SessionCallback<Boolean> sessionCallback = new SessionCallback<Boolean>() {
            List<Object> exec = null;
            @Override
            @SuppressWarnings("unchecked")
            public Boolean execute(RedisOperations operations) throws DataAccessException {operations.multi();

                redisTemplate.opsForValue().setIfAbsent(key, value);
                redisTemplate.expire(key,timeout, TimeUnit.SECONDS);

                exec = operations.exec();

                if(exec.size() > 0) {return (Boolean) exec.get(0);
                }
                return false;
            }
        };
        return (Boolean) redisTemplate.execute(sessionCallback);
    }

不便比照，以下贴上本来的 setIfAbsent() 办法。

 /**
   * 只有 key 不存在时, 才设置值, 返回 true, 否则返回 false [正告：事务或者管道状况下会报错 - 可应用 setLock 办法]
   *
   * @param key     key 不能为 null
   * @param value   value 不能为 null
   * @param timeout 过期时长, 单位为妙
   * @return
   */
  @Deprecated
  public static <T> Boolean setIfAbsent(String key, T value, long timeout) {// redisTemplate.multi();
      ValueOperations<String, T> valueOperations = redisTemplate.opsForValue();
      Boolean aBoolean = valueOperations.setIfAbsent(key, value, timeout, TimeUnit.SECONDS);
     // redisTemplate.exec();
    return aBoolean;
  }

2、MySQL 的锁机制

在并发场景下 MySQL 会报错，报错信息如下：

### Cause: com.mysql.cj.jdbc.exceptions.MySQLTransactionRollbackException: Lock wait timeout exceeded; try restarting transaction
; SQL []; Lock wait timeout exceeded; try restarting transaction; nested exception is com.mysql.cj.jdbc.exceptions.MySQLTransactionRollbackException: Lock wait timeout exceeded; try restarting transaction

问题呈现的起因是，某一种表频繁被锁表，导致另外一个事务超时，呈现问题的起因是 MySQL 的机制。

MySQL 更新时如果 where 字段存在索引会应用行锁，否则会应用表锁。

咱们应用 navichat 在 where 字段上加上索引，问题顺利的迎刃而解。