Tâches longues en Node.js : traiter vos tâches en préservant l'expérience utilisateur avec BullMQ

Les applications web embarquent de plus en plus de tâches qui prennent un certain temps à être traitées. On parle de plusieurs secondes, voire plusieurs minutes. Et c'est d'autant plus vrai depuis l'arrivée de l'IA dans les services web : OCR, analyse IA, conversations avec des LLM...

Deux questions se posent alors : comment traiter ces tâches sans bloquer le serveur ? Et comment informer l'utilisateur de leur avancement en temps réel ?

Chez Lonestone, on a adopté un système de queues (BullMQ) couplé à des Server-Sent Events (SSE). On vous détaille l'architecture, les choix techniques, et on livre un exemple complet avec NestJS côté API et React côté front.

Que vos applications embarquent ou non de l'IA, cette architecture s'applique à bien des cas d'usage.

Dans ce premier article, nous aborderons les queues, leur utilité, les solutions disponibles et comment les mettre en place dans NestJS. Dans un second article, nous passerons aux Server-Sent Events, pour renvoyer l'information aux clients qui consomment notre API.

Le problème des tâches longues

Pourquoi l'utilisation d'un await côté API ne suffit plus

Dans la plupart des cas, lorsque l'on fait une requête HTTP classique vers notre API, un simple await suffit. Les opérations étant généralement I/O et rapides, Node.js les gère sans bloquer le thread principal.

Dès qu'on touche à des requêtes HTTP longues ou à des tâches intensives pour le CPU, on se retrouve confronté à des problèmes : timeout de la requête, couplage fort, scalabilité de l'application... Et en prime, l'expérience utilisateur se dégrade, puisque l'utilisateur reste dans le flou sur l'avancée du traitement.

Le couplage fort (tight coupling) entre la requête HTTP et le traitement lourd est la racine du problème. Il faut donc découpler, en déléguant le traitement à un processus en arrière-plan pour répondre immédiatement au client.

Deux problèmes distincts à résoudre

Chaque problème a sa propre réponse technique :

1. Exécuter la tâche sans bloquer le serveur ni la requête, c'est le rôle du système de queues.
2. Notifier le client de l'avancée et du résultat, c'est le rôle des Server-Sent Events (SSE).

Concentrons-nous d'abord sur le système de queues pour traiter les tâches.

Les systèmes de queues

Qu'est-ce qu'une job queue ?

Une queue (ou file d'attente pour ceux qui ont horreur des anglicismes) est une liste de jobs (de tâches) en attente de traitement. Les jobs arrivent un par un dans la file, puis sont distribués à des workers qui les traitent en parallèle.

flowchart LR
	A@{ shape: processes, label: "Jobs" } --> B@{ "shape": "database", label: "Queue" }
	B --> C@{ shape: lin-rect, label: "Worker 1" }
	B --> D@{ shape: lin-rect, label: "Worker 2" }
	B --> E@{ shape: lin-rect, label: "Worker 3" }

Le code qui reçoit la requête HTTP ajoute un job dans la queue et répond immédiatement au client. En arrière-plan, un ou plusieurs workers récupèrent les jobs et les traitent à leur rythme, indépendamment du cycle requête/réponse HTTP.

Ce découplage règle déjà un de nos problèmes, le couplage fort.

BullMQ vs RabbitMQ : comment choisir

Reste à choisir comment ajouter un système de queue dans vos applications. Et comme pour tout en informatique, il existe plein d'options plus ou moins intéressantes. Ici, nous avons décidé de nous concentrer sur deux outils très populaires pour créer des systèmes de queues : BullMQ et RabbitMQ.

Les deux choix se valent. Si je devais résumer ma vie avec vous, je dirais que c'est d'abord des rencontres. Tout dépend de vos besoins, de la complexité de votre système, etc.

Si vous cherchez un système simple qui fonctionnera à la perfection dans une stack JavaScript ou TypeScript unifiée, alors BullMQ saura répondre à vos attentes. Mais si vous avez besoin d'un système plus agnostique, idéal pour une architecture micro-services, alors RabbitMQ saura sûrement vous séduire.

Pourquoi nous utilisons BullMQ avec NestJS

Chez Lonestone, notre stack technique nous pousse à partir sur BullMQ :

• Nous utilisons NestJS, et BullMQ s'y branche avec le package @nestjs/bullmq.
• Nos applications sont le plus souvent monolithiques ou modulaires.
• BullMQ reste simple à vivre, avec Redis comme base de données clé-valeur légère et une configuration rapide pour la concurrence ou la récurrence du nettoyage. Rien de bien sorcier.

Bien entendu, vous pouvez chercher d'autres solutions adaptées à votre contexte. Nous, en tout cas, on a porté notre dévolu sur BullMQ.

Place à un exemple : l'analyse du contenu d'un fichier

Faisons une pause dans la théorie, voulez-vous ? Pour rendre l'exemple plus digeste, on s'en tient au système de queues.

Prenons un cas rencontré chez Lonestone, où l'on veut extraire d'un fichier des informations importantes.

L'architecture du système

Le système sépare la requête HTTP et l'analyse.

Lorsque l'utilisateur fait une requête POST vers l'API, il reçoit une réponse lui indiquant si la requête est un succès, et en parallèle nous ajoutons un job. Une fois dépilé par un worker, ce job lance l'analyse, indépendamment de la requête.

Le flux de notre système ressemble donc à ça pour le moment :

sequenceDiagram
    participant Client
    participant API
    participant Queue
    participant Worker

    Client->>API: POST /analyze
    API->>Queue: Ajoute un job
    API-->>Client: 200 OK

    Queue->>Worker: Traite le job
    Worker->>Worker: Étape 1 (extraction)
    Worker->>Worker: Étape 2 (analyse)

    Worker->>Worker: Terminé

Un peu de code : Implémentation d'une queue avec NestJS

L'exemple utilise NestJS avec @nestjs/bullmq pour les queues.

Ce code est une version simplifiée de l'exemple complet, disponible sur GitHub. Le dépôt suit la stack technique Lonestone, le découpage des modules et la configuration y sont donc plus fournis.

Initialisation du module et de la queue

La première étape est de configurer BullMQ pour NestJS, au niveau du module racine, avec la connexion Redis :

@Module({
  imports: [
    BullModule.forRoot({
      connection: {
      host: config.redis.host,
      port: config.redis.port,
      },
    }),
    //[...]
  ],
  controllers: [
    // [...]
  ],
  providers: [
    // [...]
  ],
})
export class AppModule {}

Ensuite, nous devons ajouter le BullModule dans le module concerné et enregistrer la queue avec un nom. Dans notre cas, le point d'entrée est le module AnalysisModule, qui câble le controller, le service et le processor :

import { BullModule } from '@nestjs/bullmq'
import { Module } from '@nestjs/common'
import { AnalysisController } from './analysis.controller'
import { ANALYSIS_QUEUE_NAME, AnalysisProcessor } from './analysis.processor'
import { AnalysisService } from './analysis.service'

@Module({
  imports: [
    BullModule.registerQueue({ name: ANALYSIS_QUEUE_NAME }),
  ],
  controllers: [AnalysisController],
  providers: [AnalysisService, AnalysisProcessor],
})
export class AnalysisModule {}

Deux éléments comptent pour notre système de queue :

• BullModule.registerQueue({ name: ANALYSIS_QUEUE_NAME }), le nom étant libre.
• AnalysisProcessor, le worker qui exécute le job.

Traiter la requête envoyée par l'utilisateur

Notre AnalysisController contient la route qui lance l'analyse.

import { Controller, HttpCode, HttpStatus, Param, Post } from '@nestjs/common'

@Controller('analysis')
export class AnalysisController {
  constructor(
    private readonly analysisService: AnalysisService,
  ) {}

  @Post('/:id/analyze')
  @HttpCode(HttpStatus.OK)
  async startAnalyze(@Param('id') id: string) {
    return this.analysisService.startAnalyze(id)
  }
}

Cette route appelle la méthode startAnalyze de notre service AnalysisService.

Le service est minimaliste. Il reçoit un identifiant d'analyse et ajoute un job dans la queue BullMQ :

import { InjectQueue } from '@nestjs/bullmq'
import { Injectable } from '@nestjs/common'
import { Queue } from 'bullmq'

@Injectable()
export class AnalysisService {
  constructor(
    @InjectQueue(ANALYSIS_QUEUE_NAME) private readonly analysisQueue: Queue,
  ) {}

  async startAnalyze(id: string) {
    this.analysisQueue.add(ANALYSIS_JOB_NAME, { analysisId: id })
  }
}

L'appel à add() est non bloquant : le job est placé dans Redis et le service retourne immédiatement. Le traitement effectif se fera dans le worker.

Le processor (worker) : traiter le job

Le processor effectue le travail lourd. Il hérite de WorkerHost, la classe de @nestjs/bullmq qui attend une méthode process chargée de traiter le job. C'est ici qu'on placera les tâches longues que l'on veut séparer de notre requête principale :

import { Processor, WorkerHost } from '@nestjs/bullmq'
import { Job } from 'bullmq'

export const ANALYSIS_QUEUE_NAME = 'analysis_queue'
export const ANALYSIS_JOB_NAME = 'analysis_job'
export const ANALYSIS_JOBS_CONCURRENCY = 10

@Processor(ANALYSIS_QUEUE_NAME, {
  concurrency: ANALYSIS_JOBS_CONCURRENCY,
  removeOnComplete: { age: 3600, count: 1000 },
  removeOnFail: { age: 24 * 3600 },
})
export class AnalysisProcessor extends WorkerHost {
  constructor() {
    super()
  }

  async process(job: Job<AnalysisJobData>) {
    await performExtraction(job.data) // Tâche longue
    await performAnalysis(job.data) // Tâche longue
  }
}

Le décorateur @Processor porte deux réglages importants :

• Avec concurrency: 10, ce worker traite jusqu'à 10 jobs en parallèle.
• removeOnComplete et removeOnFail nettoient automatiquement les jobs terminés ou échoués, ce qui garde la taille de Redis sous contrôle.

Une petite partie côté front avec React

Déclencher l'analyse

Côté front, on crée un hook de mutation simple, qui appelle l'API pour lancer l'analyse. Le retour est immédiat (puisque le serveur ajoute le job en queue et répond tout de suite). Les mises à jour de progression arriveront via les SSE, sujet du deuxième article.

import { useMutation } from '@tanstack/react-query'

export function useStartAnalysis() {
  return useMutation({
    mutationFn: async ({ id }: { id: string }) => {
      const response = await analysisControllerStartAnalyze({
        path: { id },
      })
      if (response.error) throw new Error(response.error as string)
      return response.data
    },
  })
}

Nous utilisons un SDK auto-généré pour nos routes, d'où l'appel à analysisControllerStartAnalyze pour notre requête POST sur l'endpoint /:id/analyze.

Lancer une analyse

Dans notre exemple, nous avons un dashboard qui liste quelques « analyses ». L'interface utilise le hook et affiche une carte par analyse :

export default function DashboardPage() {
  const { data: analyses } = useAnalyses()
  const { mutate } = useStartAnalysis()

  return (
    <main className="container mx-auto py-8 px-4 space-y-6">
      <h1 className="text-3xl font-bold">Analyses</h1>
      {analyses?.map((analysis) => (
        <Card key={analysis.id}>
          <CardHeader>
            <CardTitle>Analysis</CardTitle>
            <AnalysisBadge step={analysis.step} />
          </CardHeader>
          <CardFooter>
            <Button
              onClick={() => mutate({ id: analysis.id })}
              disabled={
                analysis.step !== 'completed' && analysis.step !== 'failed'
              }
            >
              Lancer l'analyse
            </Button>
          </CardFooter>
        </Card>
      ))}
    </main>
  )
}

On peut donc lancer une analyse pour chaque « analyse » (désolé pour la logique métier, on a déjà vu des règles métier plus intelligentes que celle-là).

Mais si vous avez suivi les étapes et lancé une analyse, vous avez sûrement remarqué qu'il manque le retour sur son avancée, et donc la confirmation qu'elle est terminée. La seule chose que nous savons pour le moment, c'est que l'analyse est lancée, puisque nous recevons une réponse HTTP 200 pour notre requête POST.

En résumé

Pour traiter des tâches lourdes en Node.js, nous utilisons BullMQ, qui rend l'architecture simple :

• BullMQ gère le traitement en arrière-plan via Redis, avec concurrence, retries et nettoyage automatique.
• La requête HTTP retourne immédiatement un statut 200, sans attendre la fin du traitement.
• Les workers traitent les jobs de façon asynchrone, indépendamment du cycle requête/réponse.

Ce système s'adapte à de nombreux cas concrets : analyse par IA, traitement OCR, génération de documents, import de données, envoi d'emails en masse…

Pour l'instant, nous répondons aux besoins de performance et nous réduisons le couplage. Mais une question subsiste : comment prévenir les utilisateurs de l'état d'avancement du traitement ?

C'est précisément l'objet de notre deuxième article (nous sommes persuadés que vous ne l'aviez pas vu venir) : comment utiliser les Server-Sent Events (SSE) et des principes d'architecture Event-Driven dans NestJS pour tenir les utilisateurs informés en temps réel !

Si ce genre d'architecture doit atterrir dans votre produit, on peut le faire avec vous chez Lonestone. L'offre Build couvre le développement du produit sprint après sprint, et quand une équipe technique existe déjà en interne, Scale lui apporte le renfort de développeurs qui ont déjà mis ces mécaniques en production.

Retrouvez la suite dans notre deuxième article sur le sujet, et le code complet de l'exemple (combinant BullMQ et SSE) sur GitHub.