0X00 测试环境
CentOS 6.6 + Redis 3.2.10 + PHP 7.0.7（+ phpredis 4.1.0）
[root@localhost ~]# cat /etc/issue
CentOS release 6.6 (Final)
Kernel \r on an \m [root@localhost ~]# redis-server -v
Redis server v=3.2. sha=: malloc=jemalloc-3.6. bits= build=8903a4502b3c9f88
[root@localhost ~]# php -v
PHP 7.0. (cli) (built: Feb ::) ( NTS )
Copyright (c) - The PHP Group
Zend Engine v3.0.0, Copyright (c) - Zend Technologies
with Xdebug v2.5.5, Copyright (c) -, by Derick Rethans
0X01 什么是分布式锁
redis 官网上对分布式锁的描述（https://redis.io/topics/distlock）是：
Distributed locks are a very useful primitive in many environments where different processes must operate with shared resources in a mutually exclusive way.
即在很多环境中，分布式锁是一种非常有用的原语，它使不同的进程必须以 互斥 的方式操作共享资源。
0x02 为什么要使用分布式锁
Redis 是 单线程（single-threaded）的内存数据结构存储，因此 Redis 所有的 基础命令 都是 原子性 的。但是 多个连贯的命令 在 高并发 的情况下数据的 一致性 就不能得到保障，数据很可能会被其他的客户端修改。
举一个并发情况下没有使用锁的例子：
例1. without_lock.php
<?php
$redis = new Redis();
$redis->connect('localhost', 6379); echo date('Y-m-d H:i:s', time()).' start'.PHP_EOL; for($i = 0; $i < 100000; $i++)
{
$count = (int)$redis->get('key');
$count += 1;
$redis->set('key', $count); usleep(0.01);
} echo date('Y-m-d H:i:s', time()).' end'.PHP_EOL;
在两个客户端中同时执行该程序，如果不考虑并发情况下数据的一致性，那么两个客户端执行完之后，键 key 的值应该是 200000，但是实际的结果小于 200000。
客户端1：
客户端2：
结果：
原因是：
get、值+1、set 这三个操作不是原子操作，在两个客户端同时执行脚本的时候，哪一个客户端先到就先执行哪个客户端的命令。例如：
a.当前 key 的值是 10；
b.客户端1 取出 key 的值是 10，此时客户端2 的命令也到了，取出 key 的值是 10；
c.客户端1 把值加 1，存入 key，此时 key 的值是 11；
d.客户端2 把值加 1，存入 key，此时 key 的值还是 11
所以在并发情况下最终 key 的值会小于希望的值。
0x03 Redis 的事务能不能解决并发下数据一致性问题
redis 中有一系列事务（https://redis.io/topics/transactions）相关的命令，包括 watch、multi、exec 等。能不能使用这些命令保证并发情况下的数据一致性呢。
根据 redis 官网的介绍，redis 可以使用 check-and-set（CAS）实现 乐观锁（Optimistic），以下是官网给出的示例：
WATCH mykey
val = GET mykey
val = val + 1
MULTI
SET mykey $val
EXEC
使用 watch 命令监视键 mykey，如果在 exec 命令执行之前，其他的客户端修改了 mykey 的值时，整个事务就会 终止，并且 exec 命令会返回 null 通知事务失败 —— 根据官网的说明，在接到事务失败的情况下，只需要重复执行上述操作，并且希望不会有新的竞态情况发生，这种形式的锁被称为乐观锁。如果把事务应用在例1 中，结果很可能会出现大量的事务失败，而并不能达到希望的结果，即最终 key 的值是 200000。
例2. whith_watch.php
<?php
$redis = new Redis();
$redis->connect('localhost', 6379); echo date('Y-m-d H:i:s', time()).' start'.PHP_EOL; $falseCount = 0;
for($i = 0; $i < 100000; $i++)
{
$redis->watch('key1');
$count = (int)$redis->get('key1');
$count += 1; $ret = $redis->multi()
->set('key1', $count)
->exec(); if(false === $ret)
{
$falseCount += 1;
}
} echo "falseCount:{$falseCount}".PHP_EOL;
echo date('Y-m-d H:i:s', time()).' end'.PHP_EOL;
客户端1：
事务失败了 33043 次
客户端2：
事务失败了 63228 次
结果：
结果也是远远小于 200000
说明：
以上为什么不能写成
for($i=0; $i < 100000; $i++)
{
$redis->watch('key1');
$redis->multi();
$count = (int)$redis->get('key1');
$count += 1;
$redis->set('key1', $count);
$redis->exec();
}
参考 phpredis 文档 https://github.com/phpredis/phpredis/#multi：
multi() returns the Redis instance and enters multi-mode. Once in multi-mode, all subsequent method calls return the same object until exec() is called.
即在 multi-mode 下，所有的后续方法都返回同一个对象（Redis Object），直到调用 exec命令。也就是说在 multi-mode 下，任何的命令都不会真正执行，而是会返回 Redis Object，直到调用 exec 命令，才真正执行事务中的每一条命令，因此
$count = (int)$redis->get('key1');
上述代码中的 get 命令，并没有真正执行，该语句实际只会返回一个 Redis 对象。
0x04 分布式锁的实现流程
基本思路是：
a.一个进程（客户端）去获取锁，如果可以获取到，则写入锁并且设置锁的有效期，当数据处理完之后，释放该锁；
b.当获取锁失败时，判断锁是否存在有效期，如果不存在，则设置锁的有效期，超出有效期后锁会自动释放
c.当数据处理完后，释放锁时，需要判断锁是否是其他进程（客户端）的锁，如果不是则释放，如果是则跳过
分布式锁需要注意的问题包括：
a. 防止持有锁的进程（客户端）意外崩溃，导致锁得不到释放，形成死锁，其他进程（客户端）一直得不到该锁；
b.防止持有锁的进程（客户端）因为操作时间过长（超过了锁的有效期）导致锁自动释放，最后到了该释放锁的时候却错误的释放了其他进程（客户端）的锁；
c.防止一个进程（客户端）的锁过期后，其他多个进程（客户端）同时尝试获取锁，并且都获取成功了
流程图：
0x05 Redis 实现分布式锁
要在高并发下消除竞争、保证数据一致性，可以采用 Redis 的分布式锁来实现。
有两种实现方式：
第一种是低于 2.6.12 版本的redis，需要使用 setnx、expire、ttl 等命令组合使用；
第二种是 2.6.12 版本起，redis 给 set 命令（https://redis.io/commands/set） 提供了更丰富的参数，来代替以上的几个命令
SET key value [EX seconds] [NX]
其中可选参数 EX seconds 表示键的过期时间为 seconds 秒，NX 表示只有键不存在时才对键进行设置，这个命令可以替代 setNx 命令加上 expire 命令，而且它是原子性的。
1.Redis Version < 2.6.12
例3.lock.php
<?php
/**
* redis 分布式锁
**/ class Lock
{
private $redis = ''; public function __construct($host, $port = 6379)
{
$this->redis = new Redis();
$this->redis->connect($host, $port);
} // 加锁
public function getLock($lockName, $timeout = 2)
{
$identifier = uniqid();
$timeout = ceil($timeout);
$end = time() + $timeout; while(time() < $end)
{
if($this->redis->setnx($lockName, $identifier))
{
$this->redis->expire($lockName, $timeout); return $identifier;
}
elseif($this->redis->ttl($lockName) == -1)
{
$this->redis->expire($lockName, $timeout);
}
usleep(0.001);
} return false;
} // 释放锁
public function releaseLock($lockName, $identifier)
{
if($this->redis->get($lockName) == $identifier)
{
$this->redis->multi();
$this->redis->del($lockName);
$this->redis->exec(); return true;
} return false;
} // test
public function test($lockName)
{
echo date('Y-m-d H:i:s', time()).' start'.PHP_EOL; for($i = 0; $i < 100000; $i++)
{
$identifier = $this->getLock($lockName);
if($identifier)
{
$count = $this->redis->get('count');
$count = intval($count) + 1;
$this->redis->set('count', $count);
$this->releaseLock($lockName, $identifier);
}
} echo date('Y-m-d H:i:s', time()).' end'.PHP_EOL;
}
} $obj = new Lock('localhost');
$obj->test('lock_name');
说明：代码参考 《Redis构建分布式锁》
客户端1：
客户端2：
结果：
2.Redis Version >= 2.6.12
例4.with_set.php
把上例中的
if($this->redis->setnx($lockName, $identifier))
{
$this->redis->expire($lockName, $timeout); return $identifier;
}
elseif($this->redis->ttl($lockName) == -1)
{
$this->redis->expire($lockName, $timeout);
}
替换为：
if($this->redis->set($lockName, $identifier, ['nx', 'ex'=>intval($timeout)]))
{
return $identifier;
}
即可。
以上使用 Redis 实现了分布式锁。