laravel在实现redis延迟队列时使用了lua脚本保证不同队列间操作的原子性
在laravel5.1中主要是通过4个lua脚本方法保证不同队列操作的原子性的
一、统计队列任务数量方法
1.llen 统计list队列数量
2.zcard统计zset队列数据量
/** * get the lua script for computing the size of queue. * * keys[1] - the name of the primary queue * keys[2] - the name of the "delayed" queue * keys[3] - the name of the "reserved" queue * * @return string */ public static function size() { return <<<'lua' return redis.call('llen', keys[1]) + redis.call('zcard', keys[2]) + redis.call('zcard', keys[3]) lua; }
二、pop队列任务放入reserved（保留）队列
/** * get the lua script for popping the next job off of the queue. * * keys[1] - the queue to pop jobs from, for example: queues:foo * keys[2] - the queue to place reserved jobs on, for example: queues:foo:reserved * argv[1] - the time at which the reserved job will expire * * @return string */ public static function pop() { return <<<'lua' -- pop the first job off of the queue... local job = redis.call('lpop', keys[1]) local reserved = false if(job ~= false) then -- increment the attempt count and place job on the reserved queue... reserved = cjson.decode(job) reserved['attempts'] = reserved['attempts'] + 1 reserved = cjson.encode(reserved) redis.call('zadd', keys[2], argv[1], reserved) end return {job, reserved} lua; }
三、将任务由添加reserved队列到delayed队列中
/** * get the lua script for releasing reserved jobs. * * keys[1] - the "delayed" queue we release jobs onto, for example: queues:foo:delayed * keys[2] - the queue the jobs are currently on, for example: queues:foo:reserved * argv[1] - the raw payload of the job to add to the "delayed" queue * argv[2] - the unix timestamp at which the job should become available * * @return string */ public static function release() { return <<<'lua' -- remove the job from the current queue... redis.call('zrem', keys[2], argv[1]) -- add the job onto the "delayed" queue... redis.call('zadd', keys[1], argv[2], argv[1]) return true lua; }
四、将reserved队列满足时间的任务合并到执行队列中
/** * get the lua script to migrate expired jobs back onto the queue. * * keys[1] - the queue we are removing jobs from, for example: queues:foo:reserved * keys[2] - the queue we are moving jobs to, for example: queues:foo * argv[1] - the current unix timestamp * * @return string */ public static function migrateexpiredjobs() { return <<<'lua' -- get all of the jobs with an expired "score"... local val = redis.call('zrangebyscore', keys[1], '-inf', argv[1]) -- if we have values in the array, we will remove them from the first queue -- and add them onto the destination queue in chunks of 100, which moves -- all of the appropriate jobs onto the destination queue very safely. if(next(val) ~= nil) then redis.call('zremrangebyrank', keys[1], 0, #val - 1) for i = 1, #val, 100 do redis.call('rpush', keys[2], unpack(val, i, math.min(i+99, #val))) end end return val lua; }
以上就是laravel延迟队列实现之lua脚本解析的详细内容。