分布式锁的Redis实现

　　当我们开始开发项目部署运行时，项目规模不大，只是在一个JVM实例中运行，对同一资源的并发访问用JDK自带的锁机制就可以解决资源同时访问的问题。而随着项目的不断发展，单体应用已经无法满足日益增长的访问需求，我们开始考虑多台部署，提高接收客户端的连接请求，提高项目的吞吐量。一台变多台，其中不可避免的问题就是如何控制解决不同线程对同一资源的并发访问。其中一种手段就是使用redis进行分布式锁的控制。

　　我们可以在获取访问资源锁之前判断redis中是否存在对应代表该资源锁key的value，如果存在，则说明已经被获取，反之还没有客户端获取该资源对应的锁，可以进行获取锁。

 1     boolean lock = false;
 2         try {
 3             lcok = getLock(taskId);   //获取锁
 4             if (lock) {
 5                 doSomething();   //业务逻辑
 6             }
 7         } finally {
 8             if (lock) {
 9                 releaseLock(taskId);   //释放锁
10             }
11         }

1 public static boolean getLock(String taskId) {
2         if (existsKey(taskId)) {
3             return false;
4         } else {
5             setKey(taskId);
6             return true;
7         }
8     }

　　上面的部分实现代码给了一个大概的解决思路，看起来没有问题的，但是仔细看看还是存在问题滴，存在什么问题呢？

　　当正在执行doSomething()方法时，突然系统宕机挂掉了，无法执行释放锁的操作，redis中对应的资源key的锁一直存在，之后运行代码就会出现问题。另一个问题就是执行getLock(taskId)方法时，该方法不是原子性的，有可能同时两个线程都判断为不存在该资源锁，都执行了setKey方法，导致同时获得锁资源的情况。

　　如何解决上面的两个问题呢？从Redis官方API中有SET my_key my_value NX PX milliseconds的方法，得到了解决方案。它提供了一个只有在某个key不存在的情况下才会设置key的值的原子命令，该命令也能设置key值过期时间。其中，NX表示只有当键key不存在的时候才会设置key的值，PX表示设置键key的过期时间，单位是毫秒。

　　到现在是否完全解决了并发获取锁的问题了呢？系统可能存在这种情况，当客户端A获取锁之后，执行业务代码的时间超过了之前设置的过期时间，导致锁的自动释放，而客户端B刚好获得新的资源锁，但客户端A恰好执行完业务操作，释放锁的时候，该锁是客户端B重新获得的锁，导致出现问题。这时，我们想到可以在设置key值时给定一个随机数，在释放资源锁的同时，判断是否和之前设置的value值相同，相同则释放，反之不释放。

1     if(getKey(taskId)==random_value){
2             deleteKey(taskId);
3         }

　　很可惜，上面的整个if操作也不是原子性的，getKey方法和deleteKey方法之间由于某种原因而延迟1秒钟操作了，而这1秒内刚好设置的的超时时间而锁释放，被新的客户端获得锁，1秒之后执行deleteKey方法又会误删除新客户端的锁，问题依旧存在。接下来我们只要想办法解决上面判断的原子性就能解决误删除锁的问题。Redis可以使用Lua脚本保证操作的原子性。

1 if redis.call("get",KEYS[1]) == ARGV[1] then
2     return redis.call("del",KEYS[1])
3 else
4     return 0
5 end

　　其中ARGV[1]表示设置key时指定的随机值。由于Lua脚本的原子性，在Redis执行该脚本的过程中，其他客户端的命令都需要等待该Lua脚本执行完才能执行，所以不会出现上面所说的误删除锁问题。至此，使用Redis实现分布式锁的方案就相对完善了。上述分布式锁的实现方案中，都是针对单节点Redis而言的。