Laravel延遲隊(duì)列實(shí)現(xiàn)之Lua腳本解析

meteor199 發(fā)布于2019-06-28 18:44 / 1406人閱讀

摘要：在實(shí)現(xiàn)延遲隊(duì)列時(shí)使用了腳本保證不同隊(duì)列間操作的原子性在中主要是通過(guò)個(gè)腳本方法保證不同隊(duì)列操作的原子性的一統(tǒng)計(jì)隊(duì)列任務(wù)數(shù)量方法統(tǒng)計(jì)隊(duì)列數(shù)量統(tǒng)計(jì)隊(duì)列數(shù)據(jù)量二隊(duì)列任務(wù)放入保留隊(duì)列三將任務(wù)由添加隊(duì)列到隊(duì)列中四將隊(duì)

Laravel在實(shí)現(xiàn)Redis延遲隊(duì)列時(shí)使用了Lua腳本保證不同隊(duì)列間操作的原子性
在Laravel5.1中主要是通過(guò)4個(gè)Lua腳本方法保證不同隊(duì)列操作的原子性的

一、統(tǒng)計(jì)隊(duì)列任務(wù)數(shù)量方法

1.llen 統(tǒng)計(jì)list隊(duì)列數(shù)量

2.zcard統(tǒng)計(jì)zset隊(duì)列數(shù)據(jù)量

    /**
     * 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隊(duì)列任務(wù)放入reserved（保留）隊(duì)列

    /**
     * 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;
    }

三、將任務(wù)由添加reserved隊(duì)列到delayed隊(duì)列中

    /**
     * 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隊(duì)列滿足時(shí)間的任務(wù)合并到執(zhí)行隊(duì)列中

    /**
     * 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;
    }