fila de jobs

Home > Colocar no ar > Fila de jobs 🚧

Fila de jobs em segundo plano

Important

Em desenvolvimento (2.0). Ainda não está numa versão lançada e pode mudar.

Vale para a 2.0 · Delphi XE em diante e Lazarus/FPC 3.2 em diante · Nível: intermediário

Em uma frase: a TRALJobQueue roda trabalhos em segundo plano que sobrevivem a um reinício, com Workers threads, novas tentativas com espera crescente e uma lista de jobs mortos para os que esgotaram as tentativas.

Você vai precisar: o pacote RALHostingJobQueuePackage (fila, memória e pasta; só requer PascalRAL); para guardar no banco, também RALHostingJobQueueDBPackage (TRALJobDBStore, requer RALDBPackage).

uses RALJobQueue, RALJobStore;

Fila := TRALJobQueue.Create(Self);
Fila.Folder := 'jobs';                        // relativa à pasta do executável
Fila.Workers := 4;
Fila.Handlers.AddHandler('email', EnviarEmail);
Fila.Server := Servidor;                      // liga e desliga com o servidor (ou Active := True)

Fila.Enqueue('email', Dados);                 // agora
Fila.Enqueue('email', Dados, 30000);          // daqui a 30 s
Fila.EnqueueAt('email', Dados, Amanha8h);     // numa data e hora (local)

procedure TForm1.EnviarEmail(Sender: TObject; AJob: TRALJob);
begin
  // roda numa thread de worker; uma exceção falha a tentativa
  case Enviar(AJob.Payload) of
    429: AJob.RetryAfter(60000, 'servidor pediu Retry-After');
    550: AJob.GiveUp('endereço inexistente');   // erro permanente: não tenta de novo
  end;
end;

Enqueue devolve o Id do job e funciona com a fila parada: o job espera. No handler, AJob.Id, Name, Payload (RawByteString) e Attempts (a tentativa atual, a partir de 1).

Tentativas e jobs mortos

Uma tentativa que falha volta para a fila depois de RetryDelay, e a espera dobra a cada tentativa até RetryMaxDelay. Depois de MaxAttempts tentativas o job vai para os jobs mortos. RetryAfter(ms, motivo) falha a tentativa sem exceção e escolhe a espera; GiveUp(motivo) manda o job direto para os mortos. Cada handler pode ter o seu MaxAttempts (0 = o da fila).

Enquanto o job roda, a fila renova o seu lease (LeaseTime); se o processo morrer, o lease acaba e outro worker pega o job.

Garantia: pelo menos uma vez

Nada que entrou na fila se perde quando o processo é morto. Em troca, um job que estava rodando roda de novo: depois de LeaseTime, ou logo ao religar com Exclusive := True (quando só este processo usa o armazenamento). Um handler que não pode repetir o efeito (cobrar, enviar) confere AJob.Id antes de agir (idempotência).

Onde os jobs ficam

armazenamento	como
memória	Folder vazio	rápido; perdido quando o processo termina
pasta	Folder (um TRALJobFileStore)	subpastas data/, ready/, leased/, dead/; cada mudança de estado é um rename; vários processos da mesma máquina podem dividir a pasta
banco	Store := TRALJobDBStore	várias máquinas dividem a tabela; o banco decide qual worker pega cada job

No TRALJobFileStore ligado em Store: SyncWrites (padrão True) grava os dados no disco antes de contar o job como escrito (sobrevive a queda de energia); ScanInterval (1000 ms) é de quanto em quanto tempo os jobs que outro processo enfileirou aparecem.

uses RALJobDBStore;

Banco := TRALJobDBStore.Create(Self);
Banco.DBModule := MeuDBModule;      // o pool de conexões
Banco.TableName := 'ral_jobs';      // o padrão
Fila.Store := Banco;

Com AutoCreate (padrão True) a tabela é criada ao ligar; CreateTableSQL mostra o comando para criar à mão. Os tipos de coluna seguem o banco do módulo (SQLite, PostgreSQL, MySQL/MariaDB, Firebird); o payload é guardado em base64.

Eventos

evento	quando
OnJobDone	o job rodou
OnJobFailed(AJob, AWillRetry)	uma tentativa falhou; AWillRetry diz se haverá outra (em AJob.Due) ou se o job morreu
OnJobDead	o job foi para os jobs mortos
OnError	o armazenamento falhou fora de um job; o resultado de um job é gravado até 5 vezes antes de chegar aqui
OnExecute	jobs cujo nome não tem handler; sem handler e sem OnExecute, o job morre com "No handler"

Os eventos de job rodam na thread do worker: para mexer em tela, sincronize.

Jobs mortos

Fila.Count(jsDead);                 // quantos
Fila.List(jsDead, Lista);           // os Ids
if Fila.ReadJob(Id, Job) then       // Job.LastError, Job.Attempts
  Fila.Requeue(Id);                 // volta agora, tentativas zeradas
Fila.RequeueAll;
Fila.PurgeDead;                     // apaga todos

Com o agendador

O agendador decide quando; a fila executa com tentativas:

procedure TForm1.FecharDia(AJob: TRALScheduledJob);
begin
  Fila.Enqueue('relatorio', FormatDateTime('yyyy-mm-dd', Date));
end;
// Agenda.Jobs.AddJob('fechamento', '0 23 * * *', FecharDia);

Propriedades

propriedade	padrão
Active	False	liga sem servidor (com Server, o servidor liga e desliga)
Exclusive	False	só este processo usa o armazenamento: ao ligar, os jobs que ficaram presos voltam na hora
Folder	''	pasta dos jobs; vazio = memória; ignorada com Store
LeaseTime	60000	ms que um worker segura o job, renovado enquanto roda
MaxAttempts	5	tentativas antes de morrer (1 = sem nova tentativa)
PollInterval	1000	ms que um worker ocioso espera antes de olhar de novo
RetryDelay	1000	ms antes da segunda tentativa; dobra a cada uma
RetryMaxDelay	3600000	a maior espera entre tentativas
Workers	1	threads que rodam jobs
Store	nil	um armazenamento ligado (banco, pasta configurada)

Stop (e parar o servidor) para de pegar jobs e espera os que estão rodando; um handler longo pode olhar Fila.Stopping para terminar antes. RunningCount diz quantos rodam agora.

Veja também

• Agendador
• Serviço e daemon
• Log, erros, limites, saúde