Laravel에서 Redis를 사용하여 대기열 작업을 처리합니다. 프레임워크에서 제공하는 기능은 매우 강력하지만 최근에 문제가 발생했습니다. 즉, 작업이 여러 번 실행되는 이유는 무엇입니까?
먼저 이유에 대해 이야기해 봅시다: Laravel에서는 큐(작업)의 실행 시간이 60초를 초과하면 실패한 것으로 간주되어 큐에 다시 추가되므로 다음과 같은 결과가 발생합니다. 동일한 작업을 반복적으로 실행합니다.
이 작업의 논리는 사용자에게 콘텐츠를 푸시하는 것입니다. 사용자는 대기열 콘텐츠를 기반으로 검색 및 이동되어야 하며 요청 백엔드 HTTP 인터페이스를 통해 전송되어야 합니다. 예를 들어 사용자가 10,000명인 경우, 사용자 수가 많거나 인터페이스 처리 속도가 그다지 빠르지 않다면 실행 시간은 당연히 60초보다 커지므로 작업이 대기열에 다시 추가됩니다. 상황은 더욱 심각합니다. 이전 작업이 60초 이내에 실행되지 않으면 대기열에 다시 추가됩니다. 이렇게 하면 동일한 작업이 한 번만 실행되는 것이 아니라 여러 번 실행됩니다.
라라벨 소스코드에서 범인을 찾아보자.
소스 코드 파일: Vendor/laravel/framework/src/Illuminate/Queue/RedisQueue.php
/** * The expiration time of a job. * * @var int|null */ protected $expire = 60;
이 $expire 멤버 변수는 60초가 걸리더라도 대기열이 실행되어야 한다고 믿습니다. . 큐를 얻는 방법:
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);
}
}
큐를 가져오는 데는 여러 단계가 있습니다. 큐 실행이 실패하거나 실행 시간이 초과되는 등의 이유로 다른 컬렉션에 저장되고 재시도를 위해 프로세스는 다음과 같습니다. :
1. 실행 실패로 인해 실패한 큐를 제거합니다. 지연된 수집이 현재 실행 중인 큐에 다시 푸시됩니다.
2. 실행 시간 초과로 인해 예약된 컬렉션에서 현재 실행 큐로 큐를 다시 푸시합니다.
3. 그런 다음 대기열에서 작업을 가져와 실행을 시작하고 대기열을 예약된 순서 컬렉션에 넣습니다.
여기에서는 이 프로세스를 실행하기 위해 eval 명령이 사용되며 여러 lua 스크립트가 사용됩니다.
실행할 대기열에서 작업 가져오기:
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}
Laravel이 Redis에서 실행할 대기열을 가져올 때 주문된 세트에 복사본을 넣고 만료 타임스탬프를 점수로 사용하는 것을 볼 수 있습니다.
작업이 완료된 경우에만 주문한 세트에서 작업을 제거하세요. 이 정렬된 컬렉션에서 큐를 제거하는 코드는 생략되었습니다. Laravel이 실행 시간이 60초를 초과하는 큐를 처리하는 방법을 살펴보겠습니다.
이 Lua 스크립트의 기능은 다음과 같습니다.
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
여기서 zrangebyscore는 무한소부터 현재 타임스탬프까지의 점수가 있는 요소, 즉 60초 전에 세트에 추가된 작업을 찾은 다음 zremrangebyrank 요소를 통해 세트에서 이러한 항목을 제거합니다. 그리고 그것을 큐에 밀어 넣습니다.
이걸 보시면 문득 깨닫게 되실 겁니다.
큐가 60초 이내에 실행되지 않으면 프로세스는 큐를 가져올 때 예약된 세트의 작업을 다시 큐로 푸시합니다.