Task-Wiederholungen

Diese Seite beschreibt, wie die Aufgabenerneuerung konfiguriert werden kann.

Standardverhalten

Standardmäßig wird asynq eine Aufgabe maximal 25 Mal erneut versuchen. Bei jedem erneuten Versuch verwendet er eine exponentielle Backoff-Strategie zur Berechnung der erneuten Verzögerung. Wenn eine Aufgabe alle ihre erneuten Versuche erschöpft hat (Standard sind 25 Mal), wird die Aufgabe zu Debugging- und Inspektionszwecken in den Zustand archiviert und wird nicht automatisch erneut versucht (Sie können die Aufgabe immer noch manuell über die CLI oder WebUI ausführen).

Die folgenden Eigenschaften der Aufgabenerneuerung können angepasst werden:

• Maximale Anzahl erneuter Versuche für jede Aufgabe
• Zeitintervall, das gewartet werden soll, bevor eine fehlgeschlagene Aufgabe erneut versucht werden kann (d. h. Verzögerung)
• Ob der erneute Versuchszähler der Aufgabe verwendet werden soll
• Ob die erneuten Versuche übersprungen und die Aufgabe direkt in das Archiv gesendet werden sollen

Die verbleibenden Teile dieser Seite beschreiben jede der oben genannten benutzerdefinierten Optionen.

Anpassen der maximalen erneuten Versuche für Aufgaben

Sie können die maximale Anzahl erneuter Versuche für eine Aufgabe angeben, wenn Sie die Aufgabe unter Verwendung der asynq.MaxRetry-Option in die Warteschlange stellen.

Beispiel:

client.Enqueue(task, asynq.MaxRetry(5))

Dies gibt an, dass die task maximal fünf Mal erneut versucht werden soll.

Alternativ können Sie, wenn Sie die maximale Anzahl erneuter Versuche für eine bestimmte Aufgabe festlegen möchten, dies als Standardoption für die Aufgabe festlegen.

task := asynq.NewTask("feed:import", nil, asynq.MaxRetry(5))
client.Enqueue(task) // MaxRetry auf 5 gesetzt

Anpassen der erneuten Verzögerung

Sie können angeben, wie die erneute Verzögerung mithilfe der RetryDelayFunc-Option in der Config-Struktur berechnet wird.

Die Signatur von RetryDelayFunc ist wie folgt:

// n ist die Anzahl der Male, die die Aufgabe erneut versucht wurde
// e ist der vom Aufgabenhandler zurückgegebene Fehler
// t ist die zugehörige Aufgabe
RetryDelayFunc func(n int, e error, t *asynq.Task) time.Duration

Beispiel:

srv := asynq.NewServer(redis, asynq.Config{
    Concurrency: 20,
    RetryDelayFunc: func(n int, e error, t *asynq.Task) time.Duration {
        return 2 * time.Second
    },
})

Dies gibt an, dass alle fehlgeschlagenen Aufgaben zwei Sekunden warten, bevor sie erneut verarbeitet werden.

Das Standardverhalten ist ein exponentieller Rückgang, definiert durch DefaultRetryDelayFunc. Das folgende Beispiel zeigt, wie die erneute Verzögerung für bestimmte Aufgabentypen angepasst wird:

srv := asynq.NewServer(redis, asynq.Config{
    // Für Aufgaben "foo" immer eine Verzögerung von 2 Sekunden verwenden, andere Aufgaben verwenden das Standardverhalten.
    RetryDelayFunc: func(n int, e error, t *asynq.Task) time.Duration {
        if t.Type() == "foo" {
            return 2 * time.Second 
        }
        return asynq.DefaultRetryDelayFunc(n, e, t) 
    },
})

Fehler ohne Ausfall

Manchmal möchten Sie möglicherweise einen Fehler aus dem Handler zurückgeben und die Aufgabe später erneut versuchen, jedoch ohne den erneuten Versuchszähler der Aufgabe zu verwenden. Zum Beispiel möchten Sie möglicherweise später erneut versuchen, weil die Arbeitsgruppe nicht über genügend Ressourcen verfügt, um die Aufgabe zu bearbeiten. Während der Serverinitialisierung können Sie wählen, eine Config für die IsFailure(error) bool-Funktion bereitzustellen. Diese Prädikatsfunktion bestimmt, ob der Fehler vom Handler als Ausfall zählt. Falls die Funktion false zurückgibt (d. h. ein Fehler ohne Ausfall), wird der Server den erneuten Versuchszähler der Aufgabe nicht verwenden und die Aufgabe einfach für einen späteren erneuten Versuch planen.

Beispiel:

var ErrResourceNotAvailable = errors.New("Keine Ressourcen verfügbar")

func HandleResourceIntensiveTask(ctx context.Context, task *asynq.Task) error {
    if !IsResourceAvailable() {
        return ErrResourceNotAvailable
    }
    // ... Logik zur Behandlung einer ressourcenintensiven Aufgabe
}

// ...

srv := asynq.NewServer(redisConnOpt, asynq.Config{
    // ... andere Konfigurationsoptionen
    IsFailure: func(err error) bool {
        return err != ErrResourceNotAvailable // Fehler ohne Ausfall, wenn keine Ressourcen verfügbar sind
    },
})

Erneuten Versuch überspringen

Wenn Handler.ProcessTask einen SkipRetry-Fehler zurückgibt, wird die Aufgabe unabhängig von der verbleibenden Anzahl erneuter Versuche archiviert. Der zurückgegebene Fehler kann SkipRetry oder ein Fehler sein, der mit einem SkipRetry-Fehler umschlossen ist.

func ExampleHandler(ctx context.Context, task *asynq.Task) error {
    // Aufgabenbearbeitungslogik hier...
    // Wenn der Handler weiß, dass die Aufgabe nicht erneut versucht werden soll, return SkipRetry
    return fmt.Errorf(": %w", asynq.SkipRetry)
}