Verwenden Sie Redis in Laravel, um Warteschlangenaufgaben zu verarbeiten, aber kürzlich bin ich auf ein Problem gestoßen, das heißt, ich habe festgestellt, dass eine Aufgabe mehrmals ausgeführt wurde.
Lass uns zuerst über den Grund sprechen: Denn wenn in Laravel die Ausführungszeit einer Warteschlange (Aufgabe) mehr als 60 Sekunden beträgt, gilt sie als fehlgeschlagen und wird erneut zur Warteschlange hinzugefügt, was dazu führt wiederholte Ausführung derselben Aufgabe.
Die Logik dieser Aufgabe besteht darin, Inhalte an Benutzer zu senden, die auf der Grundlage des Warteschlangeninhalts abgerufen und durchlaufen und über die HTTP-Schnittstelle des Anforderungs-Backends gesendet werden müssen. Wenn beispielsweise 10.000 Benutzer vorhanden sind, die Anzahl der Benutzer groß ist oder die Verarbeitungsgeschwindigkeit der Schnittstelle nicht so hoch ist, beträgt die Ausführungszeit definitiv mehr als 60 Sekunden, sodass die Aufgabe erneut zur Warteschlange hinzugefügt wird. Noch schlimmer ist die Situation, wenn die vorherigen Aufgaben nicht innerhalb von 60 Sekunden ausgeführt werden. Auf diese Weise wird dieselbe Aufgabe nicht nur einmal, sondern mehrmals ausgeführt.
Lassen Sie uns den Schuldigen anhand des Laravel-Quellcodes finden.
Quellcodedatei: seller/laravel/framework/src/Illuminate/Queue/RedisQueue.php
/** * The expiration time of a job. * * @var int|null */ protected $expire = 60;
Diese $expire-Mitgliedsvariable ist ein fester Wert, dass eine Warteschlange ausgeführt werden sollte, egal wie lange sie 60 Sekunden dauert . So erhalten Sie die Warteschlange:
rrreeDa die Warteschlangenausführung fehlschlägt oder die Ausführung abläuft usw., wird sie in eine andere Sammlung verschoben und für einen erneuten Versuch gespeichert :
1. Entfernen Sie die Warteschlange, die aufgrund eines Ausführungsfehlers fehlgeschlagen ist. Die verzögerte Sammlung wird erneut in die aktuell ausgeführte Warteschlange verschoben.
2. Verschieben Sie die Warteschlange aufgrund eines Ausführungszeitlimits erneut von der reservierten Sammlung in die aktuell ausgeführte Warteschlange.
3. Nehmen Sie dann die Aufgabe aus der Warteschlange, beginnen Sie mit der Ausführung und stellen Sie die Warteschlange in die reservierte geordnete Sammlung.
Der Befehl eval wird hier verwendet, um diesen Prozess auszuführen, und es werden mehrere Lua-Skripte verwendet.
Holen Sie sich die auszuführende Aufgabe aus der Warteschlange:
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);
}
}
Sie können sehen, dass Laravel, wenn es die von Redis auszuführende Warteschlange erhält, auch eine Kopie in einen geordneten Satz einfügt und den Ablaufzeitstempel als Punktzahl verwendet.
Erst wenn die Aufgabe abgeschlossen ist, entfernen Sie die Aufgabe aus dem bestellten Satz. Der Code zum Entfernen der Warteschlange aus dieser geordneten Sammlung wird weggelassen. Schauen wir uns an, wie Laravel mit Warteschlangen umgeht, deren Ausführungszeit mehr als 60 Sekunden beträgt.
Das macht dieses Lua-Skript:
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}
Hier findet zrangebyscore die Elemente mit Werten von infinitesimal bis zum aktuellen Zeitstempel, also die Aufgaben, die vor 60 Sekunden zum Satz hinzugefügt wurden, und entfernt diese dann über das zremrangebyrank-Element aus dem Satz und schieben Sie es in die Warteschlange.
Du solltest es plötzlich merken, wenn du das siehst.
Wenn eine Warteschlange nicht innerhalb von 60 Sekunden abgeschlossen wird, verschiebt der Prozess beim Abrufen der Warteschlange die Aufgaben aus dem reservierten Satz erneut in die Warteschlange.