Zaplanowane zadania

Przegląd

Możesz uruchomić Scheduler jednocześnie z Serwerem, aby obsługiwać zadania okresowo. Scheduler periodycznie będzie dodawał zadania do kolejki, które następnie będą wykonywane przez dostępne serwery robocze w klastrze.

Musisz upewnić się, że dla każdego harmonogramu uruchomiony jest tylko jeden Scheduler, aby uniknąć zduplikowanych zadań. Użycie podejścia scentralizowanego oznacza, że synchronizacja nie jest konieczna, a usługa może działać bez użycia blokad.

Jeśli musisz dynamicznie dodawać i usuwać zadania okresowe, użyj PeriodicTaskManager zamiast bezpośrednio korzystać z Scheduler. Zobacz ten wiki dla bardziej szczegółowych informacji.

Strefa czasowa

Domyślnie zadania okresowe korzystają ze strefy czasu UTC, ale można użyć SchedulerOpts do zmiany używanej strefy czasowej.

// Na przykład, użyj strefy czasowej America/Los_Angeles zamiast domyślnej strefy czasowej UTC.
loc, err := time.LoadLocation("America/Los_Angeles")
if err != nil {
    panic(err)
}
scheduler := asynq.NewScheduler(
    redisConnOpt, 
    &asynq.SchedulerOpts{
        Location: loc,
    },
)

Rejestracja zadania

Aby periodycznie dodawać zadania do kolejki, musisz zarejestrować rekord zadania z Schedulerem.

scheduler := asynq.NewScheduler(redisConnOpt, nil)

task := asynq.NewTask("example_task", nil)

// Możesz użyć ciągu specyfikacji cron do określenia harmonogramu.
entryID, err := scheduler.Register("* * * * *", task)
if err != nil {
    log.Fatal(err)
}
log.Printf("zarejestrowano wpis: %q\n", entryID)

// Możesz także użyć "@every " do określenia interwałów.
entryID, err = scheduler.Register("@every 30s", task)
if err != nil {
    log.Fatal(err)
}
log.Printf("zarejestrowano wpis: %q\n", entryID)

// Możesz także przekazać opcje.
entryID, err = scheduler.Register("@every 24h", task, asynq.Queue("myqueue"))
if err != nil {
    log.Fatal(err)
}
log.Printf("zarejestrowano wpis: %q\n", entryID)

Uruchamianie Schedulera

Aby uruchomić Scheduler, wywołaj Run na Schedulerze.

scheduler := asynq.NewScheduler(redisConnOpt, nil)

// ... Rejestrowanie zadań

if err := scheduler.Run(); err != nil {
    log.Fatal(err)
}

Wywołanie Run będzie oczekiwać na sygnał TERM lub INT (np. Ctrl-C).

Obsługa błędów

Możesz dostarczyć funkcję obsługi do obsługi błędów, jeśli Scheduler nie może umieścić zadań w kolejce.

func handleEnqueueError(task *asynq.Task, opts []asynq.Option, err error) {
    // Twoja logika obsługi błędów
}

scheduler := asynq.NewScheduler(
    redisConnOpt, 
    &asynq.SchedulerOpts{
        EnqueueErrorHandler: handleEnqueueError,
    },
)

Sprawdzanie za pomocą CLI

CLI ma podpolecenie o nazwie cron do sprawdzania rekordów Schedulera.

Aby wyświetlić wszystkie rekordy aktualnie działającego Schedulera, możesz wykonać poniższą komendę:

asynq cron ls

Ta komenda wyświetli listę zawierającą ID, specyfikację harmonogramu, następny czas dodania do kolejki oraz ostatni czas dodania dla każdego rekordu.

Możesz także wykonać poniższą komendę, aby wyświetlić historię każdego rekordu:

asynq cron history