Redis上实现分布式锁以提高性能的方案研究

背景：

在很多互联网产品应用中，有些场景需要加锁处理，比如：秒杀，全局递增ID，楼层生成等等。大部分是解决方案基于DB实现的，Redis为单进程单线程模式，采用队列模式将并发访问变成串行访问，且多客户端对Redis的连接并不存在竞争关系。

项目实践

任务队列用到分布式锁的情况比较多，在将业务逻辑中可以异步处理的操作放入队列，在其他线程中处理后出队，此时队列中使用了分布式锁，保证入队和出队的一致性。关于redis队列这块的逻辑分析，我将在下一次对其进行总结，此处先略过。

接下来对redis实现的分布式锁的逻辑代码进行详细的分析和理解：

1、为避免特殊原因导致锁无法释放, 在加锁成功后, 锁会被赋予一个生存时间(通过 lock 方法的参数设置或者使用默认值), 超出生存时间锁将被自动释放.

2、锁的生存时间默认比较短(秒级, 具体见 lock 方法), 因此若需要长时间加锁, 可以通过 expire 方法延长锁的生存时间为适当的时间. 比如在循环内调用 expire

3、系统级的锁当进程无论因为任何原因出现crash，操作系统会自己回收锁，所以不会出现资源丢失。

4、但分布式锁不同。若一次性设置很长的时间，一旦由于各种原因进程 crash 或其他异常导致 unlock 未被调用，则该锁在剩下的时间就变成了垃圾锁，导致其他进程或进程重启后无法进入加锁区域。

1. <?php
3. require_once ‘RedisFactory.php’;
5. /**
6. * 在 Redis 上实现的分布式锁
7. */
8. class RedisLock {
10. //单例模式
11.   private static $_instance = null;
12.   public static function instance() {
13.     if(self::$_instance == null) {
14.       self::$_instance = new RedisLock();
15.     }
16.     return self::$_instance;
17.   }
20. //redis对象变量
21.   private $redis;
23. //存放被锁的标志名的数组
24.   private $lockedNames = array();
26.   public function __construct() {
28. //获取一个 RedisString 实例
29.     $this->redis = RedisFactory::instance()->getString();
30.   }
33. /**
35. * 加锁
37. *
39. * @param string 锁的标识名
41. * @param int 获取锁失败时的等待超时时间(秒), 在此时间之内会一直尝试获取锁直到超时. 为 0 表示失败后直接返回不等待
43. * @param int 当前锁的最大生存时间(秒), 必须大于 0 . 如果超过生存时间后锁仍未被释放, 则系统会自动将其强制释放
45. * @param int 获取锁失败后挂起再试的时间间隔(微秒)
47. */
48.   public function lock($name, $timeout = 0, $expire = 15, $waitIntervalUs = 100000) {
49.     if(empty($name)) return false;
51.     $timeout = (int)$timeout;
52.     $expire = max((int)$expire, 5);
53.     $now = microtime(true);
54.     $timeoutAt = $now + $timeout;
55.     $expireAt = $now + $expire;
57.     $redisKey = “Lock:$name”;
58.     while(true) {
59.       $result = $this->redis->setnx($redisKey, (string)$expireAt);
60.       if($result !== false) {
62. //对$redisKey设置生存时间
63.         $this->redis->expire($redisKey, $expire);
65. //将最大生存时刻记录在一个数组里面
66.         $this->lockedNames[$name] = $expireAt;
67.         return true;
68.       }
71. //以秒为单位，返回$redisKey 的剩余生存时间
72.       $ttl = $this->redis->ttl($redisKey);
74. // TTL 小于 0 表示 key 上没有设置生存时间(key 不会不存在, 因为前面 setnx 会自动创建)
76. // 如果出现这种情况, 那就是进程在某个实例 setnx 成功后 crash 导致紧跟着的 expire 没有被调用. 这时可以直接设置 expire 并把锁纳为己用
77.       if($ttl < 0) {
78.         $this->redis->set($redisKey, (string)$expireAt, $expire);
79.         $this->lockedNames[$name] = $expireAt;
80.         return true;
81.       }
84. // 设置了不等待或者已超时
85.       if($timeout <= 0 || microtime(true) > $timeoutAt) break;
88. // 挂起一段时间再试
89.       usleep($waitIntervalUs);
90.     }
92.     return false;
93.   }
96. /**
98. * 给当前锁增加指定的生存时间(秒), 必须大于 0
100. *
102. * @param string 锁的标识名
104. * @param int 生存时间(秒), 必须大于 0
106. */
107.   public function expire($name, $expire) {
108.     if($this->isLocking($name)) {
109.       if($this->redis->expire(“Lock:$name”, max($expire, 1))) {
110.         return true;
111.       }
112.     }
113.     return false;
114.   }
117. /**
119. * 判断当前是否拥有指定名称的锁
121. *
123. * @param mixed $name
125. */
126.   public function isLocking($name) {
127.     if(isset($this->lockedNames[$name])) {
128.       return (string)$this->lockedNames[$name] == (string)$this->redis->get(“Lock:$name”);
129.     }
130.     return false;
131.   }
134. /**
136. * 释放锁
138. *
140. * @param string 锁的标识名
142. */
143.   public function unlock($name) {
144.     if($this->isLocking($name)) {
145.       if($this->redis->deleteKey(“Lock:$name”)) {
146.         unset($this->lockedNames[$name]);
147.         return true;
148.       }
149.     }
150.     return false;
151.   }
154. /** 释放当前已经获取到的所有锁 */
155.   public function unlockAll() {
156.     $allSuccess = true;
157.     foreach($this->lockedNames as $name => $item) {
158.       if(false === $this->unlock($name)) {
159.         $allSuccess = false;
160.       }
161.     }
162.     return $allSuccess;
163.   }
164. }

此类很多代码都写上了注释，只要认真理解下，就很容易懂得如何在redis实现分布式锁了。