Redis 延迟队列实现(基于PHP)

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]

require_once("../RedisClient.php");
use Predis\Command\ScriptCommand;
$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);