背景

开发活动报名业务，涉及到活动人数限制的问题，当并发量上来的时候，多人同时提交报名信息，将会导致活动已报名人数的不准确，对业务造成影响，如下图：

分析出现问题的原因是，设置操作发生的时候，并没有确保当前人数的准确性，即没有确保当前查询出来的已报名人数与数据库的一致性，导致客户端并发的两次操作有被覆盖的情况发生

传统数据库 VS NoSql

mysql

针对如上场景，若报名人数字段保存在mysql数据库中，可以使用一种常见的降低读写锁冲突，保证数据一致性的乐观锁机制(Compare and Set CAS)，实现方案如下

将原来的操作sql代码 update act set num=#{numNew} where actId=#{actId}

复制代码

改为 update act set num=#{numNew} where actId=#{actId} and num=#{numOld}

复制代码

即只有当查询出来的数据与当前数据库的数据一致时，才可以进行赋值操作，否则失败

redis

若使用redis，则活动报名人数以键值对的形式存在内存中，业务代码将会对内存中的人数进行操作，相比mysql，redis的效率更高，不会造成很大的延迟(若当并发量很大时，使用mysql进行报名人数记录，CAS的方案将会导致很多客户端操作失败，用户体验不好)，但使用redis，其没有很好的事务支持，以上mysql的解决方案不能很好的运用在redis上，因此如何设计redis锁，进行共享资源(已报名活动人数)的操作，是需要解决的问题

使用到的命令说明

设计Redis锁之前，需要介绍下即将用到的几个命令

SETNX

将key设置值为value，如果key不存在，这种情况下等同SET命令，返回值1。 当key存在时，什么也不做，返回值0。

watch && MULTI

watch：标记所有指定的key 被监视起来，在事务中有条件的执行(乐观锁)

MULTI：标记一个事务块的开始。 随后的指令将在执行EXEC时作为一个原子执行

当两者一起使用的时候，首先key被watch监视，若在调用 EXEC 命令执行事务时， 如果任意一个被监视的键被其他客户端修改了， 那么整个事务不再执行， 直接返回失败。如下表：

时间

客户端A

客户端B

T1

WATCH name

T2

MULTI

T3

SET name owen

T4

SET name tom

T5

EXEC

在时间 T4 ，客户端 B 修改了 name 键的值， 当客户端 A 在 T5 执行 EXEC 时，Redis 会发现 name 这个被监视的键已经被修改， 因此客户端 A 的事务不会被执行，而是直接返回失败。

GETSET

GETSET key value 返回之前的旧值value，之后设置key的新值

Redis基本解决思路以及遇到的问题

以下列举使用redis锁的基本思路

注：例子使用spring-data-redis库，setnx命令变为setIfAbsent,并且返回true or false private StringRedisTemplate stringRedisTemplate;

public Boolean setConcurrentLock(String key) throws InterruptedException {

ValueOperations ops = stringRedisTemplate.opsForValue();

while (!ops.setIfAbsent(key, "lock"))) {

TimeUnit.MILLISECONDS.sleep(3);

}

return true;

}

public void deleteConcurrentLock(String key) {

stringRedisTemplate.delete(key);

}

复制代码

如上获取redis锁使用了setnx命令，若lock被占用，则返回false，一直循环，直到lock被删除后可以赋值成功，才能获得锁，实现对共享资源加锁。

但是，很明显，while存在死循环死锁的可能，当如下场景：

线程1获取到lock，线程2，线程3在执行while循环等待lock删除，若线程1突然挂掉，没能删除lock，则导致线程2，线程3死循环，死锁

想到解决方案为对锁设置超时，防止无限制循环，代码如下： private StringRedisTemplate stringRedisTemplate;

public Boolean setConcurrentLock(String key, long expireTime) throws InterruptedException {

ValueOperations ops = stringRedisTemplate.opsForValue();

//expireTime 为锁超时时间

while (!ops.setIfAbsent(key, String.valueOf(System.currentTimeMillis() + expireTime))) {

Long expire = Long.parseLong(ops.get(key));

//判断是否超时

if (expire != null && expire < System.currentTimeMillis()) {

//getset获取旧的时间，并且设置新的超时时间

Long oldExpire = Long.parseLong(ops.getAndSet(key, String.valueOf(System.currentTimeMillis() + expireTime)));

if (oldExpire != null && oldExpire < System.currentTimeMillis()) {

break;

}

}

TimeUnit.MILLISECONDS.sleep(3);

}

return true;

}

public void deleteConcurrentLock(String key) {

stringRedisTemplate.delete(key);

}

复制代码

若获取锁失败，进入while循环，判断超时时间是否已到，if判断为真，证明lock已经超时。所以执行getset命令，获取旧的时间，并设置新的超时时间，若获取的旧的时间超时了，则证明获取lock成功，跳出循环

但此处添加超时控制仍然存在问题，如下场景

场景一：

线程1获取lock并且挂掉，线程2，线程3 进入while循环后，同时判断出lock已经超时，线程2首先执行getset命令，返回了线程1设置的超时时间，确实超时，线程2获取锁；线程3执行getset命令，返回了线程2设置的超时时间，并未超时，但是线程3重新设置了超时时间

场景二：

有关删除锁的方法，若线程2持有锁期间超时，但是操作没有执行完，锁被线程3重新设置，变为线程3的锁，线程2执行完毕后，直接执行del，则会把线程3的锁删除，出现问题

Redis最终实践方案

针对上面列举的两个问题，修改代码的最终实践版如下： private StringRedisTemplate stringRedisTemplate;

public static ThreadLocal holder = new ThreadLocal<>();

public Boolean setConcurrentLock(String key, long expireTime) throws InterruptedException {

ValueOperations ops = stringRedisTemplate.opsForValue();

while (!ops.setIfAbsent(key, String.valueOf(System.currentTimeMillis() + expireTime))) {

stringRedisTemplate.watch(key);

Long expire = Long.parseLong(ops.get(key));

if (expire != null && expire < System.currentTimeMillis()) {

stringRedisTemplate.multi();

Long oldExpire = Long.parseLong(ops.getAndSet(key, String.valueOf(System.currentTimeMillis() + expireTime)));

if (stringRedisTemplate.exec() != null && oldExpire != null && oldExpire < System.currentTimeMillis()) {

break;

}

} else {

stringRedisTemplate.unwatch();

}

TimeUnit.MILLISECONDS.sleep(3);

}

holder.set(ops.get(key));

return true;

}

public void deleteConcurrentLock(String key) {

ValueOperations ops = stringRedisTemplate.opsForValue();

Long expire = Long.valueOf(ops.get(key));

if(exprie.equals(holder.get())){

stringRedisTemplate.delete(key);

}

holder.remove();

}

复制代码

如上面的场景一，首先线程2，线程3同时判断出lock超时后，对lock进行watch监视，然后将getset操作放到事务中执行，若线程2执行完事务，修改了lock的时间后，线程3由于执行事务命令lock被修改而失败，不会覆盖设置线程2的超时时间，解决场景一问题

对于场景二，为了防止已经超时的线程误删其他正在执行的线程lock，引入ThreadLock变量，将本线程设置的超时时间放入ThreadLock中，若删除的时候，从Redis取出的时间变化了，证明该线程超时，时间被其他线程重新设置过，就不需要删除lock。最后需要注意的是使用ThreadLocal需要在判断是够删除lock锁时手动删除，防止web服务器中的线程池对线程复用，造成ThreadLocal重复使用。

总结

本篇实践是基于单点redis服务器情况下的锁(若工程在多机器下部署，可以装逼的叫redis分布式锁)。但在redis集群架构下，如果master节点down机，由于redis主从复制是异步的，会有明显的race-condition。Redis文档中提供了一种解决方案：RedLock，后续有机会再去实践学习吧。。。