Analisi della ragione per cui Laravel esegue la stessa attività di coda più volte

Introduzione

Il servizio di coda di Laravel offre un'API unificata per vari servizi di coda backend. La coda ti permette di eseguire in ritardo attività che richiedono molto tempo, come inviare una email. Questo può ridurre efficacemente il tempo di risposta delle richieste.

Scoperta del problema

Utilizzare Redis in Laravel per gestire le attività di coda, il framework fornisce funzionalità molto potenti, ma recentemente ho riscontrato un problema, ossia ho scoperto che un'attività viene eseguita più volte, ma perché?

Prima di tutto, parliamo delle cause:

Poiché in Laravel se un'attività (coda) ha un tempo di esecuzione superiore ai 60 secondi, viene considerata fallita e nuovamente aggiunta alla coda, il che può portare a una ripetizione dell'esecuzione della stessa attività.

La logica di questa attività è inviare contenuti agli utenti, è necessario estrarre gli utenti dalla coda e iterarli, inviando richieste a un'interfaccia HTTP backend. Ad esempio, ci sono 10000 utenti, se il numero di utenti è elevato o la velocità di elaborazione dell'interfaccia non è così veloce, il tempo di esecuzione sarà sicuramente superiore ai 60 secondi, quindi questa attività verrà nuovamente aggiunta alla coda. In casi ancora peggiori, se le attività precedenti non vengono eseguite entro 60 secondi, tutte le attività verranno nuovamente aggiunte alla coda, quindi la stessa attività non verrà eseguita una volta, ma più volte.

Di seguito, cerchiamo il responsabile di questo problema nei sorgenti di Laravel.

File di origine: vendor/laravel/framework/src/Illuminate/Queue/RedisQueue.php

/**
 * The expiration time of a job.
 *
 * @var int|null
 */
protected $expire = 60;

Questa variabile membro $expire è un valore fisso, Laravel pensa che un'intera coda dovrebbe essere eseguita in qualsiasi caso entro 60 secondi. Il metodo per ottenere la coda:

public function pop($queue = null)
{
 $original = $queue ?: $this->default; 
 $queue = $this->getQueue($queue); 
 $this->migrateExpiredJobs($queue.':delayed', $queue); 
 if (! is_null($this->expire)) {}}
  $this->migrateExpiredJobs($queue.':reserved', $queue);
 } 
 list($job, $reserved) = $this->getConnection()->eval(
  LuaScripts::pop(), 2, $queue, $queue.':reserved', $this->getTime() + $this->expire
 ); 
 if ($reserved) {
  return new RedisJob($this->container, $this, $job, $reserved, $original);
 }
}

Ci sono diverse operazioni per prelevare la coda, poiché le code che eseguono in modo anomalo o superano il tempo di esecuzione vengono salvate in un altro insieme per il retry, il processo è il seguente:

　　　 1. Ripush la coda che è stata eseguita in modo anomalo dall'insieme delayed alla coda corrente.

　　　 2. Ripush la coda che è stata eseguita in modo anomalo dall'insieme reserved alla coda corrente.

　　　 3. Poi è il momento di prelevare l'attività dalla coda per iniziare l'esecuzione, e inserire la coda nell'insieme ordinato di reserved.

Questo processo è eseguito utilizzando il comando eval, con diversi script lua.

Estrazione dell'attività dalla coda da eseguire:

local job = redis.call('lpop', KEYS[1])
local reserved = false
if(job ~= false) then
 reserved = cjson.decode(job)
 reserved['attempts'] = reserved['attempts'] + 1
 reserved = cjson.encode(reserved)
 redis.call('zadd', KEYS[2], ARGV[1], reserved)
end
return {job, reserved}

Si può vedere che Laravel, nel prelevare la coda da Redis da eseguire, mette anche una copia in un insieme ordinato e utilizza un timestamp di scadenza come valore.

Solo dopo aver completato questa attività, rimuovere questa attività dall'insieme ordinato. Il codice per rimuovere la coda dall'insieme ordinato è omesso, vediamo come Laravel gestisce le code con un tempo di esecuzione superiore ai 60 secondi.

ovvero l'operazione eseguita da questo script lua:

local val = redis.call('zrangebyscore', KEYS[1], '-inf', ARGV[1])
if(next(val) ~= nil) then
 redis.call('zremrangebyrank', KEYS[1], 0, #val - 1)
 for i = 1, #val, 100 do
  redis.call('rpush', KEYS[2], unpack(val, i, math.min(i+99, #val)))
 end
end
return true

Qui zrangebyscore trova gli elementi con punteggio da infinito piccolo al timestamp corrente, ovvero i task aggiunti alla coda 60 secondi fa, quindi rimuove questi elementi dal set e li riporta nella coda tramite zremrangebyrank.

Dopo aver letto questo, probabilmente avrai capito.

Se una coda non viene eseguita entro 60 secondi, il processo riporta questi task nella coda quando li preleva dal set riservato.

Sommario

Questo è tutto il contenuto dell'articolo, speriamo che il contenuto di questo articolo abbia un valore di riferimento per lo studio o il lavoro di tutti. Se hai domande, puoi lasciare un messaggio per scambiare opinioni, grazie per il supporto del tutorial di urla.

Dichiarazione: il contenuto di questo articolo è stato tratto da Internet, il diritto d'autore è dell'autore originale, il contenuto è stato contribuito e caricato autonomamente dagli utenti di Internet, questo sito non possiede il diritto di proprietà, non è stato elaborato manualmente e non assume responsabilità legali correlate. Se trovi contenuti sospetti di violazione del copyright, ti preghiamo di inviare una e-mail a notice#oldtoolbag.com (sostituisci # con @ quando invii l'e-mail) per segnalare e fornire prove pertinenti. Una volta verificata, questo sito eliminerà immediatamente il contenuto sospetto di violazione del copyright.