HI Ich führe mehrere n8n-Worker aus, die Jobs aus PostgreSQL verarbeiten. Die Worker beanspruchen Jobs, indem sie ihren Status von „pending
Hey @Mary_Berry, während du auf eine Antwort wartest, könnten dir diese Dinge helfen:
Empfohlene Ressourcen
Automatisch zu deiner Frage zugeordnet.
Dokumentation:
Forum:
@Niffzy, @Emmas - ihr habt schon bei ähnlichen Problemen geholfen, könnt ihr mal schauen?
Automatisch vorgeschlagen vom Community-Bot von n8n. Es ist ein Pilotprojekt – gib hier gerne Feedback.
Verwenden Sie ein Lease-/Heartbeat-Modell, anstatt einfach davon auszugehen, dass jeder Verarbeitungsjob älter als 15 Minuten inaktiv ist.
Ich würde jedem beanspruchten Job einen locked_at-Zeitstempel und vorzugsweise eine worker_id geben. Worker können den Heartbeat während der Verarbeitung regelmäßig aktualisieren.
Zur Wiederherstellung würde ich ein atomares Update verwenden, damit zwei Worker denselben Job nicht wiederherstellen können:
UPDATE jobs
SET status = 'pending',
locked_at = NULL,
worker_id = NULL
WHERE id = $1
AND status = 'processing'
AND locked_at < NOW() - INTERVAL '15 minutes'
RETURNING id;
Wenn die Abfrage eine Zeile zurückgibt, hat dieser Worker sie erfolgreich zurückgefordert. Wenn sie nichts zurückgibt, hat ein anderer Worker sie bereits bearbeitet oder der Job ist nicht mehr berechtigt.
Bei lang laufenden Jobs würde ich definitiv einen Heartbeat einem festen Timeout vorziehen:
Job beanspruchen
↓
status = processing
↓
Heartbeat aktualisieren
↓
Job verarbeiten
↓
Als abgeschlossen markieren
Dann hätte ich einen kleinen Wiederherstellungs-Workflow, der regelmäßig nach Jobs sucht, deren Heartbeat abgelaufen ist.
Hallo @Mary_Berry
Mach die Job-Verantwortung explizit, anstatt dich nur auf das Alter des Verarbeitungsstatus zu verlassen.
Wenn ein Worker einen Job übernimmt, gib ihm eine worker_id und eine Lease-Ablaufzeit:
UPDATE jobs
SET status = 'processing',
worker_id = $1,
lease_until = NOW() + INTERVAL '15 minutes'
WHERE id = $2
AND status = 'pending'
RETURNING id;
While the job is running, the worker can extend its lease:
UPDATE jobs
SET lease_until = NOW() + INTERVAL '15 minutes'
WHERE id = $1
AND worker_id = $2
AND status = 'processing';
Ein Wiederherstellungsprozess kann dann nach abgelaufenen Leases suchen und diese Jobs wieder in die Warteschlange einreihen.
Job beanspruchen
↓
Lease aktiv
↓
Verarbeiten + Lease erneuern
↓
Abgeschlossen
Wenn der Worker abstürzt, läuft die Lease irgendwann ab und ein anderer Worker kann die Aufgabe übernehmen.
Ich würde auch die tatsächliche Job-Verarbeitung idempotent gestalten und einen Wiederholungszähler führen, damit ein wiederholt fehlgeschlagener Job schließlich in eine Dead-Letter-Queue verschoben werden kann, anstatt unendlich erneut versucht zu werden.
Hallo @Mary_Berry
Ein Lease in deiner Tabelle kann sich nicht selbst erneuern, wie es n8n tut. n8n hält ein 60-Sekunden-Lease auf die Ausführung und erhöht es alle 10 Sekunden über einen Timer, aber eine Ausführung ist ein direkter Durchlauf durch Nodes ohne Background-Thread, daher wird locked_at nicht berührt, während ein HTTP Request-Node blockiert. Die Erneuerung findet nur dort statt, wo ein Node ausgeführt wird, was bedeutet eine Loop Over Items mit einem Renew-Node darin. Alles andere muss dimensioniert werden, anstatt Heartbeat zu verwenden.
Dimensionieren bedeutet, zwei Uhren abzustimmen. Die Uhr, die entscheidet, dass dein Worker tot ist, gehört n8n, nicht dir, und ein Worker, der zu beschäftigt ist, um sich innerhalb von 60 Sekunden zu erneuern, hat seine Ausführung als ins Stocken geraten deklariert, während die Zeile immer noch als deine markiert ist. Bei 2.x wird diese Ausführung nicht erneut zugestellt, also ist sie weg und die Zeile wartet deine vollen 15 Minuten ab. Setze n8ns Lease über deinen langsamsten Job, dann dein Job-Lease darüber:
QUEUE_WORKER_LOCK_DURATION=900000
Wenn die Reihenfolge anders herum ist, gibt der Reaper den Job an einen zweiten Worker, während der erste ihn noch ausführt.