Quante volte hai avuto bisogno di disaccoppiare i processi utilizzando una coda, solo per scoprire che non ne avevi una a disposizione a causa di vincoli di risorse o dell’impossibilità di installare software aggiuntivo?
In questa serie, ti mostrerò come risolvere questo problema utilizzando diverse librerie e strumenti, anche in assenza di una coda reale. Queste soluzioni sono versatili e adattabili a vari scenari, permettendoti di ottenere lo stesso risultato senza dipendere da infrastrutture complesse.
Prima di iniziare, voglio rassicurarti: questi approcci non sono utili solo quando una coda tradizionale non è disponibile. Sono abbastanza flessibili da essere adottati anche quando potresti utilizzarne una vera, senza alcuna controindicazione.
Esplorando gli esempi di questa serie, potrai valutare quale strategia sia più adatta al tuo caso d’uso. 🚀
Fastq
Il primo articolo tratta di fastq, un pacchetto npm per Node.js sviluppato da Matteo Collina.
Come puoi immaginare, questa soluzione funziona esclusivamente in un ambiente Node.js.
Utilizzando fastq, puoi creare una o più code in memoria. Tuttavia, questo significa che se il processo viene terminato, tutti i dati nella coda andranno persi. Per questo motivo, è necessario implementare una strategia per ricreare la coda all’avvio dell’applicazione.
Come iniziare con fastq
Per iniziare, devi installare fastq all’interno del tuo progetto Node.js. Ho già preparato un progetto semplice che puoi seguire passo dopo passo, lo trovi [qui].
Dopo aver eseguito git clone e npm install, il primo passo è lanciare il comando:
npm install fastqCosì facendo, installerai la dipendenza nel tuo progetto e sarai pronto a usare fastq.
Come puoi vedere, l’installazione e la configurazione di questo pacchetto sono rapide e semplici.
La tua prima coda
Ora è il momento di creare la tua prima coda con fastq!
Fastq offre un’API semplice per interagire con le code. Per iniziare, devi definire una funzione worker che gestirà gli elementi all’interno della coda.
Questa funzione accetta due argomenti:
1️⃣ Il primo è l’elemento da processare nella coda.
2️⃣ Il secondo varia in base all’approccio scelto per completare l’esecuzione:
- Callback approach: il secondo argomento è una funzione
done, che deve essere chiamata al termine del processo. In caso di successo, devi invocarla passandonullcome primo parametro; in caso di errore, devi passare l’errore. - Promise approach: puoi omettere il secondo argomento e gestire eventuali errori lanciandoli direttamente.
Per questo esempio, utilizzeremo l’approccio basato su Promise, ma sentiti libero di implementarlo anche con le callback se lo preferisci.
Creiamo la nostra coda con fastq
Creiamo un nuovo file src/queue.ts che conterrà il nostro worker per la coda.
Per rendere questo esempio più concreto, simuleremo un sistema di registrazione utenti, in cui la coda avrà il compito di inviare un’email di conferma a un nuovo utente. Ovviamente, non ci addentreremo nei dettagli reali dell’invio email, ma utilizzeremo servizi fittizi per dimostrare il funzionamento di fastq.
Ora è il momento di costruire la tua prima coda!
Nel file queue.ts, iniziamo definendo l’interfaccia per il task da gestire:
interface UserCreatedTask {
id: string;
}Code language: CSS (css)L’interfaccia descrive la struttura di ogni elemento che verrà inserito nella coda. Ora è il momento di utilizzarla per gestire il processo all’interno della coda. Prima di procedere, dobbiamo importare alcune dipendenze necessarie.
import type { queueAsPromised } from 'fastq';
import * as fastq from 'fastq';
import logger from 'logger.js';
import { setTimeout } from 'node:timers/promises';Code language: JavaScript (javascript)I primi due import sono utilizzati per la gestione della coda; il logger viene impiegato per mostrare alcune informazioni nel terminale durante l’esecuzione, mentre setTimeout simula il tempo di esecuzione di un servizio esterno (in questo caso, un finto server SMTP).
Ora che hai tutte le dipendenze necessarie, puoi creare il tuo worker.
async function userInsertHandler(arg: UserCreatedTask) {
logger.info(arg, 'User created task received');
const fakeImplementation = Math.random() > 0.8 ? 'success' : 'error'
const timeout = fakeImplementation === 'success' ? 2000 : 1000;
await setTimeout(timeout);
if (fakeImplementation === 'error') {
throw new Error(`User created task got error with id: ${arg.id}`);
}Code language: JavaScript (javascript)Come puoi vedere, questa funzione è semplice, ma ti fornisce tutte le basi necessarie per gestire una coda con fastq.
Per prima cosa, l’argomento è di tipo UserCreatedTask, il che significa che l’elemento nella coda deve rispettare questa struttura.
Successivamente, hai simulato un risultato di successo o errore utilizzando un semplice Math.random(). Se l’operazione ha esito positivo, l’esecuzione attenderà 2 secondi prima di risolvere la Promise; in caso contrario, attenderà 1 secondo prima di generare un errore.
Semplice, vero? Ora devi associare questo worker a una vera coda fastq. Per farlo, inserisci questo codice
const queue: queueAsPromised<UserCreatedTask> = fastq.promise(userInsertHandler, 1);
export default queue;Code language: JavaScript (javascript)Utilizzando il metodo fastq.promise, crei la tua coda con fastq e associ il worker userInsertHandler alla coda.
Un dettaglio importante: il secondo parametro definisce il numero di worker concorrenti che vuoi avere. In questo caso, la coda ha un solo worker, ma puoi aumentarlo a seconda delle tue esigenze.
Infine, l’ultima riga esporta la coda al di fuori di questo modulo JS, permettendoti di aggiungere elementi alla coda in qualsiasi punto del tuo codice.
Inserire dati nella coda
L’ultimo tassello del puzzle riguarda l’inserimento dei dati nella coda.
Per farlo, devi importare la coda nel modulo consumer e chiamare il metodo push.
Ora, spostati nel file src/index.ts e importa la coda.
import userInsertQueue from './queue.js';Code language: JavaScript (javascript)Ora creerai un metodo di esecuzione fittizio che genererà ID utente casuali e li inserirà nella coda.
async function run() {
const usersToHandle = new Array(5).fill(undefined).map(() => randomUUID())
for (const id of usersToHandle) {
const task = {
id
}
userInsertQueue.push(task)
.then(() => {
logger.info(task, `Task with id ${task.id} has been completed`);
})
.catch((error) => {
logger.error(task, `Task with id ${task.id} has failed`, error);
})
logger.info(task, `Task with id ${task.id} has been pushed`);
}
}
run()Code language: JavaScript (javascript)Come puoi vedere, il metodo push restituisce una promise, quindi puoi gestire il risultato per capire se il task è stato completato con successo o rigettato.
Tieni a mente questo dettaglio, perché tornerà utile quando parleremo della strategia di retry.
Ora torniamo all’esempio. Nel metodo run, creiamo una lista di ID utente casuali utilizzando il metodo randomUUID. Poi, iteriamo la lista e inseriamo ogni elemento nella coda. Utilizzando i metodi then e catch, registriamo nel log il risultato dell’esecuzione per ogni ID.
Perfetto! Ora è il momento di vedere il risultato in azione. Apri il terminale ed esegui il comando: npm run start.
Dovresti vedere un output simile a questo nel terminale:
[14:35:13.067] INFO (59175): User created task received
id: "a966759c-f385-4651-a0c1-55742ceb4017"
[14:35:13.067] INFO (59175): Task with id a966759c-f385-4651-a0c1-55742ceb4017 has been pushed
id: "a966759c-f385-4651-a0c1-55742ceb4017"
[14:35:13.067] INFO (59175): Task with id 24d9d735-2399-471e-b64c-55ee0c769daa has been pushed
id: "24d9d735-2399-471e-b64c-55ee0c769daa"
[14:35:13.067] INFO (59175): Task with id 726b4bc5-794d-4581-a96c-e41a474c446d has been pushed
id: "726b4bc5-794d-4581-a96c-e41a474c446d"
[14:35:13.067] INFO (59175): Task with id 1a6dd1b4-2123-410b-a2ad-f420a14d53e9 has been pushed
id: "1a6dd1b4-2123-410b-a2ad-f420a14d53e9"
[14:35:13.067] INFO (59175): Task with id 6a8681e4-bba4-42e0-b484-976de8be9bcb has been pushed
id: "6a8681e4-bba4-42e0-b484-976de8be9bcb"
[14:35:14.069] INFO (59175): User created task received
id: "24d9d735-2399-471e-b64c-55ee0c769daa"
[14:35:14.070] ERROR (59175): Task with id a966759c-f385-4651-a0c1-55742ceb4017 has failed
id: "a966759c-f385-4651-a0c1-55742ceb4017"
[14:35:15.070] INFO (59175): User created task received
id: "726b4bc5-794d-4581-a96c-e41a474c446d"
[14:35:15.071] ERROR (59175): Task with id 24d9d735-2399-471e-b64c-55ee0c769daa has failed
id: "24d9d735-2399-471e-b64c-55ee0c769daa"
[14:35:16.071] INFO (59175): User created task received
id: "1a6dd1b4-2123-410b-a2ad-f420a14d53e9"
[14:35:16.071] ERROR (59175): Task with id 726b4bc5-794d-4581-a96c-e41a474c446d has failed
id: "726b4bc5-794d-4581-a96c-e41a474c446d"
[14:35:17.072] INFO (59175): User created task received
id: "6a8681e4-bba4-42e0-b484-976de8be9bcb"
[14:35:17.073] ERROR (59175): Task with id 1a6dd1b4-2123-410b-a2ad-f420a14d53e9 has failed
id: "1a6dd1b4-2123-410b-a2ad-f420a14d53e9"
[14:35:19.074] INFO (59175): Task with id 6a8681e4-bba4-42e0-b484-976de8be9bcb has been completed
id: "6a8681e4-bba4-42e0-b484-976de8be9bcb"Code language: JavaScript (javascript)Verificando i risultati, puoi notare che alcuni task sono falliti mentre altri sono stati completati con successo. Quindi, come puoi intuire, fastq non offre una strategia di retry integrata, il che significa che i task falliti rimangono tali alla fine dell’esecuzione.
Come puoi implementare una strategia di retry?
In questo esempio, possiamo utilizzare una mappa per salvare lo stato di ogni task e il numero di tentativi effettuati.
Vediamo l’implementazione:
import logger from 'logger.js';
import { randomUUID } from 'node:crypto';
import userInsertQueue from './queue.js';
type Status = 'not-handle' | 'error' | 'success'
async function run() {
const usersToHandle = new Array(5).fill(undefined).map(() => randomUUID())
const userProcessStatus: Record<string, { status: Status, retryValue: number }> = {}
for (const id of usersToHandle) {
userProcessStatus[id] = {
status: 'not-handle',
retryValue: 0
};
const task = {
id
}
const pushMessage = () => userInsertQueue.push(task)
.then(() => {
logger.info(`Task with id ${task.id} has been proceed`);
userProcessStatus[id].status = 'success';
})
.catch(err => {
const state = userProcessStatus[id]
logger.error(err, `Task with id ${task.id} got error after retry ${state.retryValue}`);
state.status = 'error';
if (state.retryValue < 3) {
state.retryValue++
pushMessage()
}
})
pushMessage()
logger.info(task, `Task with id ${task.id} has been pushed`);
}
process.on('exit', () => {
for (const [id, { status, retryValue }] of Object.entries(userProcessStatus)) {
logger.info(`Id ${id} is ${status} after ${retryValue} retries`);
}
})
}
run()Code language: JavaScript (javascript)In questo approccio, la mappa userProcessStatus contiene gli ID utente come chiavi e un oggetto con due campi come valore: lo stato dell’utente e il numero di tentativi di retry effettuati per quell’ID.
Come funziona?
Utilizzando il metodo pushMessage, creiamo una funzione che gestisce questa mappa:
1️⃣ Per prima cosa, l’elemento viene inserito nella coda e si attende il risultato.
2️⃣ Se l’operazione ha successo, l’elemento viene segnato come completato nella mappa.
3️⃣ Se l’operazione fallisce, lo stato viene impostato su errore. Se il numero di retry è inferiore a tre, il contatore viene aumentato e il pushMessage viene richiamato per riprovare l’elaborazione dell’elemento.
🔍 Debugging extra: Ho anche aggiunto un hook prima della chiusura del processo per riepilogare i risultati dei task, utile per il debug.
Ora puoi eseguire nuovamente npm run start e osservare i risultati: noterai che il programma tenta di elaborare nuovamente i task falliti prima di terminare.
Conclusioni
È il momento di concludere questo articolo.
In questo post hai imparato:
✅ Come creare il tuo worker
✅ Come creare una coda con fastq e associarla al worker
✅ Come inserire elementi nella coda
✅ Come costruire una strategia di retry per gli elementi nella coda
Pro e Contro
Questa soluzione è utile in alcuni scenari ed è molto semplice da implementare, ma presenta alcune sfide da gestire:
- Strategia di retry: Se ne hai bisogno, devi implementarla manualmente.
- Gestione della memoria: Tutto viene mantenuto in memoria, quindi devi trovare un modo per salvare lo stato della tua entità. Altrimenti, se il processo si interrompe, perderai tutti i dati della tua coda.
E con questo, abbiamo terminato! Spero che l’articolo ti sia stato utile e ci vediamo presto per il prossimo capitolo.
