Source code analysis explains why Laravel repeatedly executes the same queue task

Laravel's queue service provides a unified API for various background queue services. The following article introduces to you through source code analysis the reasons why Laravel repeatedly executes the same queue task. The article introduces it through sample code. For details, friends in need can refer to it. Let’s take a look below.

Preface

Laravel’s queue service provides a unified API for various background queue services. Queues allow you to delay the execution of time-consuming tasks, such as sending an email. This can effectively reduce the request response time.

Found the problem

Use Redis in Laravel to process queue tasks. The functions provided by the framework are very powerful, but recently I encountered a problem, that is It is found that a task is executed multiple times. Why is this?

Let’s talk about the reason first:

Because in Laravel, if a queue (task) execution time is greater than After 60 seconds, it will be considered an execution failure and rejoined in the queue, which will cause the same task to be executed repeatedly.

The logic of this task is to push content to users. Users need to be retrieved and traversed based on the queue content, and sent through the request backend HTTP interface. For example, if there are 10,000 users, if the number of users is large or the interface processing speed is not that fast, the execution time will definitely be greater than 60 seconds, so the task will be re-added to the queue. The situation is even worse. If the previous tasks are not executed within 60 seconds, they will be re-added to the queue, so that the same task will be executed not only once, but multiple times.

Let’s find the culprit from the Laravel source code.

Source code file: vendor/laravel/framework/src/Illuminate/Queue/RedisQueue.php

/**
 * The expiration time of a job.
 *
 * @var int|null
 */
protected $expire = 60;

The $expire member variable is a fixed value. Laravel believes that a queue should be executed no matter how long it takes 60 seconds. Method to get the queue:

public function pop($queue = null)
{
 $original = $queue ?: $this->default; 
 $queue = $this->getQueue($queue); 
 $this->migrateExpiredJobs($queue.':delayed', $queue); 
 if (! is_null($this->expire)) {
  $this->migrateExpiredJobs($queue.':reserved', $queue);
 } 
 list($job, $reserved) = $this->getConnection()->eval(
  LuaScripts::pop(), 2, $queue, $queue.':reserved', $this->getTime() + $this->expire
 ); 
 if ($reserved) {
  return new RedisJob($this->container, $this, $job, $reserved, $original);
 }
}

There are several steps to get the queue, because if the queue fails to execute or the execution times out, it will be put into another collection and saved. , in order to retry, the process is as follows:

1. Re-rpush the queue that failed due to execution from the delayed collection to the currently executed queue.

2. Re-rpush the queue due to execution timeout from the reserved collection to the currently executed queue.

3. Then take the task from the queue and start executing it, and put the queue into the reserved ordered collection.

The eval command is used here to execute this process, and several lua scripts are used.

Get the task from the queue to be executed:

local job = redis.call('lpop', KEYS[1])
local reserved = false
if(job ~= false) then
 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}

You can see that Laravel is getting the task to be executed by Redis When queuing, a copy will be put into an ordered collection at the same time, and the expiration timestamp will be used as the score.

Only when the task is completed, the task will be removed from the ordered set. The code to remove the queue from this ordered collection is omitted. Let's take a look at how Laravel handles queues whose execution time is greater than 60 seconds.

This is the operation performed by this lua script:

local val = redis.call('zrangebyscore', KEYS[1], '-inf', ARGV[1])
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 true

Here zrangebyscore finds the score from infinitesimal To the elements of the current timestamp, that is, the tasks added to the collection 60 seconds ago, these elements are then removed from the collection through zremrangebyrank and rpush to the queue.

You should suddenly realize it when you see this.

If a queue has not been executed within 60 seconds, the process will rpush the tasks from the reserved set to the queue again when fetching the queue.

Summary

Articles you may be interested in:

About Laravel Redis Detailed explanation of the problem of multiple processes taking queues at the same time

Detailed explanation of php-msf source code

thinkphp5 URL and routing Detailed explanation of functions and examples

The above is the detailed content of Source code analysis explains why Laravel repeatedly executes the same queue task. For more information, please follow other related articles on the PHP Chinese website!