带你一步步用php实现redis分布式、高并发库存问题
开始正文, 有任何疑问都可以在评论区留言,以laravel5.8框架为基础来编写业务逻辑。
普通减库存(使用redis简单模拟减库存操作)
<?php
use \Illuminate\Support\Facades\Redis;
$redis = Redis::connection(); //步骤1: redis实例
$stockKey = 'stock'; //步骤2: 库存key
//$redis->set($stockKey, 50); //步骤3: 模拟初始化库存50
$stock = $redis->get('stock'); //步骤4: 获取库存值
if ($stock > 0) { //步骤5: 库存大于0
$stock = $stock - 1; //步骤6: 减库存
$redis->set('lock', $stock); //步骤7: 重新设置到缓存
echo true; //步骤8: 减库存成功返回true
} else {
echo false; //步骤9: 减库存失败返false
}
?>并发用户在同一时间点到达步骤4(获取库存值)得到同一库存值并进行库存减一操作即会引起超卖现象。
加锁
用setnx命令,给当前活动加一把锁(value的话，这里的话，我们暂且设置为1)。
<?php
use \Illuminate\Support\Facades\Redis;
$redis = Redis::connection(); //步骤1: redis实例
$lockKey = 'lockKey'; //步骤2: 线程锁键key
$isLock = $redis->setnx($lockKey, 1); //步骤3: 加锁
if (!$isLock) { //步骤4:未获得锁的线程(用户)直接返回,稍后再试~
return '服务器繁忙,请稍后再试~';
}
$stockKey = 'stock'; //步骤5: 库存key
//$redis->set($stockKey, 50); //步骤6: 模拟初始化库存50
$stock = $redis->get('stock'); //步骤7: 获取库存值
if ($stock > 0) { //步骤8: 库存大于0
$stock = $stock - 1; //步骤9: 减库存
$redis->set('lock', $stock); //步骤10: 重新设置到缓存
echo true; //步骤11: 减库存成功返回true
} else {
echo false; //步骤12: 减库存失败返false
}
$redis->del($isLock); //步骤13: 删除当前锁
?>
步骤3加锁如果执行setnx返回1，说明lockKey不存在，获取锁成功；当返回结果为0，说明lockKey已经存在，获取锁失败。
如果一个拿到锁的线程，在执行任务的过程中挂掉了，来不及显示的释放锁，则会一直占用着资源，导致其他线程无法拿到锁， 没法执行任务。所以在执行setnx命令之后，需要给锁显示设置一个锁超时时间，以保证即使拿到锁的线程挂掉了，也能在超过一定时间自动释放锁，让出资源。而setnx不支持设置超时参数，所以需要其他命令来执行。
如果执行完setnx之后，节点1宕机了，还没来得及执行expire命令：(即步骤3-4过程中加锁时设置一个过期时间,但是两个 程序依然不是原子块执行,步骤3直接宕机依然存在以上问题)，这时候我们就需要添加异常捕获优先删除锁try{}finally{},redis 从2.6.12版本开始，redis为SET命令可以保证加锁和设置一个过期时间在一个原子块内操作。
设置锁超时并且添加异常捕获优先删除锁
<?php
use \Illuminate\Support\Facades\Redis;
$redis = Redis::connection(); //步骤1: redis实例
$lockKey = 'lockKey'; //步骤2: 线程锁键key
$isLock = $redis->setnx($lockKey, 1); //步骤3: 加锁
$redis->expire($lockKey, 10); //步骤4: 给锁设置超时时间
//2.6.12版本可用,如版本低于2.6.12请使用lua脚本执行保证原子性操作
$isLock = $redis->set($lockKey, 1, 'ex', 10, 'nx');
if (!$isLock) { //步骤5:未获得锁的线程(用户)直接返回,稍后再试~
return '服务器繁忙,请稍后再试~';
}
try{
$stockKey = 'stock'; //步骤6: 库存键key
//$redis->set($stockKey, 50); //步骤7: 模拟初始化库存50
$stock = $redis->get('stock'); //步骤8: 获取库存值
if ($stock > 0) { //步骤9: 库存大于0
$stock = $stock - 1; //步骤10: 减库存
$redis->set('lock', $stock); //步骤11: 重新设置到缓存
echo true; //步骤12: 减库存成功返回true
} else {
echo false; //步骤13: 减库存成功返false
}
}finally{
$redis->del($isLock); //步骤14: 删除当前锁
}
?>又是一个极端场景，假设节点1的线程A通过set拿到了锁，并设置了过期时间30秒。
由于某些原因，导致线程A执行的很慢，超时时间30秒过去了，但线程A还没执行完，这个时候锁自动释放，线程B得到了锁。
随后，线程A任务执行完，进行del操作释放锁，这个时候线程B还没执行完，线程A实际上删除的是线程B加的锁。
如何解决这个问题呢？
每个线程在set操作的时候，可以给value设置一个唯一的值，然后在del释放锁之前加一个判断，验证当前的锁是不是自身加的锁。
<?php
use \Illuminate\Support\Facades\Redis;
use \Godruoyi\Snowflake\Snowflake;
$redis = Redis::connection(); //步骤1: redis实例
$datacenterId = 123124354; //指定数据中心ID
$workerId = 1122435; //计算机ID
$uuid = new Snowflake($datacenterId, $workerId); //步骤2:分布式生成唯一uuid(https://github.com/godruoyi/php-snowflake)
//$uuid = session_create_id()
$lockKey = 'lockKey'; //步骤3: 线程锁键key
$isLock = $redis->set($lockKey, $uuid, 'ex', 10, 'nx'); //步骤4: 加锁并设置超时时间,设置值为uuid
if (!$isLock) { //步骤5:未获得锁的线程(用户)直接返回,稍后再试~
return '服务器繁忙,请稍后再试~';
}
try {
$stockKey = 'stock'; //步骤6: 库存键key
//$redis->set($stockKey, 50); //步骤7: 第一次运行,初始化库存(注意：初次执行)
$stock = $redis->get('stock'); //步骤8: 获取库存值
if ($stock > 0) { //步骤9: 库存大于0
$stock = $stock - 1; //步骤10: 减库存
$redis->set('lock', $stock); //步骤11: 重新设置到缓存
echo true; //步骤12: 减库存成功返回true
} else {
echo false; //步骤13: 减库存成功返false
}
} finally {
//这一步不是原子性操作，还是会有问题，我们用lua原子性去处理
if ($uuid === $redis->get($lockKey)) { //步骤14: 保证用户删除的是自己的锁
$redis->del($lockKey); //步骤15: 删除当前锁
}
//lua原子性去处理如下
$script = <<<EOF
local key = KEYS[1]
local value = ARGV[1]
if (redis.call('exists', key) == 1 and redis.call('get', key) == value)
then
return redis.call('del', key)
end
return 0
EOF;
$redis->eval($script, [$lockKey,$uuid]);
}
?>
依然存在get和del非原子性操作(步骤14和步骤15)，需要通过lua脚本进行原子性处理。
<?php
class RedisLock
{
/**
* @var 当前锁标识，用于解锁
*/
private $_lockFlag;
private $_redis;
public function __construct($host = '127.0.0.1', $port = '6379', $passwd = '')
{
$this->_redis = new Redis();
$this->_redis->connect($host, $port);
if ($passwd) {
$this->_redis->auth($passwd);
}
}
public function lock($key, $expire = 5)
{
$now= time();
$expireTime = $expire + $now;
if ($this->_redis->setnx($key, $expireTime)) {
$this->_lockFlag = $expireTime;
return true;
}
// 获取上一个锁的到期时间
$currentLockTime = $this->_redis->get($key);
if ($currentLockTime < $now) {
/* 用于解决
C0超时了,还持有锁,加入C1/C2/...同时请求进入了方法里面
C1/C2都执行了getset方法(由于getset方法的原子性,
所以两个请求返回的值必定不相等保证了C1/C2只有一个获取了锁) */
$oldLockTime = $this->_redis->getset($key, $expireTime);
if ($currentLockTime == $oldLockTime) {
$this->_lockFlag = $expireTime;
return true;
}
}
return false;
}
public function lockByLua($key, $expire = 5)
{
$script = <<<EOF
local key = KEYS[1]
local value = ARGV[1]
local ttl = ARGV[2]
if (redis.call('setnx', key, value) == 1) then
return redis.call('expire', key, ttl)
elseif (redis.call('ttl', key) == -1) then
return redis.call('expire', key, ttl)
end
return 0
EOF;
$this->_lockFlag = md5(microtime(true));
return $this->_eval($script, [$key, $this->_lockFlag, $expire]);
}
public function unlock($key)
{
$script = <<<EOF
local key = KEYS[1]
local value = ARGV[1]
if (redis.call('exists', key) == 1 and redis.call('get', key) == value)
then
return redis.call('del', key)
end
return 0
EOF;
if ($this->_lockFlag) {
return $this->_eval($script, [$key, $this->_lockFlag]);
}
}
private function _eval($script, array $params, $keyNum = 1)
{
$hash = $this->_redis->script('load', $script);
return $this->_redis->evalSha($hash, $params, $keyNum);
}
}
$redisLock = new RedisLock();
$key = 'lock';
if ($redisLock->lockByLua($key)) {
// to do...
$redisLock->unlock($key);
}
?>
目前并发情况下还有一些问题，当某个进程执行时间大于锁过期时间，进行延时。
可以在加锁的时候开一个子进程去监控 主进程是否完成，未完成则给主进程延时，目前未实现代码。
本文参与 腾讯云自媒体同步曝光计划，分享自作者个人站点/博客。
如有侵权请联系 cloudcommunity@tencent.com 删除
评论
登录后参与评论
推荐阅读
目录
相关产品与服务
云数据库 Redis
腾讯云数据库 Redis（TencentDB for Redis）是腾讯云打造的兼容 Redis 协议的缓存和存储服务。丰富的数据结构能帮助您完成不同类型的业务场景开发。支持主从热备，提供自动容灾切换、数据备份、故障迁移、实例监控、在线扩容、数据回档等全套的数据库服务。
腾讯云开发者
