PHP Redis 분산 잠금 및 작업 대기열 코드 예제에 대한 자세한 설명
- 伊谢尔伦원래의
- 2017-07-17 09:42:592450검색
1. 분산 잠금 아이디어의 Redis 구현
아이디어는 매우 간단합니다. 사용되는 주요 Redis 기능은 분산 잠금을 구현하는 데 가장 중요한 기능이어야 합니다. 첫 번째는 특정 작업 식별 이름(여기서는 식별 이름의 예로 Lock:order가 사용됨)을 Redis의 키로 저장하고 다른 Lock:order 요청이 있는 경우 해당 작업에 대한 만료 시간을 설정하는 것입니다. setnx() Lock:order를 redis에 삽입할 수 있는지 확인합니다. 가능하면 true를 반환하고, 그렇지 않으면 false를 반환합니다. 물론 내 코드는 이 아이디어보다 더 복잡할 것이므로 코드를 분석할 때 더 자세히 설명하겠습니다.
2. Redis는 작업 대기열을 구현합니다.
여기서 구현에서는 주로 Redis의 정렬된 집합 데이터 구조를 사용하여 위의 Redis 분산 잠금 메커니즘을 사용합니다. 예를 들어, 대기열에 합류할 때는 zset의 add() 함수를 사용하여 대기열에 합류하고, 대기열에서 나갈 때는 zset의 getScore() 함수를 사용할 수 있습니다. 또한 상단에 여러 작업이 팝업될 수 있습니다.
3. 코드 분석
(1) 먼저 Redis 분산 잠금의 코드 구현을 분석해 보겠습니다
(1) 특별한 이유로 잠금이 해제되지 않는 것을 방지하기 위해 잠금이 성공적으로 추가되면 잠금은 생존 시간을 제공합니다(잠금 방법의 매개변수 설정을 통해 또는 기본값 사용). 잠금의 기본 생존 시간은 다음과 같습니다. 상대적으로 짧습니다(초). 따라서 오랜 시간 동안 잠가야 하는 경우에는 루프 내에서와 같이 잠금의 수명을 적절한 시간으로 연장하는 만료 메소드를 전달할 수 있습니다.
(2) 시스템 수준 잠금. 어떤 이유로든 프로세스가 충돌하면 운영 체제가 자체적으로 잠금을 재활용하므로 리소스 손실이 없습니다. 그러나 분산 잠금이 설정된 경우에는 사용되지 않습니다. 여러 가지 이유로 프로세스 충돌이나 기타 예외로 인해 잠금 해제가 호출되지 않으면 남은 시간 동안 잠금이 가비지 잠금이 되어 다른 프로세스나 프로세스가 잠금 상태에 들어갈 수 없게 됩니다. 다시 시작한 후 영역.
먼저 잠금 구현 코드를 살펴보겠습니다. 여기에는 두 개의 주요 매개변수가 필요합니다. 그 중 하나는 $timeout입니다. 이는 주기적으로 잠금을 획득하기 위한 대기 시간입니다. 이 시간 동안 시간이 초과될 때까지 잠금 획득을 계속 시도합니다. 0이면 획득을 의미합니다. 잠금이 실패한 후 기다리지 않고 바로 반환됩니다. 또 다른 중요한 매개변수는 $expire입니다. 이 매개변수는 현재 잠금의 최대 생존 시간을 나타내며 0보다 커야 합니다. 생존 시간 이후에도 잠금이 해제되지 않으면 시스템이 자동으로 강제 해제됩니다. 이 매개변수의 가장 중요한 기능에 대해서는 위 (1)의 설명을 참조하십시오.
여기서 먼저 현재 시간을 얻은 다음 잠금이 실패할 때 대기 시간 초과(타임스탬프)를 얻은 다음 잠금의 최대 생존 시간을 얻습니다. 여기서 redis의 키는 "잠금: 잠금의 식별 이름" 형식을 사용합니다. 먼저, redis에 데이터를 삽입하고 setnx() 함수를 사용합니다. 존재하는 경우 데이터를 삽입하고 최대 생존 시간을 값으로 저장합니다. 삽입에 성공하면 키의 만료 시간을 설정하고 키를 $lockedName 배열에 넣습니다. 이는 잠금이 성공했음을 의미합니다. 키가 존재하는 경우 여기에는 현재 키의 남은 시간을 가져오는 엄격한 작업이 없습니다. 이 시간이 0보다 작으면 생존 시간이 설정되지 않았음을 의미합니다. key (이전 setnx가 자동으로 생성되므로 해당 키는 존재하지 않습니다.) 이 경우 setnx 성공 후 프로세스의 인스턴스가 충돌하여 후속 만료가 호출되지 않게 됩니다. 이 때 직접 설정할 수 있습니다. 만료된 후 자신의 용도로 자물쇠를 사용하십시오. 잠금 실패에 대한 대기 시간이 설정되지 않았거나 최대 대기 시간을 초과한 경우 루프를 종료합니다. 그렇지 않으면 $waitIntervalUs 이후에 요청을 계속합니다. 잠금에 대한 전체 코드 분석입니다. /**
* 加锁
* @param [type] $name 锁的标识名
* @param integer $timeout 循环获取锁的等待超时时间，在此时间内会一直尝试获取锁直到超时，为0表示失败后直接返回不等待
* @param integer $expire 当前锁的最大生存时间(秒)，必须大于0，如果超过生存时间锁仍未被释放，则系统会自动强制释放
* @param integer $waitIntervalUs 获取锁失败后挂起再试的时间间隔(微秒)
* @return [type] [description]
*/
public function lock($name, $timeout = 0, $expire = 15, $waitIntervalUs = 100000) {
if ($name == null) return false;
//取得当前时间
$now = time();
//获取锁失败时的等待超时时刻
$timeoutAt = $now + $timeout;
//锁的最大生存时刻
$expireAt = $now + $expire;
$redisKey = "Lock:{$name}";
while (true) {
//将rediskey的最大生存时刻存到redis里，过了这个时刻该锁会被自动释放
$result = $this->redisString->setnx($redisKey, $expireAt);
if ($result != false) {
//设置key的失效时间
$this->redisString->expire($redisKey, $expireAt);
//将锁标志放到lockedNames数组里
$this->lockedNames[$name] = $expireAt;
return true;
}
//以秒为单位，返回给定key的剩余生存时间
$ttl = $this->redisString->ttl($redisKey);
//ttl小于0 表示key上没有设置生存时间（key是不会不存在的，因为前面setnx会自动创建）
//如果出现这种状况，那就是进程的某个实例setnx成功后 crash 导致紧跟着的expire没有被调用
//这时可以直接设置expire并把锁纳为己用
if ($ttl < 0) {
$this->redisString->set($redisKey, $expireAt);
$this->lockedNames[$name] = $expireAt;
return true;
}
/*****循环请求锁部分*****/
//如果没设置锁失败的等待时间 或者 已超过最大等待时间了，那就退出
if ($timeout <= 0 || $timeoutAt < microtime(true)) break;
//隔 $waitIntervalUs 后继续 请求
usleep($waitIntervalUs);
}
return false;
} 다음으로 잠금 해제 코드 분석을 살펴보세요. 잠금 해제가 훨씬 간단합니다. 먼저 잠금이 존재하는지 확인하고 deleteKey() 함수를 통해 잠금 ID를 삭제하세요. .
/**
* 解锁
* @param [type] $name [description]
* @return [type] [description]
*/
public function unlock($name) {
//先判断是否存在此锁
if ($this->isLocking($name)) {
//删除锁
if ($this->redisString->deleteKey("Lock:$name")) {
//清掉lockedNames里的锁标志
unset($this->lockedNames[$name]);
return true;
}
}
return false;
}
在贴上删除掉所有锁的方法，其实都一个样，多了个循环遍历而已。
/**
* 释放当前所有获得的锁
* @return [type] [description]
*/
public function unlockAll() {
//此标志是用来标志是否释放所有锁成功
$allSuccess = true;
foreach ($this->lockedNames as $name => $expireAt) {
if (false === $this->unlock($name)) {
$allSuccess = false;
}
}
return $allSuccess;
}위 내용은 Redis를 사용한 분산 잠금의 전체 아이디어 및 코드 구현을 요약하고 공유한 것입니다. 여기에는 기본적으로 모든 사람이 쉽게 이해할 수 있도록 코드의 각 줄에 주석을 달았습니다. 애플리케이션을 시뮬레이션할 수 있습니다. 더 자세히 알고 싶다면 전체 수업의 코드를 참고하세요:
/**
*在redis上实现分布式锁
*/
class RedisLock {
private $redisString;
private $lockedNames = [];
public function construct($param = NULL) {
$this->redisString = RedisFactory::get($param)->string;
}
/**
* 加锁
* @param [type] $name 锁的标识名
* @param integer $timeout 循环获取锁的等待超时时间，在此时间内会一直尝试获取锁直到超时，为0表示失败后直接返回不等待
* @param integer $expire 当前锁的最大生存时间(秒)，必须大于0，如果超过生存时间锁仍未被释放，则系统会自动强制释放
* @param integer $waitIntervalUs 获取锁失败后挂起再试的时间间隔(微秒)
* @return [type] [description]
*/
public function lock($name, $timeout = 0, $expire = 15, $waitIntervalUs = 100000) {
if ($name == null) return false;
//取得当前时间
$now = time();
//获取锁失败时的等待超时时刻
$timeoutAt = $now + $timeout;
//锁的最大生存时刻
$expireAt = $now + $expire;
$redisKey = "Lock:{$name}";
while (true) {
//将rediskey的最大生存时刻存到redis里，过了这个时刻该锁会被自动释放
$result = $this->redisString->setnx($redisKey, $expireAt);
if ($result != false) {
//设置key的失效时间
$this->redisString->expire($redisKey, $expireAt);
//将锁标志放到lockedNames数组里
$this->lockedNames[$name] = $expireAt;
return true;
}
//以秒为单位，返回给定key的剩余生存时间
$ttl = $this->redisString->ttl($redisKey);
//ttl小于0 表示key上没有设置生存时间（key是不会不存在的，因为前面setnx会自动创建）
//如果出现这种状况，那就是进程的某个实例setnx成功后 crash 导致紧跟着的expire没有被调用
//这时可以直接设置expire并把锁纳为己用
if ($ttl < 0) {
$this->redisString->set($redisKey, $expireAt);
$this->lockedNames[$name] = $expireAt;
return true;
}
/*****循环请求锁部分*****/
//如果没设置锁失败的等待时间 或者 已超过最大等待时间了，那就退出
if ($timeout <= 0 || $timeoutAt < microtime(true)) break;
//隔 $waitIntervalUs 后继续 请求
usleep($waitIntervalUs);
}
return false;
}
/**
* 解锁
* @param [type] $name [description]
* @return [type] [description]
*/
public function unlock($name) {
//先判断是否存在此锁
if ($this->isLocking($name)) {
//删除锁
if ($this->redisString->deleteKey("Lock:$name")) {
//清掉lockedNames里的锁标志
unset($this->lockedNames[$name]);
return true;
}
}
return false;
}
/**
* 释放当前所有获得的锁
* @return [type] [description]
*/
public function unlockAll() {
//此标志是用来标志是否释放所有锁成功
$allSuccess = true;
foreach ($this->lockedNames as $name => $expireAt) {
if (false === $this->unlock($name)) {
$allSuccess = false;
}
}
return $allSuccess;
}
/**
* 给当前所增加指定生存时间，必须大于0
* @param [type] $name [description]
* @return [type] [description]
*/
public function expire($name, $expire) {
//先判断是否存在该锁
if ($this->isLocking($name)) {
//所指定的生存时间必须大于0
$expire = max($expire, 1);
//增加锁生存时间
if ($this->redisString->expire("Lock:$name", $expire)) {
return true;
}
}
return false;
}
/**
* 判断当前是否拥有指定名字的所
* @param [type] $name [description]
* @return boolean [description]
*/
public function isLocking($name) {
//先看lonkedName[$name]是否存在该锁标志名
if (isset($this->lockedNames[$name])) {
//从redis返回该锁的生存时间
return (string)$this->lockedNames[$name] = (string)$this->redisString->get("Lock:$name");
}
return false;
}
}
(2) Redis를 이용한 작업 대기열 구현 코드 분석 (1) 작업 대기열을 배치하는 데 사용되는 작업 대기열 비즈니스 로직에서 비동기적으로 처리할 수 있는 작업입니다. 대기열에 들어가서 다른 스레드에서 처리한 다음 대기열에서 빼냅니다
(2) 대기열에 넣기와 대기열에서 빼기의 일관성을 보장하기 위해 대기열에서 분산 잠금 및 기타 논리가 사용됩니다.
（3）这个队列和普通队列不一样，入队时的id是用来区分重复入队的，队列里面只会有一条记录，同一个id后入的覆盖前入的，而不是追加， 如果需求要求重复入队当做不用的任务，请使用不同的id区分
先看入队的代码分析：首先当然是对参数的合法性检测，接着就用到上面加锁机制的内容了，就是开始加锁，入队时我这里选择当前时间戳作为score，接着就是入队了，使用的是zset数据结构的add()方法，入队完成后，就对该任务解锁，即完成了一个入队的操作。
/**
* 入队一个 Task
* @param [type] $name 队列名称
* @param [type] $id 任务id（或者其数组）
* @param integer $timeout 入队超时时间(秒)
* @param integer $afterInterval [description]
* @return [type] [description]
*/
public function enqueue($name, $id, $timeout = 10, $afterInterval = 0) {
//合法性检测
if (empty($name) || empty($id) || $timeout <= 0) return false;
//加锁
if (!$this->_redis->lock->lock("Queue:{$name}", $timeout)) {
Logger::get('queue')->error("enqueue faild becouse of lock failure: name = $name, id = $id");
return false;
}
//入队时以当前时间戳作为 score
$score = microtime(true) + $afterInterval;
//入队
foreach ((array)$id as $item) {
//先判断下是否已经存在该id了
if (false === $this->_redis->zset->getScore("Queue:$name", $item)) {
$this->_redis->zset->add("Queue:$name", $score, $item);
}
}
//解锁
$this->_redis->lock->unlock("Queue:$name");
return true;
}接着来看一下出队的代码分析：出队一个Task，需要指定它的$id 和 $score，如果$score与队列中的匹配则出队，否则认为该Task已被重新入队过，当前操作按失败处理。首先和对参数进行合法性检测，接着又用到加锁的功能了，然后及时出队了，先使用getScore()从Redis里获取到该id的score，然后将传入的$score和Redis里存储的score进行对比，如果两者相等就进行出队操作，也就是使用zset里的delete()方法删掉该任务id，最后当前就是解锁了。这就是出队的代码分析。
/**
* 出队一个Task，需要指定$id 和 $score
* 如果$score 与队列中的匹配则出队，否则认为该Task已被重新入队过，当前操作按失败处理
*
* @param [type] $name 队列名称
* @param [type] $id 任务标识
* @param [type] $score 任务对应score，从队列中获取任务时会返回一个score，只有$score和队列中的值匹配时Task才会被出队
* @param integer $timeout 超时时间(秒)
* @return [type] Task是否成功，返回false可能是redis操作失败，也有可能是$score与队列中的值不匹配（这表示该Task自从获取到本地之后被其他线程入队过）
*/
public function dequeue($name, $id, $score, $timeout = 10) {
//合法性检测
if (empty($name) || empty($id) || empty($score)) return false;
//加锁
if (!$this->_redis->lock->lock("Queue:$name", $timeout)) {
Logger:get('queue')->error("dequeue faild becouse of lock lailure:name=$name, id = $id");
return false;
}
//出队
//先取出redis的score
$serverScore = $this->_redis->zset->getScore("Queue:$name", $id);
$result = false;
//先判断传进来的score和redis的score是否是一样
if ($serverScore == $score) {
//删掉该$id
$result = (float)$this->_redis->zset->delete("Queue:$name", $id);
if ($result == false) {
Logger::get('queue')->error("dequeue faild because of redis delete failure: name =$name, id = $id");
}
}
//解锁
$this->_redis->lock->unlock("Queue:$name");
return $result;
}学过数据结构这门课的朋友都应该知道，队列操作还有弹出顶部某个值的方法等等，这里处理入队出队操作
/**
* 获取队列顶部若干个Task 并将其出队
* @param [type] $name 队列名称
* @param integer $count 数量
* @param integer $timeout 超时时间
* @return [type] 返回数组[0=>['id'=> , 'score'=> ], 1=>['id'=> , 'score'=> ], 2=>['id'=> , 'score'=> ]]
*/
public function pop($name, $count = 1, $timeout = 10) {
//合法性检测
if (empty($name) || $count <= 0) return [];
//加锁
if (!$this->_redis->lock->lock("Queue:$name")) {
Log::get('queue')->error("pop faild because of pop failure: name = $name, count = $count");
return false;
}
//取出若干的Task
$result = [];
$array = $this->_redis->zset->getByScore("Queue:$name", false, microtime(true), true, false, [0, $count]);
//将其放在$result数组里 并 删除掉redis对应的id
foreach ($array as $id => $score) {
$result[] = ['id'=>$id, 'score'=>$score];
$this->_redis->zset->delete("Queue:$name", $id);
}
//解锁
$this->_redis->lock->unlock("Queue:$name");
return $count == 1 ? (empty($result) ? false : $result[0]) : $result;
}以上就是用Redis实现任务队列的整一套思路和代码实现的总结和分享
/**
* 任务队列
*
*/
class RedisQueue {
private $_redis;
public function construct($param = null) {
$this->_redis = RedisFactory::get($param);
}
/**
* 入队一个 Task
* @param [type] $name 队列名称
* @param [type] $id 任务id（或者其数组）
* @param integer $timeout 入队超时时间(秒)
* @param integer $afterInterval [description]
* @return [type] [description]
*/
public function enqueue($name, $id, $timeout = 10, $afterInterval = 0) {
//合法性检测
if (empty($name) || empty($id) || $timeout <= 0) return false;
//加锁
if (!$this->_redis->lock->lock("Queue:{$name}", $timeout)) {
Logger::get('queue')->error("enqueue faild becouse of lock failure: name = $name, id = $id");
return false;
}
//入队时以当前时间戳作为 score
$score = microtime(true) + $afterInterval;
//入队
foreach ((array)$id as $item) {
//先判断下是否已经存在该id了
if (false === $this->_redis->zset->getScore("Queue:$name", $item)) {
$this->_redis->zset->add("Queue:$name", $score, $item);
}
}
//解锁
$this->_redis->lock->unlock("Queue:$name");
return true;
}
/**
* 出队一个Task，需要指定$id 和 $score
* 如果$score 与队列中的匹配则出队，否则认为该Task已被重新入队过，当前操作按失败处理
*
* @param [type] $name 队列名称
* @param [type] $id 任务标识
* @param [type] $score 任务对应score，从队列中获取任务时会返回一个score，只有$score和队列中的值匹配时Task才会被出队
* @param integer $timeout 超时时间(秒)
* @return [type] Task是否成功，返回false可能是redis操作失败，也有可能是$score与队列中的值不匹配（这表示该Task自从获取到本地之后被其他线程入队过）
*/
public function dequeue($name, $id, $score, $timeout = 10) {
//合法性检测
if (empty($name) || empty($id) || empty($score)) return false;
//加锁
if (!$this->_redis->lock->lock("Queue:$name", $timeout)) {
Logger:get('queue')->error("dequeue faild becouse of lock lailure:name=$name, id = $id");
return false;
}
//出队
//先取出redis的score
$serverScore = $this->_redis->zset->getScore("Queue:$name", $id);
$result = false;
//先判断传进来的score和redis的score是否是一样
if ($serverScore == $score) {
//删掉该$id
$result = (float)$this->_redis->zset->delete("Queue:$name", $id);
if ($result == false) {
Logger::get('queue')->error("dequeue faild because of redis delete failure: name =$name, id = $id");
}
}
//解锁
$this->_redis->lock->unlock("Queue:$name");
return $result;
}
/**
* 获取队列顶部若干个Task 并将其出队
* @param [type] $name 队列名称
* @param integer $count 数量
* @param integer $timeout 超时时间
* @return [type] 返回数组[0=>['id'=> , 'score'=> ], 1=>['id'=> , 'score'=> ], 2=>['id'=> , 'score'=> ]]
*/
public function pop($name, $count = 1, $timeout = 10) {
//合法性检测
if (empty($name) || $count <= 0) return [];
//加锁
if (!$this->_redis->lock->lock("Queue:$name")) {
Logger::get('queue')->error("pop faild because of pop failure: name = $name, count = $count");
return false;
}
//取出若干的Task
$result = [];
$array = $this->_redis->zset->getByScore("Queue:$name", false, microtime(true), true, false, [0, $count]);
//将其放在$result数组里 并 删除掉redis对应的id
foreach ($array as $id => $score) {
$result[] = ['id'=>$id, 'score'=>$score];
$this->_redis->zset->delete("Queue:$name", $id);
}
//解锁
$this->_redis->lock->unlock("Queue:$name");
return $count == 1 ? (empty($result) ? false : $result[0]) : $result;
}
/**
* 获取队列顶部的若干个Task
* @param [type] $name 队列名称
* @param integer $count 数量
* @return [type] 返回数组[0=>['id'=> , 'score'=> ], 1=>['id'=> , 'score'=> ], 2=>['id'=> , 'score'=> ]]
*/
public function top($name, $count = 1) {
//合法性检测
if (empty($name) || $count < 1) return [];
//取错若干个Task
$result = [];
$array = $this->_redis->zset->getByScore("Queue:$name", false, microtime(true), true, false, [0, $count]);
//将Task存放在数组里
foreach ($array as $id => $score) {
$result[] = ['id'=>$id, 'score'=>$score];
}
//返回数组
return $count == 1 ? (empty($result) ? false : $result[0]) : $result;
}
}到此，这两大块功能基本讲解完毕，对于任务队列，你可以写一个shell脚本，让服务器定时运行某些程序，实现入队出队等操作，这里我就不在将其与实际应用结合起来去实现了，大家理解好这两大功能的实现思路即可，由于代码用的是PHP语言来写的，如果你理解了实现思路，你完全可以使用java或者是.net等等其他语言去实现这两个功能。这两大功能的应用场景十分多，特别是秒杀，另一个就是春运抢火车票，这两个是最鲜明的例子了。当然还有很多地方用到，这里我不再一一列举。
附上分布式锁和任务队列这两个类：
/**
*在redis上实现分布式锁
*/
class RedisLock {
private $redisString;
private $lockedNames = [];
public function construct($param = NULL) {
$this->redisString = RedisFactory::get($param)->string;
}
/**
* 加锁
* @param [type] $name 锁的标识名
* @param integer $timeout 循环获取锁的等待超时时间，在此时间内会一直尝试获取锁直到超时，为0表示失败后直接返回不等待
* @param integer $expire 当前锁的最大生存时间(秒)，必须大于0，如果超过生存时间锁仍未被释放，则系统会自动强制释放
* @param integer $waitIntervalUs 获取锁失败后挂起再试的时间间隔(微秒)
* @return [type] [description]
*/
public function lock($name, $timeout = 0, $expire = 15, $waitIntervalUs = 100000) {
if ($name == null) return false;
//取得当前时间
$now = time();
//获取锁失败时的等待超时时刻
$timeoutAt = $now + $timeout;
//锁的最大生存时刻
$expireAt = $now + $expire;
$redisKey = "Lock:{$name}";
while (true) {
//将rediskey的最大生存时刻存到redis里，过了这个时刻该锁会被自动释放
$result = $this->redisString->setnx($redisKey, $expireAt);
if ($result != false) {
//设置key的失效时间
$this->redisString->expire($redisKey, $expireAt);
//将锁标志放到lockedNames数组里
$this->lockedNames[$name] = $expireAt;
return true;
}
//以秒为单位，返回给定key的剩余生存时间
$ttl = $this->redisString->ttl($redisKey);
//ttl小于0 表示key上没有设置生存时间（key是不会不存在的，因为前面setnx会自动创建）
//如果出现这种状况，那就是进程的某个实例setnx成功后 crash 导致紧跟着的expire没有被调用
//这时可以直接设置expire并把锁纳为己用
if ($ttl < 0) {
$this->redisString->set($redisKey, $expireAt);
$this->lockedNames[$name] = $expireAt;
return true;
}
/*****循环请求锁部分*****/
//如果没设置锁失败的等待时间 或者 已超过最大等待时间了，那就退出
if ($timeout <= 0 || $timeoutAt < microtime(true)) break;
//隔 $waitIntervalUs 后继续 请求
usleep($waitIntervalUs);
}
return false;
}
/**
* 解锁
* @param [type] $name [description]
* @return [type] [description]
*/
public function unlock($name) {
//先判断是否存在此锁
if ($this->isLocking($name)) {
//删除锁
if ($this->redisString->deleteKey("Lock:$name")) {
//清掉lockedNames里的锁标志
unset($this->lockedNames[$name]);
return true;
}
}
return false;
}
/**
* 释放当前所有获得的锁
* @return [type] [description]
*/
public function unlockAll() {
//此标志是用来标志是否释放所有锁成功
$allSuccess = true;
foreach ($this->lockedNames as $name => $expireAt) {
if (false === $this->unlock($name)) {
$allSuccess = false;
}
}
return $allSuccess;
}
/**
* 给当前所增加指定生存时间，必须大于0
* @param [type] $name [description]
* @return [type] [description]
*/
public function expire($name, $expire) {
//先判断是否存在该锁
if ($this->isLocking($name)) {
//所指定的生存时间必须大于0
$expire = max($expire, 1);
//增加锁生存时间
if ($this->redisString->expire("Lock:$name", $expire)) {
return true;
}
}
return false;
}
/**
* 判断当前是否拥有指定名字的所
* @param [type] $name [description]
* @return boolean [description]
*/
public function isLocking($name) {
//先看lonkedName[$name]是否存在该锁标志名
if (isset($this->lockedNames[$name])) {
//从redis返回该锁的生存时间
return (string)$this->lockedNames[$name] = (string)$this->redisString->get("Lock:$name");
}
return false;
}
}
/**
* 任务队列
*/
class RedisQueue {
private $_redis;
public function construct($param = null) {
$this->_redis = RedisFactory::get($param);
}
/**
* 入队一个 Task
* @param [type] $name 队列名称
* @param [type] $id 任务id（或者其数组）
* @param integer $timeout 入队超时时间(秒)
* @param integer $afterInterval [description]
* @return [type] [description]
*/
public function enqueue($name, $id, $timeout = 10, $afterInterval = 0) {
//合法性检测
if (empty($name) || empty($id) || $timeout <= 0) return false;
//加锁
if (!$this->_redis->lock->lock("Queue:{$name}", $timeout)) {
Logger::get('queue')->error("enqueue faild becouse of lock failure: name = $name, id = $id");
return false;
}
//入队时以当前时间戳作为 score
$score = microtime(true) + $afterInterval;
//入队
foreach ((array)$id as $item) {
//先判断下是否已经存在该id了
if (false === $this->_redis->zset->getScore("Queue:$name", $item)) {
$this->_redis->zset->add("Queue:$name", $score, $item);
}
}
//解锁
$this->_redis->lock->unlock("Queue:$name");
return true;
}
/**
* 出队一个Task，需要指定$id 和 $score
* 如果$score 与队列中的匹配则出队，否则认为该Task已被重新入队过，当前操作按失败处理
*
* @param [type] $name 队列名称
* @param [type] $id 任务标识
* @param [type] $score 任务对应score，从队列中获取任务时会返回一个score，只有$score和队列中的值匹配时Task才会被出队
* @param integer $timeout 超时时间(秒)
* @return [type] Task是否成功，返回false可能是redis操作失败，也有可能是$score与队列中的值不匹配（这表示该Task自从获取到本地之后被其他线程入队过）
*/
public function dequeue($name, $id, $score, $timeout = 10) {
//合法性检测
if (empty($name) || empty($id) || empty($score)) return false;
//加锁
if (!$this->_redis->lock->lock("Queue:$name", $timeout)) {
Logger:get('queue')->error("dequeue faild becouse of lock lailure:name=$name, id = $id");
return false;
}
//出队
//先取出redis的score
$serverScore = $this->_redis->zset->getScore("Queue:$name", $id);
$result = false;
//先判断传进来的score和redis的score是否是一样
if ($serverScore == $score) {
//删掉该$id
$result = (float)$this->_redis->zset->delete("Queue:$name", $id);
if ($result == false) {
Logger::get('queue')->error("dequeue faild because of redis delete failure: name =$name, id = $id");
}
}
//解锁
$this->_redis->lock->unlock("Queue:$name");
return $result;
}
/**
* 获取队列顶部若干个Task 并将其出队
* @param [type] $name 队列名称
* @param integer $count 数量
* @param integer $timeout 超时时间
* @return [type] 返回数组[0=>['id'=> , 'score'=> ], 1=>['id'=> , 'score'=> ], 2=>['id'=> , 'score'=> ]]
*/
public function pop($name, $count = 1, $timeout = 10) {
//合法性检测
if (empty($name) || $count <= 0) return [];
//加锁
if (!$this->_redis->lock->lock("Queue:$name")) {
Logger::get('queue')->error("pop faild because of pop failure: name = $name, count = $count");
return false;
}
//取出若干的Task
$result = [];
$array = $this->_redis->zset->getByScore("Queue:$name", false, microtime(true), true, false, [0, $count]);
//将其放在$result数组里 并 删除掉redis对应的id
foreach ($array as $id => $score) {
$result[] = ['id'=>$id, 'score'=>$score];
$this->_redis->zset->delete("Queue:$name", $id);
}
//解锁
$this->_redis->lock->unlock("Queue:$name");
return $count == 1 ? (empty($result) ? false : $result[0]) : $result;
}
/**
* 获取队列顶部的若干个Task
* @param [type] $name 队列名称
* @param integer $count 数量
* @return [type] 返回数组[0=>['id'=> , 'score'=> ], 1=>['id'=> , 'score'=> ], 2=>['id'=> , 'score'=> ]]
*/
public function top($name, $count = 1) {
//合法性检测
if (empty($name) || $count < 1) return [];
//取错若干个Task
$result = [];
$array = $this->_redis->zset->getByScore("Queue:$name", false, microtime(true), true, false, [0, $count]);
//将Task存放在数组里
foreach ($array as $id => $score) {
$result[] = ['id'=>$id, 'score'=>$score];
}
//返回数组
return $count == 1 ? (empty($result) ? false : $result[0]) : $result;
}
}위 내용은 PHP Redis 분산 잠금 및 작업 대기열 코드 예제에 대한 자세한 설명의 상세 내용입니다. 자세한 내용은 PHP 중국어 웹사이트의 기타 관련 기사를 참조하세요!