ThinkPHP5.1 如何使用 Redis 实现悲观锁

ThinkPHP 如何使用 Redis 实现悲观锁解决高并发情况下读写带来的脏读问题 / ThinkPHP5.1 / Redis Cache / File Cache 测试。

在用户量 / 客户端数量比较少的时候，只要系统的业务逻辑是正确的，一般都不会发现有什么问题。但随着用户量 / 客户端数量逐渐增多，高并发带来的问题就会逐渐出现，而脏读是众多问题的其中之一。

一、无并发控制，会带来什么问题？

本文以 ThinkPHP5.1.39 的代码作为案例，下面是一个 File Cache 读写操作：

public function fileCacheCase(){
    $keyName  = "test";
    $keyValue = 996;
 
    //写入缓存
    Cache::set($keyName, $keyValue, 3600);
 
    //从缓存中获取值
    $data = Cache::get($keyName);
 
    //删除缓存
    Cache::rm($keyName);
 
    echo "OK! $data";
}

访问这个 function，会输出 OK! 996 。无论你访问几次，结果都是如此，但仅限于单线程的情况（只有你自己一个人在访问这个 function），如果是多个人同时不停的访问这个 function，还会是这样吗？想一想 😛

使用 jmeter 测试一下，120 线程测试了十几秒，发现了 3 种不同的返回结果。

1.1、返回了 OK! 996

与单线程时的结果一致，是正常处理逻辑。

1.2、只返回了 OK！而不是 OK! 996

说明缓存不存在，原因是：在 A 线程将 996 写入缓存后，B 线程将缓存删除了。此时 A 线程从缓存中读出来的数据为 null，所以 A 线程输出了 OK! ，而不是 OK! 996。

1.3、返回了一个 500 错误

报错的内容是： file_get_contents（…）No such file or directory。

显然是 cache 文件夹下的某个缓存文件不存在，所以引起了这个错误。原因是：A 线程在删除缓存后，B 线程也在执行删除缓存的操作。当缓存文件已被删除时，再执行删除缓存文件的操作，自然就报了文件不存在的错误。（实测 120 个线程并发，总计 500 个请求，异常率 0.20%）

尽管我修改了 File Cache 的 133 行，在删除前判断文件是否存在，虽然异常率降低了，但依然无法从根本上解决问题。可以看到的是，在高并发场景下，问题已经显现出来了。

下面我们用 redis 缓存试试看：

public function fileCacheCase(){
    $keyName  = "test";
    $keyValue = 996;
 
    //写入缓存
    Cache::store('redis')->set($keyName, $keyValue, 3600);
 
    //从缓存中获取值
    $data = Cache::store('redis')->get($keyName);
 
    //删除缓存
    Cache::store('redis')->rm($keyName);
 
    echo "OK! $data";
}

经过测试，与上面的 3 种情况一致。（根据 thinkphp5.1 的官方文档，我使用的是 store 来切换到 redis，但不知道为何，仍然会报 File Cache 驱动的 No such file or directory/unlink 错误，十分诡异）。

如何解决高并发场景下带来的脏读问题？

答案是：使用锁机制。

二、关于锁机制

根据锁的控制范围，可分为单机锁 / 分布式锁 2 种。根据锁的实现思想，可分为悲观锁 / 乐观锁 2 种。

2.1、单机锁

即为单机环境的锁，无分布式设计。

常用的实现工具：

• Redis

• Memcached

2.2、分布式锁

为了防止分布式系统中的多个进程之间相互干扰，我们需要一种分布式协调技术来对这些进程进行调度。而这个分布式协调技术的核心就是来实现这个分布式锁。

• 在分布式系统环境下，一个方法在同一时间只能被一个机器的一个线程执行

• 高可用的获取锁与释放锁

• 高性能的获取锁与释放锁

• 具备锁失效机制，防止死锁

• 具备非阻塞锁特性，即没有获取到锁将直接返回获取锁失败

常用的实现工具：

• Zookeeper

• Redis

• Memcached

• Chubby

2.3、悲观锁

总是假设最坏的情况，每次去拿数据的时候都认为别人会修改，所以每次在拿数据的时候都会上锁，这样别人想拿这个数据就会阻塞直到它拿到锁（共享资源每次只给一个线程使用，其它线程阻塞，用完后再把资源转让给其它线程）。传统的关系型数据库里边就用到了很多这种锁机制，比如行锁，表锁等，读锁，写锁等，都是在做操作之前先上锁。Java 中 synchronized 和 ReentrantLock 等独占锁就是悲观锁思想的实现。

2.4、乐观锁

总是假设最好的情况，每次去拿数据的时候都认为别人不会修改，所以不会上锁，但是在更新的时候会判断一下在此期间别人有没有去更新这个数据，可以使用版本号机制和 CAS 算法实现。乐观锁适用于多读的应用类型，这样可以提高吞吐量，像数据库提供的类似于 write_condition 机制，其实都是提供的乐观锁。在 Java 中 java.util.concurrent.atomic 包下面的原子变量类就是使用了乐观锁的一种实现方式 CAS 实现的。

2.5、如何选择悲观 / 乐观锁？

从上面对两种锁的介绍，我们知道两种锁各有优缺点，不可认为一种好于另一种，像乐观锁适用于写比较少的情况下（多读场景），即冲突真的很少发生的时候，这样可以省去了锁的开销，加大了系统的整个吞吐量。但如果是多写的情况，一般会经常产生冲突，这就会导致上层应用会不断的进行 retry，这样反倒是降低了性能，所以一般多写的场景下用悲观锁就比较合适。

三、Redis 实现悲观锁

在商品秒杀活动活动中，流量峰值相对平常时的流量是高出非常多的。使用 Redis 实现悲观锁机制，可以解决商品库存脏读的问题。

初始化库存：

public function stockInit()
{
    $key       = "stock";
    $stockInit = 699;
 
    //清空所有缓存
    Cache::clear();
    Cache::store('redis')->clear();
 
    //写入库存初始值
    Cache::store('redis')->set($key, $stockInit);
 
    echo 'stock Init';
}

3.1、悲观锁实现（一）非最佳实践

看似符合逻辑的商品秒杀：

public function flashSale()
{
    $key        = "stock";
    $lockSuffix = "_lock";
 
    //判断库存锁是否存在
    while (Cache::get($key . $lockSuffix) == true) {
        // 存在锁定则等待
        usleep(200000);
    }
 
    //库存上锁
    Cache::store('redis')->set($key . $lockSuffix, 1, 30);
 
    //获取库存值
    $stock = Cache::store('redis')->get($key);
 
    //减库存
    if ($stock > 0) {
        $temp  = $stock;
        $stock -= 1;
    } else {
        //打开库存锁
        Cache::store('redis')->set($key . $lockSuffix, false);
        return "已售罄";
    }
    Cache::store('redis')->set($key, $stock);
 
    //打开库存锁
    Cache::store('redis')->set($key . $lockSuffix, false);
 
    return "恭喜，您抢到了第 {$temp}个库存！";
}

实测 150 线程并发，异常率 0%，虽然引用了锁机制，看似符合逻辑的锁机制，但仍会有极低的概率脏读，原因无他，有 N 个线程同时抢到了锁。虽然概率低，但线程一多仍然会脏读。所以需要改用 redis 原生支持的 setnx 来保证只有一个线程抢到了锁。

如下，两个线程同时抢到了第 80 个库存：

3.2、悲观锁实现（二）

setnx 是 set if not exists 的简写，在 key 不存在时等价于 set，如果 key 存在，则不更新缓存内容，且返回 false。使用这个特性，可以保证锁只有一个线程抢到了。

使用 redis setnx 实现悲观锁的商品秒杀：

public function flashSale()
{
    $redisConifg = config('cache.redis');                  //获取当前模块下的config文件夹中的cache文件的redis配置数组
    $redis       = Cache::connect($redisConifg);           //获取thinkPHP官方封装的Redis Cache对象
    $handler     = Cache::connect($redisConifg)->handler();//获取php redis扩展原生redis对象 https://github.com/phpredis/phpredis
 
    $key        = "stock";//商品库存缓存名
    $lockSuffix = "_lock";//商品库存锁后缀名
    $timeOut    = 10;     //库存锁过期时间
 
    //抢库存锁
    while ($handler->set($key . $lockSuffix, 1, ['nx', 'ex' => $timeOut]) == false) {
        // 没有抢到则等待
        usleep(20000);
    }
 
    //当前线程抢到库存锁了
 
    //获取库存值
    $stock = $redis->get($key);
 
    //减库存
    if ($stock > 0) {
        $temp  = $stock;
        $stock -= 1;
    } else {
        //删除库存锁
        $redis->rm($key . $lockSuffix);
        return "已售罄";
    }
 
    //更新库存值
    $redis->set($key, $stock);
 
    //删除库存锁
    $redis->rm($key . $lockSuffix);
 
    return "恭喜，您抢到了第 {$temp}个库存！";
}

150 线程并发测试后，并没有发现有异常情况了。根据实际业务需求，可以增加等待超时机制。

REF

https://redis.io/commands/set

http://www.redis.cn/commands/set.html

https://github.com/phpredis/phpredis#set

https://www.jianshu.com/p/a1ebab8ce78a

https://blog.csdn.net/qq_34337272/article/details/81072874