Travaux en arrière-plan dans SvelteKit avec BullMQ

Votre application SvelteKit a-t-elle du mal à effectuer des tâches telles que l'envoi d'e-mails, le redimensionnement d'images ou le traitement de données ? Avec BullMQ, vous pouvez décharger ces tâches lourdes en arrière-plan et conserver votre application ultra-rapide. Dans cet article, nous allons vous montrer comment le configurer et aborder des tâches du monde réel comme un pro. Allons-y !

tl;dr Configurez les Workers BullMQ dans hooks.server.js. Découvrez l'exemple

Qu’est-ce que BullMQ ?

BullMQ est une bibliothèque Node.js permettant de créer et de gérer des files d'attente de tâches avec Redis. Il vous aide à exécuter efficacement des tâches fastidieuses en arrière-plan. Grâce à des fonctionnalités intégrées telles que les tentatives, la planification des tâches et le contrôle de la concurrence, BullMQ rend la gestion des flux de travail complexes dans votre application simple et fiable.

Étape 1 : Installer les dépendances

Tout d'abord, installez ioredis (client Redis pour node.js) et bullmq :

pnpm i -D ioredis bullmq

Même si vous pouvez ajouter des tâches à une file d'attente bullmq à partir d'un environnement sans serveur comme Vercel, les travailleurs doivent s'exécuter sur un serveur node.js traditionnel de longue durée. Par conséquent, remplacez adapter-auto par adapter-node :

pnpm rm @sveltejs/adapter-auto && pnpm i -D @sveltejs/adapter-node

N'oubliez pas de mettre à jour votre configuration Svelte (svelte.config.js) avec l'adaptateur de nœud nouvellement installé.

Étape 2 : configuration de la file d'attente des tâches et du processeur

Ensuite, configurons une file d'attente de tâches BullMQ et son processeur. Créez un fichier .js dans le répertoire src/lib/server/ :

// src/lib/server/background-jobs.js

import { REDIS_URL } from "$env/static/private";
import { Queue, Worker } from "bullmq";
import IORedis from "ioredis";

const Q_NAME = "q";

export const jobsQueue = new Queue(Q_NAME, {
  connection: new IORedis(REDIS_URL),
});

const sleep = (t) => new Promise((resolve) => setTimeout(resolve, t * 100));

export const setupBullMQProcessor = () => {
  new Worker(
    Q_NAME,
    async (job) => {
      for (let i = 0; i <= 100; i++) {
        await sleep(Math.random());
        await job.updateProgress(i);
        await job.log(`Processing job at interval ${i}`);

        if (Math.random() * 200 < 1) throw new Error(`Random error at ${i}`);
      }

      return `This is the return value of job (${job.id})`;
    },
    // https://docs.bullmq.io/bull/patterns/persistent-connections#maxretriesperrequest
    { connection: new IORedis(REDIS_URL, { maxRetriesPerRequest: null }) }
  );
};

Ici, nous avons également créé une fonction utilitaire qui instancie un travailleur BullMQ pour écouter et traiter les tâches dans la file d'attente Q_NAME.

Nous devons appeler cette fonction dans notre fichier hooks.server.js, soit au niveau supérieur, soit dans le hook d'initialisation.

// src/hooks.server.js

// ...
import { building } from "$app/environment";
import { setupBullMQProcessor } from "$lib/server/background-jobs";
// ...
if (!building) {
  setupBullMQProcessor();
}
// ...

La vérification !building ignore la configuration du travailleur (et, par conséquent, une connexion Redis) pendant la construction, accélérant ainsi le processus.

? BullMQ est maintenant prêt à être utilisé dans notre application SvelteKit ?

Heure de démonstration

Afin de tester la configuration, créons un point de terminaison POST pour mettre une tâche en file d'attente.

// src/routes/+server.ts

import { jobsQueue } from "$lib/server/background-jobs";

export const POST = async () => {
  const { id: jobId } = await jobsQueue.add("job", {});

  /*
  The following code passes the job's progress to the client as a stream.
  If you don't need to update the client with the progress, you can skip
  the following. You can also use web-sockets or polling for that.
  */
  const stream = new ReadableStream({
    async pull(controller) {
      const job = await jobsQueue.getJob(jobId);
      controller.enqueue(
        JSON.stringify(
          job.failedReason
            ? { error: job.failedReason }
            : job.returnvalue
            ? { data: job.returnvalue }
            : { progress: job.progress }
        )
      );
      controller.enqueue("\n");

      if (job.finishedOn) {
        controller.close();
      }

      // wait for 1-second before sending the next status update
      await new Promise((r) => setTimeout(r, 1e3));
    },
  });

  return new Response(stream, {
    headers: { "content-type": "text/plain" },
  });
};

Et sur le frontend, ajoutons un bouton pour déclencher le point de terminaison ci-dessus et afficher ensuite l'état de la tâche :

<!-- src/routes/+page.svelte -->

<script>
  let result = $state();

  $inspect(result);

  const handleClick = async () => {
    const response = await fetch("/", { method: "post" });
    const reader = await response.body.getReader();
    while (true) {
      const { done, value } = await reader.read();
      if (done) break;
      result = JSON.parse(new TextDecoder().decode(value));
    }

    setTimeout(() => (result = undefined), 3e3);
  };
</script>

{#if result?.error}
  <div>



<p>Here is the output:</p>

<p><img src="https://img.php.cn/upload/article/000/000/000/173785153552047.jpg" alt="background jobs demo using sveltekit and bullmq" loading="lazy"    style="max-width:90%"  style="max-width:90%" data-animated="true"></p>


<hr>

<h3>
  
  
  Bonus ?
</h3>

<p>You can also mount a bull-board dashboard in your SvelteKit app for easy monitoring of background jobs.</p>

<p>Install bull-board dependencies<br>
</p>

<pre class="brush:php;toolbar:false">pnpm i -D @bull-board/api @bull-board/hono @hono/node-server hono

et modifiez votre hooks.server.js :

// src/hooks.server.js

import { building } from "$app/environment";
import { jobsQueue, setupBullMQProcessor } from "$lib/server/background-jobs";
import { createBullBoard } from "@bull-board/api";
import { BullMQAdapter } from "@bull-board/api/bullMQAdapter";
import { HonoAdapter } from "@bull-board/hono";
import { serveStatic } from "@hono/node-server/serve-static";
import { Hono } from "hono";

if (!building) {
  setupBullMQProcessor();
}

const bullboard = (() => {
  const serverAdapter = new HonoAdapter(serveStatic);

  createBullBoard({
    queues: [new BullMQAdapter(jobsQueue)],
    serverAdapter,
  });
  const app = new Hono({ strict: false });
  const basePath = "/jobs";
  serverAdapter.setBasePath(basePath);
  app.route(basePath, serverAdapter.registerPlugin());

  return app;
})();

export const handle = async ({ event, resolve }) => {
  if (event.url.pathname.match(/^\/jobs($|\/)/)) {
    return bullboard.fetch(event.request);
  }

  return resolve(event);
};

Ensuite, visitez /jobs pour voir le tableau de bord du bull-board

Ce qui précède est le contenu détaillé de. pour plus d'informations, suivez d'autres articles connexes sur le site Web de PHP en chinois!