Tarefas Agendadas

Visão Geral

Você pode executar um Agendador em paralelo com o Servidor para lidar com tarefas periodicamente. O Agendador adicionará periodicamente tarefas à fila, as quais então serão executadas pelos servidores trabalhadores disponíveis no cluster.

É necessário garantir que apenas um Agendador esteja em execução para cada agendamento, a fim de evitar tarefas duplicadas. Utilizar uma abordagem centralizada significa que a sincronização não é necessária e o serviço pode ser executado sem utilizar travas.

Se precisar adicionar e remover dinamicamente tarefas periódicas, utilize o GerenciadorDeTarefasPeriódicas em vez de utilizar diretamente o Agendador. Consulte este wiki para obter informações mais detalhadas.

Fuso Horário

Por padrão, as tarefas periódicas utilizam o horário UTC, mas é possível utilizar SchedulerOpts para alterar o fuso horário utilizado.

// Por exemplo, utilize o fuso horário America/Los_Angeles em vez do fuso horário UTC padrão.
loc, err := time.LoadLocation("America/Los_Angeles")
if err != nil {
    panic(err)
}
scheduler := asynq.NewScheduler(
    redisConnOpt, 
    &asynq.SchedulerOpts{
        Location: loc,
    },
)

Registro de Tarefa

Para adicionar periodicamente tarefas à fila, é necessário registrar um registro de tarefa com o Agendador.

scheduler := asynq.NewScheduler(redisConnOpt, nil)

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

// É possível utilizar uma string de especificação do cron para definir o agendamento.
entryID, err := scheduler.Register("* * * * *", task)
if err != nil {
    log.Fatal(err)
}
log.Printf("registrou uma entrada: %q\n", entryID)

// Também é possível utilizar "@every " para especificar intervalos.
entryID, err = scheduler.Register("@every 30s", task)
if err != nil {
    log.Fatal(err)
}
log.Printf("registrou uma entrada: %q\n", entryID)

// Também é possível passar opções.
entryID, err = scheduler.Register("@every 24h", task, asynq.Queue("minhafila"))
if err != nil {
    log.Fatal(err)
}
log.Printf("registrou uma entrada: %q\n", entryID)

Executando o Agendador

Para iniciar o Agendador, chame Run no Agendador.

scheduler := asynq.NewScheduler(redisConnOpt, nil)

// ... Registre tarefas

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

Chamar Run aguardará o sinal TERM ou INT (por exemplo, Ctrl-C).

Tratamento de Erros

É possível fornecer uma função de manipulação para lidar com erros se o Agendador falhar ao enfileirar tarefas.

func handleEnqueueError(task *asynq.Task, opts []asynq.Option, err error) {
    // Sua lógica de tratamento de erro
}

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

Verificação via CLI

A CLI possui um subcomando denominado cron para verificar registros do Agendador.

Para visualizar todos os registros do Agendador em execução atualmente, é possível executar o seguinte comando:

asynq cron ls

Este comando fornecerá uma lista contendo o ID, a especificação do agendamento, o próximo horário de enfileiramento e o último horário de enfileiramento para cada registro.

Também é possível executar o seguinte comando para visualizar o histórico de cada registro:

asynq cron history