延迟队列介绍
顾名思义，延迟队列就是进入该队列的消息会被延迟消费的队列。
例如：滴滴打车订单完成后，如果用户一直不评价，48小时后会将自动评价为5星。
Redis实现延迟队列
Redis 可以利用 zset (有序列表)来实现，将消息序列化成一个字符串作为 zset的 value；
这个消息的到期处理时间作为 score，利用多个线程轮询 zset 获取到期的任务进行处理。
多线程是为了保证可用性，万一挂了一个线程还有其他线程可以继续处理;
因为有多个线程，所以需要考虑并发争抢任务，确保任务不会多次执行。
代码实现
代码语言：javascript复制require_once("../RedisClient.php");
$client = RedisClient::getInstance();
//延时队列
function delay(string $message, int $timeout = 5)
{
global $client;
$time = microtime(true) $timeout;
return $client->zadd('delay:', [$message => $time]);
}
//顺序消费延迟队列中的消息
function loop()
{
global $client;
while (true) {
//从延迟队列获取一条最近时间的消息
$message_data = $client->zrangebyscore('delay:', '-inf', microtime(true), ['withscores' => true, 'limit' => [0, 1]]);
//延迟队列中无消息
if (!$message_data) {
sleep(1);
continue;
}
//提取消息数据
$message = key($message_data);
//从延迟队列中删除刚获取的消息
$success = $client->zrem('delay:', $message);
//多线程或多进程争抢消息时,
//根据zrem返回值判断,当前实例有没有抢到任务
//抢到任务,做业务处理后返回
if ($success) {
//do something..
echo sprintf("消费的消息,[%s]", $message) . PHP_EOL;
}
}
}
//delay('test1');
//delay('test2');
//delay('test2');
//loop();
# php queue.php
消费的消息,[mmm1]
消费的消息,[mmm2]
消费的消息,[mmm3]进一步优化
细心的同学会发现上面算法代码中，有几处问题
- 同一个任务被多个进程取到后再使用
zrem进行争抢，没有抢到的进程白白浪费了一次任务； - 取出条数和删除只能一条，且
zrangebyscore和zrem不是原子操作； - 消息取出后，执行了一部分逻辑，服务器突然重启了，剩下的逻辑没有执行完成该如何处理？
我们可以通过使用lua脚本，解决前面两个问题，至于第三个问题可以通过代码层面其他数据库事务解决。
require_once("../RedisClient.php");
use PredisCommandScriptCommand;
$client = RedisClient::getInstance();
/**
* 从消息队列中搜索符合条件的最近n条消息
* 返回消息内容并从消息队列中删除
* @param string queue_key 消息队列的key
* @param int $min 搜索时间戳开始时间
* @param int $max 搜索时间戳结束时间
* @param int $offset 要跳过的消息数量
* @param int $limit 获取消息数量
* @return array 删除成功的消息的消息内容
* @
*/
class getAndDeleteRecentMessageScript extends ScriptCommand {
public function getKeysCount()
{
return 1;
}
public function getScript()
{
return <<<LUA
-- 消息队列的redisKey
local queue = KEYS[1]
-- 搜索范围的最大/最小值,偏移量和取值数量
local min, max, offset, count = ARGV[1], ARGV[2], ARGV[3], ARGV[4]
local message = false
local messages = {}
local queue_value = {}
local insert = table.insert
-- 获取最近n条消息并删除消息
queue_value = redis.call("ZRANGEBYSCORE",queue,min,max,"LIMIT",offset,count)
for idx, message in pairs(queue_value) do
if redis.call("ZREM",queue,message) then
insert(messages, idx, message)
end
end
-- 返回删除成功的消息
return messages
LUA;
}
}
$client->getProfile()->defineCommand('get_and_delete_recent_message','getAndDeleteRecentMessageScript');
//向延迟队列中写入10条数据
foreach(range(1,10) as $msg_id){
$success = delay("msg{$msg_id}");
if($success){
echo "写入消息[msg{$msg_id}], 成功" . PHP_EOL;
}
}
//删除最近写入的 2条
$ret = $conn->get_and_delete_recent_message('delay:',0,microtime(true),3,2);
var_export($ret);后记
延时队列是一个实现“延时消息”的好方法，解决了业务问题。至于可达性、幂等性未来另述。
