Hoe om te gaan met asynchrone taken met Node.js en BullMQ

De auteur heeft de Vereniging van Vrouwelijke Ingenieurs geselecteerd om een donatie te ontvangen als onderdeel van het Write for DOnations-programma.

Introductie

Webapplicaties hebben verzoek-/reactiecycli. Wanneer je een URL bezoekt, stuurt de browser een verzoek naar de server waarop een app draait die gegevens verwerkt of vragen in de database uitvoert. Tijdens dit proces moet de gebruiker wachten totdat de app een reactie terugstuurt. Voor sommige taken kan de gebruiker snel een reactie krijgen; voor tijdsintensieve taken, zoals het verwerken van afbeeldingen, het analyseren van gegevens, het genereren van rapporten of het verzenden van e-mails, duren deze taken lang en kunnen ze de verzoek-/reactiecyclus vertragen. Stel bijvoorbeeld dat je een applicatie hebt waarin gebruikers afbeeldingen uploaden. In dat geval moet je mogelijk de afbeelding verkleinen, comprimeren of converteren naar een ander formaat om de schijfruimte van je server te behouden voordat je de afbeelding aan de gebruiker laat zien. Het verwerken van een afbeelding is een CPU-intensieve taak, die een Node.js-thread kan blokkeren totdat de taak is voltooid. Dat kan enkele seconden of minuten duren. Gebruikers moeten wachten totdat de taak is voltooid om een reactie van de server te krijgen.

Om het verzoek-/responscyclus niet te vertragen, kunt u bullmq gebruiken, een gedistribueerde taak (job) wachtrij die u toelaat om tijdrovende taken uit uw Node.js app te offloaden naar bullmq, waardoor de verzoek-/responscyclus vrijkomt. Deze tool stelt uw app in staat om snel op gebruikers te reageren terwijl bullmq de taken asynchroon op de achtergrond en onafhankelijk van uw app uitvoert. Om de taken bij te houden, gebruikt bullmq Redis om een korte beschrijving van elke taak in een wachtrij op te slaan. Een bullmq worker haalt vervolgens elke taak in de wachtrij uit en voert deze uit, en markeert deze als voltooid zodra dit gedaan is.

In dit artikel zult u bullmq gebruiken om een tijdrovende taak op de achtergrond te plaatsen, wat een applicatie in staat zal stellen om snel op gebruikers te reageren. Eerst zult u een app creëren met een tijdrovende taak zonder gebruik te maken van bullmq. Daarna zult u bullmq gebruiken om de taak asynchroon uit te voeren. Ten slotte zult u een visueel dashboard installeren om bullmq taken in een Redis wachtrij te beheren.

Vereisten

Om deze tutorial te volgen, heeft u het volgende nodig:

• Node.js ontwikkelomgeving instellen. Voor Ubuntu 22.04, volg onze tutorial over Hoe Node.js te installeren op Ubuntu 22.04. Voor andere systemen, zie Hoe Node.js te installeren en een lokale ontwikkelomgeving te maken.

• Redis geïnstalleerd op uw systeem. Op Ubuntu 22, volg Stappen 1 tot 3 in onze tutorial over Hoe Redis te installeren en te beveiligen op Ubuntu 22.04. Voor andere systemen, zie onze tutorial over Hoe Redis te installeren en te beveiligen.

• Bekendheid met promises en async/await functies, die u kunt ontwikkelen in onze tutorial Begrip van de Event Loop, Callbacks, Promises en Async/Await in JavaScript.

• Basiskennis van hoe Express te gebruiken. Zie onze tutorial over Aan de slag met Node.js en Express.

• Vertrouwdheid met Embedded JavaScript (EJS). Bekijk onze tutorial over Hoe EJS te gebruiken om je Node-applicatie te templaten voor meer details.

• Basisbegrip van hoe afbeeldingen te verwerken met sharp, wat je kunt leren in onze tutorial over Hoe afbeeldingen te verwerken in Node.js met Sharp.

Stap 1 — Het instellen van het projectdirectory

In deze stap maak je een directory aan en installeer je de benodigde afhankelijkheden voor je toepassing. De toepassing die je in deze handleiding zult bouwen, stelt gebruikers in staat om een afbeelding te uploaden, die vervolgens wordt verwerkt met behulp van het sharp-pakket. Beeldverwerking kost veel tijd en kan de aanvraag-/responscyclus vertragen, waardoor de taak een goede kandidaat is om in de achtergrond te worden uitgevoerd met bullmq. De techniek die je zult gebruiken om de taak uit te schakelen, werkt ook voor andere tijdrovende taken.

Begin met het aanmaken van een directory genaamd image_processor en navigeer naar de directory:

mkdir image_processor && cd image_processor

Vervolgens initialiseer je de directory als een npm-pakket:

npm init -y

De opdracht maakt een package.json-bestand aan. De -y-optie vertelt npm om alle standaardinstellingen te accepteren.

Na het uitvoeren van de opdracht komt je uitvoer overeen met het volgende:

Output
Wrote to /home/sammy/image_processor/package.json:

{
  "name": "image_processor",
  "version": "1.0.0",
  "description": "",
  "main": "index.js",
  "scripts": {
    "test": "echo \"Error: no test specified\" && exit 1"
  },
  "keywords": [],
  "author": "",
  "license": "ISC"
}

De uitvoer bevestigt dat het package.json-bestand is aangemaakt. Belangrijke eigenschappen zijn onder andere de naam van je app (name), het versienummer van je toepassing (version) en het startpunt van je project (main). Als je meer wilt weten over de andere eigenschappen, kun je de documentatie van npm’s package.json raadplegen.

De toepassing die je zult bouwen in deze tutorial zal de volgende afhankelijkheden vereisen:

• express: een webframework voor het bouwen van web-apps.
• express-fileupload: een middleware waarmee je bestanden kunt uploaden via je formulieren.
• sharp: een bibliotheek voor beeldverwerking.
• ejs: een sjabloon taal waarmee je HTML-markering kunt genereren met behulp van Node.js.
• bullmq: een gedistribueerde taakwachtrij.
• bull-board: een dashboard dat voortbouwt op bullmq en de status van taken weergeeft met een gebruiksvriendelijke interface (UI).

Om al deze afhankelijkheden te installeren, voer je de volgende opdracht uit:

npm install express express-fileupload sharp ejs bullmq @bull-board/express

Naast de geïnstalleerde afhankelijkheden, zul je ook de volgende afbeelding gebruiken later in deze tutorial:

Gebruik curl om de afbeelding naar de locatie van jouw keuze op je lokale computer te downloaden

curl -O https://deved-images.nyc3.cdn.digitaloceanspaces.com/CART-68886/underwater.png

Je hebt nu de nodige afhankelijkheden om een Node.js-app te bouwen zonder bullmq, wat je vervolgens zult doen.

Stap 2 — Implementatie van een Tijdsintensieve Taak Zonder bullmq

In deze stap zul je een applicatie bouwen met Express waarmee gebruikers afbeeldingen kunnen uploaden. De app zal een tijdsintensieve taak starten met behulp van sharp om de afbeelding naar meerdere formaten te verkleinen, die vervolgens aan de gebruiker worden getoond nadat een reactie is verzonden. Deze stap zal je helpen begrijpen hoe tijdsintensieve taken van invloed zijn op de request/response-cyclus.

Gebruik nano, of je voorkeursteksteditor, om het bestand index.js aan te maken:

nano index.js

Voeg in je index.js-bestand de volgende code toe om afhankelijkheden te importeren:

const path = require("path");
const fs = require("fs");
const express = require("express");
const bodyParser = require("body-parser");
const sharp = require("sharp");
const fileUpload = require("express-fileupload");

In de eerste regel importeer je de path-module voor het berekenen van bestandspaden met Node. In de tweede regel importeer je de fs-module voor interactie met mappen. Vervolgens importeer je het express-webframework. Je importeert de body-parser-module om middleware toe te voegen om gegevens in HTTP-verzoeken te parsen. Daarna importeer je de sharp-module voor afbeeldingsverwerking. Tenslotte importeer je express-fileupload voor het verwerken van uploads vanuit een HTML-formulier.

Voeg vervolgens de volgende code toe om middleware te implementeren in je app:

...
const app = express();
app.set("view engine", "ejs");
app.use(bodyParser.json());
app.use(
  bodyParser.urlencoded({
    extended: true,
  })
);

Eerst stel je de variabele app in op een instantie van Express. Ten tweede, door gebruik te maken van de variabele app, configureer je met de set() methode Express om de ejs template taal te gebruiken. Vervolgens voeg je de body-parser module middleware toe met de use() methode om JSON-gegevens in HTTP-verzoeken om te zetten in variabelen die met JavaScript kunnen worden benaderd. In de volgende regel doe je hetzelfde met URL-gecodeerde invoer.

Voeg vervolgens de volgende regels toe om meer middleware toe te voegen om bestandsuploads te verwerken en statische bestanden te serveren:

...
app.use(fileUpload());
app.use(express.static("public"));

Je voegt middleware toe om geüploade bestanden te analyseren door de fileUpload() methode aan te roepen, en je stelt een directory in waar Express zal kijken en statische bestanden zal serveren, zoals afbeeldingen en CSS.

Met de middleware ingesteld, maak een route aan die een HTML-formulier weergeeft voor het uploaden van een afbeelding:

...
app.get("/", function (req, res) {
  res.render("form");
});

Hier gebruik je de get() methode van de Express module om de / route en de callback te specificeren die moet worden uitgevoerd wanneer de gebruiker de homepage of de / route bezoekt. In de callback roep je res.render() aan om het form.ejs bestand in de views directory te renderen. Je hebt nog niet het form.ejs bestand of de views directory aangemaakt.

Om dit te maken, sla je eerst je bestand op en sluit je het. Voer vervolgens in je terminal het volgende commando in om de views directory in je project root directory aan te maken:

mkdir views

Ga naar de views directory:

cd views

Maak het form.ejs bestand aan in je editor:

nano form.ejs

In je form.ejs bestand, voeg de volgende code toe om het formulier te maken:

<!DOCTYPE html>
<html lang="en">
  <%- include('./head'); %>
  <body>
    <div class="home-wrapper">
      <h1>Image Processor</h1>
      <p>
        Resizes an image to multiple sizes and converts it to a
        <a href="https://en.wikipedia.org/wiki/WebP">webp</a> format.
      </p>
      <form action="/upload" method="POST" enctype="multipart/form-data">
        <input
          type="file"
          name="image"
          placeholder="Select image from your computer"
        />
        <button type="submit">Upload Image</button>
      </form>
    </div>
  </body>
</html>

Eerst verwijs je naar het bestand head.ejs, dat je nog niet hebt aangemaakt. Het bestand head.ejs zal het HTML-element head bevatten waarnaar je kunt verwijzen in andere HTML-pagina’s.

In de body-tag maak je een formulier met de volgende attributen:

• action specificeert de route waar de formuliergegevens naartoe moeten worden gestuurd wanneer het formulier wordt ingediend.
• method specificeert de HTTP-methode voor het verzenden van gegevens. De POST-methode incorporeert de gegevens in een HTTP-verzoek.
• encytype specificeert hoe de formuliergegevens moeten worden gecodeerd. De waarde multipart/form-data maakt het mogelijk voor HTML-input-elementen om bestandsgegevens te uploaden.

In het form-element maak je een input-tag om bestanden te uploaden. Vervolgens definieer je het button-element met het attribuut type ingesteld op submit, waarmee je formulieren kunt verzenden.

Zodra je klaar bent, sla je het bestand op en sluit je het.

Vervolgens maak je een head.ejs-bestand aan:

nano head.ejs

In je head.ejs-bestand voeg je de volgende code toe om de kopsectie van de app te maken:

<head>
  <meta charset="UTF-8" />
  <meta http-equiv="X-UA-Compatible" content="IE=edge" />
  <meta name="viewport" content="width=device-width, initial-scale=1.0" />
  <title>Image Processor</title>
  <link rel="stylesheet" href="css/main.css" />
</head>

Hier verwijs je naar het bestand main.css, dat je later in deze stap zult aanmaken in de public-directory. Dat bestand zal de stijlen voor deze applicatie bevatten. Voor nu blijf je de processen voor statische assets instellen.

Sla het bestand op en sluit het.

Om gegevens die vanuit het formulier zijn verzonden te verwerken, moet je een post-methode definiëren in Express. Ga daarvoor terug naar de hoofdmap van je project:

cd ..

Open je index.js-bestand opnieuw:

nano index.js

Voeg in je index.js-bestand de gemarkeerde regels toe om een methode te definiëren voor het verwerken van formulierinzendingen op de route /upload:

app.get("/", function (req, res) {
  ...
});

app.post("/upload", async function (req, res) {
  const { image } = req.files;

  if (!image) return res.sendStatus(400);

});

Gebruik de app-variabele om de post()-methode aan te roepen, die de ingediende formulier op de route /upload zal verwerken. Vervolgens haal je de geüploade afbeeldingsgegevens uit het HTTP-verzoek en sla je deze op in de variabele image. Daarna stel je een reactie in om een statuscode 400 terug te geven als de gebruiker geen afbeelding uploadt.

Om het proces voor de geüploade afbeelding in te stellen, voeg je de volgende gemarkeerde code toe:

...
app.post("/upload", async function (req, res) {
  const { image } = req.files;

  if (!image) return res.sendStatus(400);
  const imageName = path.parse(image.name).name;
  const processImage = (size) =>
    sharp(image.data)
      .resize(size, size)
      .webp({ lossless: true })
      .toFile(`./public/images/${imageName}-${size}.webp`);

  sizes = [90, 96, 120, 144, 160, 180, 240, 288, 360, 480, 720, 1440];
  Promise.all(sizes.map(processImage));
});

Deze regels vertegenwoordigen hoe je app de afbeelding zal verwerken. Eerst verwijder je de afbeeldingsextensie van de geüploade afbeelding en sla je de naam op in de variabele imageName. Vervolgens definieer je de functie processImage(). Deze functie neemt de parameter size, waarvan de waarde zal worden gebruikt om de afmetingen van de afbeelding tijdens het verkleinen te bepalen. In de functie roep je sharp() aan met image.data, dat een buffer bevat met de binaire gegevens van de geüploade afbeelding. sharp verkleint de afbeelding volgens de waarde in de grootteparameter. Je gebruikt de methode webp() van sharp om de afbeelding naar het webp-beeldformaat om te zetten. Vervolgens sla je de afbeelding op in de map public/images/.

De daaropvolgende lijst met nummers bepaalt de formaten die zullen worden gebruikt om de geüploade afbeelding te herschalen. Vervolgens gebruik je de map()-methode van JavaScript om processImage() aan te roepen voor elk element in de array sizes, waarna het een nieuwe array zal retourneren. Elke keer dat de map()-methode de functie processImage() aanroept, retourneert het een belofte aan de nieuwe array. Je gebruikt de Promise.all()-methode om ze op te lossen.

Computer processing-snelheden variëren, evenals de grootte van afbeeldingen die een gebruiker kan uploaden, wat de snelheid van de afbeeldingsverwerking kan beïnvloeden. Om deze code voor demonstratiedoeleinden te vertragen, voeg je de gemarkeerde regels in om een CPU-intensieve increment loop toe te voegen en een doorverwijzing naar een pagina die de herschaalde afbeeldingen zal weergeven met de gemarkeerde regels:

...
app.post("/upload", async function (req, res) {
  ...
  let counter = 0;
  for (let i = 0; i < 10_000_000_000; i++) {
    counter++;
  }

  res.redirect("/result");
});

De loop zal 10 miljard keer worden uitgevoerd om de variabele counter te verhogen. Je roept de functie res.redirect() aan om de app door te verwijzen naar de route /result. De route zal een HTML-pagina renderen die de afbeeldingen in de map public/images zal weergeven.

De route /result bestaat nog niet. Voeg om deze te maken de gemarkeerde code toe aan je index.js-bestand:

...

app.get("/", function (req, res) {
 ...
});

app.get("/result", (req, res) => {
  const imgDirPath = path.join(__dirname, "./public/images");
  let imgFiles = fs.readdirSync(imgDirPath).map((image) => {
    return `images/${image}`;
  });
  res.render("result", { imgFiles });
});

app.post("/upload", async function (req, res) {
  ...
});

Je definieert de /result route met de app.get() methode. In de functie, definieer je de imgDirPath variabele met het volledige pad naar de public/images map. Je gebruikt de readdirSync() methode van de fs module om alle bestanden in de opgegeven map te lezen. Van daaruit keten je de map() methode om een nieuwe array te retourneren met de paden van de afbeeldingen voorafgegaan door images/.

Tenslotte roep je res.render() aan om het result.ejs bestand te renderen, dat nog niet bestaat. Je geeft de imgFiles variabele door, die een array bevat van alle relatieve paden van de afbeeldingen, naar het result.ejs bestand.

Sla je bestand op en sluit het.

Om het result.ejs bestand te maken, keer terug naar de views map:

cd views

Maak en open het result.ejs bestand in je editor:

nano result.ejs

In je result.ejs bestand, voeg de volgende regels toe om afbeeldingen weer te geven:

<!DOCTYPE html>
<html lang="en">
  <%- include('./head'); %>
  <body>
    <div class="gallery-wrapper">
      <% if (imgFiles.length > 0){%>
      <p>The following are the processed images:</p>
      <ul>
        <% for (let imgFile of imgFiles){ %>
        <li><img src=<%= imgFile %> /></li>
        <% } %>
      </ul>
      <% } else{ %>
      <p>
        The image is being processed. Refresh after a few seconds to view the
        resized images.
      </p>
      <% } %>
    </div>
  </body>
</html>

Eerst refereer je naar het head.ejs bestand. In de body tag, controleer je of de imgFiles variabele leeg is. Als het gegevens bevat, itereer je over elk bestand en maak je een afbeelding voor elk array element. Als imgFiles leeg is, print je een bericht dat de gebruiker vertelt om Na een paar seconden vernieuwen om de afbeeldingen te bekijken..

Sla je bestand op en sluit het.

Keer vervolgens terug naar de hoofdmap en maak de public map aan die je statische assets zal bevatten:

cd .. && mkdir public

Verplaats naar de map public:

cd public

Maak een map images aan waarin de geüploade afbeeldingen worden bewaard:

mkdir images

Vervolgens maak je de map css aan en navigeer je ernaartoe:

mkdir css && cd css

In je editor maak en open je het bestand main.css, dat je eerder hebt aangeroepen in het bestand head.ejs:

nano main.css

Voeg in je main.css bestand de volgende stijlen toe:

body {
  background: #f8f8f8;
}

h1 {
  text-align: center;
}

p {
  margin-bottom: 20px;
}

a:link,
a:visited {
  color: #00bcd4;
}

/** Stijlen voor de knop "Bestand kiezen" **/
button[type="submit"] {
  background: none;
  border: 1px solid orange;
  padding: 10px 30px;
  border-radius: 30px;
  transition: all 1s;
}

button[type="submit"]:hover {
  background: orange;
}

/** Stijlen voor de knop "Afbeelding uploaden" **/
input[type="file"]::file-selector-button {
  border: 2px solid #2196f3;
  padding: 10px 20px;
  border-radius: 0.2em;
  background-color: #2196f3;
}

ul {
  list-style: none;
  padding: 0;
  display: flex;
  flex-wrap: wrap;
  gap: 20px;
}

.home-wrapper {
  max-width: 500px;
  margin: 0 auto;
  padding-top: 100px;
}

.gallery-wrapper {
  max-width: 1200px;
  margin: 0 auto;
}

Deze regels zullen elementen in de app stylen. Met HTML-attributen stijl je de achtergrond van de knop Bestand kiezen met de hex code #2196f3 (een tint blauw) en de rand van de knop Afbeelding uploaden naar oranje. Je stylt ook de elementen op de route /result om ze representatiever te maken.

Als je klaar bent, sla je het bestand op en sluit je het.

Keer terug naar de hoofdmap van het project:

cd ../..

Open index.js in je editor:

nano index.js

Voeg in je index.js het volgende code toe, waarmee de server wordt gestart:

...
app.listen(3000, function () {
  console.log("Server running on port 3000");
});

Het volledige bestand index.js komt er nu als volgt uit te zien:

const path = require("path");
const fs = require("fs");
const express = require("express");
const bodyParser = require("body-parser");
const sharp = require("sharp");
const fileUpload = require("express-fileupload");

const app = express();
app.set("view engine", "ejs");
app.use(bodyParser.json());
app.use(
  bodyParser.urlencoded({
    extended: true,
  })
);

app.use(fileUpload());

app.use(express.static("public"));

app.get("/", function (req, res) {
  res.render("form");
});

app.get("/result", (req, res) => {
  const imgDirPath = path.join(__dirname, "./public/images");
  let imgFiles = fs.readdirSync(imgDirPath).map((image) => {
    return `images/${image}`;
  });
  res.render("result", { imgFiles });
});

app.post("/upload", async function (req, res) {
  const { image } = req.files;

  if (!image) return res.sendStatus(400);
  const imageName = path.parse(image.name).name;
  const processImage = (size) =>
    sharp(image.data)
      .resize(size, size)
      .webp({ lossless: true })
      .toFile(`./public/images/${imageName}-${size}.webp`);

  sizes = [90, 96, 120, 144, 160, 180, 240, 288, 360, 480, 720, 1440];
  Promise.all(sizes.map(processImage));
  let counter = 0;
  for (let i = 0; i < 10_000_000_000; i++) {
    counter++;
  }

  res.redirect("/result");
});

app.listen(3000, function () {
  console.log("Server running on port 3000");
});

Als je klaar bent met de wijzigingen, sla je het bestand op en sluit je het.

Start de app met het node commando:

node index.js

Je krijgt een uitvoer zoals deze:

Output
Server running on port 3000

Deze uitvoer bevestigt dat de server zonder problemen draait.

Open je favoriete browser en ga naar http://localhost:3000/.

ssh -L 3000:localhost:3000  your-non-root-user@yourserver-ip

Wanneer de pagina geladen is, zal het overeenkomen met het volgende:

Druk vervolgens op de Kies Bestand knop en selecteer het underwater.png afbeeldingsbestand op je lokale machine. Het scherm zal veranderen van Geen bestand gekozen naar underwater.png. Daarna druk je op de Afbeelding Uploaden knop. De app zal even laden terwijl het de afbeelding verwerkt en de incrementerende lus uitvoert.

Zodra de taak is voltooid, zal de /resultaat route geladen worden met de verkleinde afbeeldingen:

Je kunt de server nu stoppen met CTRL+C. Node.js herlaadt de server niet automatisch wanneer bestanden worden gewijzigd, dus je moet de server stoppen en opnieuw starten telkens wanneer je de bestanden bijwerkt.

Je weet nu hoe een tijdsintensieve taak het verzoek-/antwoordcyclus van een applicatie kan beïnvloeden. Je gaat de taak nu asynchroon uitvoeren.

Stap 3 — Uitvoeren van Tijdsintensieve Taken Asynchroon met bullmq

In deze stap zal je een tijdsintensieve taak naar de achtergrond verplaatsen met behulp van bullmq. Deze aanpassing zal de verzoek-/antwoordcyclus vrijmaken en je app in staat stellen onmiddellijk te reageren op gebruikers terwijl de afbeelding wordt verwerkt.

Om dat te doen, moet je een beknopte beschrijving van de taak maken en deze toevoegen aan een wachtrij met bullmq. Een wachtrij is een gegevensstructuur die vergelijkbaar werkt met hoe een wachtrij werkt in het echte leven. Wanneer mensen in de rij staan om een ruimte binnen te gaan, zal de eerste persoon in de rij de eerste persoon zijn die de ruimte binnengaat. Iedereen die later komt, sluit achteraan in de rij en zal de ruimte betreden nadat iedereen die hen voorgaat in de rij de ruimte is binnengegaan, totdat de laatste persoon de ruimte binnengaat. Met het First-In, First-Out (FIFO)-proces van de wachtrijgegevensstructuur is het eerste item dat aan de wachtrij wordt toegevoegd, het eerste item dat wordt verwijderd (dequeue). Met bullmq zal een producent een taak toevoegen aan een wachtrij, en een consument (of werker) zal een taak uit de wachtrij verwijderen en uitvoeren.

De wachtrij in bullmq bevindt zich in Redis. Wanneer je een taak beschrijft en deze aan de wachtrij toevoegt, wordt er een vermelding voor de taak gemaakt in een Redis-wachtrij. Een taakbeschrijving kan een tekenreeks zijn of een object met eigenschappen die minimale gegevens bevatten of verwijzingen naar de gegevens die bullmq later in staat zullen stellen om de taak uit te voeren. Zodra je de functionaliteit hebt gedefinieerd om taken aan de wachtrij toe te voegen, verplaats je de tijdsintensieve code naar een afzonderlijke functie. Later zal bullmq deze functie aanroepen met de gegevens die je in de wachtrij hebt opgeslagen wanneer de taak uit de wachtrij wordt gehaald. Nadat de taak is voltooid, zal bullmq deze markeren als voltooid, een andere taak uit de wachtrij halen en deze uitvoeren.

Open index.js in je editor:

nano index.js

In je index.js-bestand voeg je de gemarkeerde regels toe om een wachtrij in Redis te creëren met bullmq:

...
const fileUpload = require("express-fileupload");
const { Queue } = require("bullmq");

const redisOptions = { host: "localhost", port: 6379 };

const imageJobQueue = new Queue("imageJobQueue", {
  connection: redisOptions,
});

async function addJob(job) {
  await imageJobQueue.add(job.type, job);
}
...

Je begint met het extraheren van de Queue-klasse uit bullmq, die wordt gebruikt om een wachtrij in Redis te maken. Vervolgens stel je de variabele redisOptions in op een object met eigenschappen die de Queue-klasse-instantie zal gebruiken om een verbinding met Redis tot stand te brengen. Je stelt de waarde van de eigenschap host in op localhost, omdat Redis op je lokale machine draait.

Vervolgens stel je de variabele imageJobQueue in op een instantie van de Queue klasse, waarbij je de naam van de queue als eerste argument en een object als tweede argument meegeeft. Het object heeft een connection eigenschap met de waarde ingesteld op een object in de redisOptions variabele. Na het instantiëren van de Queue klasse zal er een queue genaamd imageJobQueue worden aangemaakt in Redis.

Tenslotte, definieer je de addJob() functie die je zult gebruiken om een job toe te voegen aan de imageJobQueue. De functie neemt een parameter van job die de informatie over de job bevat (je zult de addJob() functie aanroepen met de gegevens die je wilt opslaan in een queue). In de functie roep je de add() methode aan van de imageJobQueue, waarbij je de naam van de job als eerste argument en de job data als tweede argument doorgeeft.

Voeg de gemarkeerde code toe om de addJob() functie aan te roepen om een job toe te voegen aan de queue:

...
app.post("/upload", async function (req, res) {
  const { image } = req.files;

  if (!image) return res.sendStatus(400);
  const imageName = path.parse(image.name).name;
  ...
  await addJob({
    type: "processUploadedImages",
    image: {
      data: image.data.toString("base64"),
      name: image.name,
    },
  });

  res.redirect("/result");
});
...

Hier roep je de functie addJob() aan met een object dat de taak beschrijft. Het object heeft het attribuut type met de waarde van de taaknaam. De tweede eigenschap, image, is ingesteld op een object dat de door de gebruiker geüploade afbeeldingsgegevens bevat. Omdat de afbeeldingsgegevens in image.data zich in een buffer (binaire vorm) bevinden, roep je de toString()-methode van JavaScript aan om deze om te zetten naar een string die kan worden opgeslagen in Redis, waardoor de data-eigenschap wordt ingesteld. De eigenschap image is ingesteld op de naam van de geüploade afbeelding (inclusief de afbeeldingsextensie).

Je hebt nu de informatie gedefinieerd die nodig is voor bullmq om deze taak later uit te voeren. Afhankelijk van je taak kun je meer of minder taakinformatie toevoegen.

Sla je bestand op en sluit het.

Maak vervolgens het bestand utils.js aan en open het, hierin komt de code voor de beeldverwerking:

nano utils.js

In je bestand utils.js voeg je de volgende code toe om de functie voor de verwerking van een afbeelding te definiëren:

const path = require("path");
const sharp = require("sharp");

function processUploadedImages(job) {
}

module.exports = { processUploadedImages };

U importeert de benodigde modules om afbeeldingen te verwerken en paden te berekenen in de eerste twee regels. Vervolgens definieert u de functie processUploadedImages(), die de tijdsintensieve taak van beeldverwerking zal bevatten. Deze functie neemt een job-parameter aan die zal worden ingevuld wanneer de worker de jobgegevens uit de wachtrij haalt en vervolgens de functie processUploadedImages() aanroept met de wachtrijgegevens. U exporteert ook de functie processUploadedImages() zodat u er naar kunt verwijzen in andere bestanden.

Sla uw bestand op en sluit het.

Keer terug naar het bestand index.js:

nano index.js

Kopieer de gemarkeerde regels uit het bestand index.js en verwijder ze vervolgens uit dit bestand. U heeft de gekopieerde code zo meteen nodig, dus sla deze op naar het klembord. Als u nano gebruikt, kunt u deze regels markeren en met de rechtermuisknop klikken om de regels te kopiëren:

...
app.post("/upload", async function (req, res) {
  const { image } = req.files;

  if (!image) return res.sendStatus(400);
  const imageName = path.parse(image.name).name;
  const processImage = (size) =>
    sharp(image.data)
      .resize(size, size)
      .webp({ lossless: true })
      .toFile(`./public/images/${imageName}-${size}.webp`);

  sizes = [90, 96, 120, 144, 160, 180, 240, 288, 360, 480, 720, 1440];
  Promise.all(sizes.map(processImage))
  let counter = 0;
  for (let i = 0; i < 10_000_000_000; i++) {
    counter++;
  };
...
  res.redirect("/result");
});

De post-methode voor de upload-route zal nu overeenkomen met het volgende:

...
app.post("/upload", async function (req, res) {
  const { image } = req.files;

  if (!image) return res.sendStatus(400);

  await addJob({
    type: "processUploadedImages",
    image: {
      data: image.data.toString("base64"),
      name: image.name,
    },
  });

  res.redirect("/result");
});
...

Sla dit bestand op en sluit het, open vervolgens het bestand utils.js:

nano utils.js

In uw bestand utils.js, plak de regels die u zojuist gekopieerd heeft voor de /upload-route callback in de functie processUploadedImages:

...
function processUploadedImages(job) {
  const imageName = path.parse(image.name).name;
  const processImage = (size) =>
    sharp(image.data)
      .resize(size, size)
      .webp({ lossless: true })
      .toFile(`./public/images/${imageName}-${size}.webp`);

  sizes = [90, 96, 120, 144, 160, 180, 240, 288, 360, 480, 720, 1440];
  Promise.all(sizes.map(processImage));
  let counter = 0;
  for (let i = 0; i < 10_000_000_000; i++) {
    counter++;
  };
}
...

Nu u de code voor het verwerken van een afbeelding heeft verplaatst, moet u deze bijwerken om de afbeeldingsgegevens te gebruiken van de job-parameter van de eerder gedefinieerde functie processUploadedImages().

Om dat te doen, voegt u de gemarkeerde regels hieronder toe en bijwerkt u ze:

function processUploadedImages(job) {
  const imageFileData = Buffer.from(job.image.data, "base64");
  const imageName = path.parse(job.image.name).name;
  const processImage = (size) =>
    sharp(imageFileData)
      .resize(size, size)
      .webp({ lossless: true })
      .toFile(`./public/images/${imageName}-${size}.webp`);
  ...
}

Je converteert de gestringifieerde versie van de beeldgegevens terug naar binair met de Buffer.from() methode. Vervolgens update je path.parse() met een verwijzing naar de afbeeldingsnaam die in de wachtrij is opgeslagen. Daarna update je de sharp() methode om de binaire beeldgegevens opgeslagen in de imageFileData variabele te gebruiken.

Het volledige utils.js bestand zal nu overeenkomen met het volgende:

const path = require("path");
const sharp = require("sharp");

function processUploadedImages(job) {
  const imageFileData = Buffer.from(job.image.data, "base64");
  const imageName = path.parse(job.image.name).name;
  const processImage = (size) =>
    sharp(imageFileData)
      .resize(size, size)
      .webp({ lossless: true })
      .toFile(`./public/images/${imageName}-${size}.webp`);

  sizes = [90, 96, 120, 144, 160, 180, 240, 288, 360, 480, 720, 1440];
  Promise.all(sizes.map(processImage));
  let counter = 0;
  for (let i = 0; i < 10_000_000_000; i++) {
    counter++;
  };
}

module.exports = { processUploadedImages };

Sla je bestand op en sluit het, keer dan terug naar de index.js:

nano index.js

De sharp variabele is niet langer nodig als afhankelijkheid omdat de afbeelding nu verwerkt wordt in het utils.js bestand. Verwijder de gemarkeerde regel uit het bestand:

const bodyParser = require("body-parser");
const sharp = require("sharp");
const fileUpload = require("express-fileupload");
const { Queue } = require("bullmq");
...

Sla je bestand op en sluit het.

Je hebt nu de functionaliteit gedefinieerd om een wachtrij in Redis te maken en een taak toe te voegen. Je hebt ook de processUploadedImages() functie gedefinieerd om geüploade afbeeldingen te verwerken.

De resterende taak is om een consument (of werker) te maken die een taak uit de wachtrij haalt en de processUploadedImages() functie aanroept met de taakgegevens.

Maak een worker.js bestand in je editor:

nano worker.js

Voeg in je worker.js bestand de volgende code toe:

const { Worker } = require("bullmq");

const { processUploadedImages } = require("./utils");

const workerHandler = (job) => {
  console.log("Starting job:", job.name);
  processUploadedImages(job.data);
  console.log("Finished job:", job.name);
  return;
};

In de eerste regel importeer je de Worker klasse vanuit bullmq; wanneer geïnstantieerd, zal dit een werker starten die taken uit de wachtrij in Redis dequeue’t en uitvoert. Vervolgens refereer je aan de processUploadedImages() functie uit het utils.js bestand zodat de werker de functie kan aanroepen met de gegevens in de wachtrij.

Je definieert een functie workerHandler() die een parameter job accepteert met de jobgegevens in de wachtrij. In de functie log je dat de job is gestart, roep je vervolgens processUploadedImages() aan met de jobgegevens. Daarna log je een succesbericht en retourneer je null.

Om de worker in staat te stellen verbinding te maken met Redis, een job uit de wachtrij te halen en workerHandler() met de jobgegevens aan te roepen, voeg je de volgende regels toe aan het bestand:

...
const workerOptions = {
  connection: {
    host: "localhost",
    port: 6379,
  },
};

const worker = new Worker("imageJobQueue", workerHandler, workerOptions);

console.log("Worker started!");

Hier stel je de variabele workerOptions in op een object met de verbindingsinstellingen voor Redis. Je stelt de variabele worker in op een instantie van de klasse Worker die de volgende parameters accepteert:

• imageJobQueue: de naam van de jobwachtrij.
• workerHandler: de functie die wordt uitgevoerd nadat een job uit de Redis-wachtrij is gehaald.
• workerOptions: de Redis-configuratie-instellingen die de worker gebruikt om verbinding te maken met Redis.

Tenslotte log je een succesbericht.

Na het toevoegen van de regels, sla je het bestand op en sluit je het.

Je hebt nu de functionaliteit van de bullmq-worker gedefinieerd om jobs uit de wachtrij te halen en ze uit te voeren.

In je terminal, verwijder de afbeeldingen in de map public/images zodat je fris kunt beginnen voor het testen van je app:

rm public/images/*

Vervolgens voer je het bestand index.js uit:

node index.js

De app zal starten:

Output
Server running on port 3000

Start nu de worker. Open een tweede terminalsessie en navigeer naar het project:

cd image_processor/

Start de worker met het volgende commando:

node worker.js

De worker zal starten:

Output
Worker started!

Bezoek http://localhost:3000/ in je browser. Druk op de Kies bestand knop en selecteer de underwater.png van je computer, druk dan op de Upload Afbeelding knop.

Je kunt een onmiddellijke reactie ontvangen die je vertelt om de pagina na een paar seconden te vernieuwen:

Als alternatief kun je een onmiddellijke reactie ontvangen met enkele verwerkte afbeeldingen op de pagina terwijl andere nog worden verwerkt:

Je kunt de pagina een paar keer vernieuwen om alle verkleinde afbeeldingen te laden.

Keer terug naar de terminal waar je worker draait. Die terminal zal een bericht hebben dat overeenkomt met het volgende:

Output
Worker started!
Starting job: processUploadedImages
Finished job: processUploadedImages

De uitvoer bevestigt dat bullmq de taak succesvol heeft uitgevoerd.

Je app kan nog steeds tijdsintensieve taken uitbesteden, zelfs als de worker niet draait. Om dit te demonstreren, stop de worker in de tweede terminal met CTRL+C.

In je initiële terminalsessie, stop de Express-server en verwijder de afbeeldingen in public/images:

rm public/images/*

Start daarna de server opnieuw:

node index.js

In je browser, bezoek http://localhost:3000/ en upload opnieuw de underwater.png afbeelding. Wanneer je wordt omgeleid naar het /result pad, zullen de afbeeldingen niet op de pagina verschijnen omdat de worker niet draait:

Keer terug naar de terminal waar je de worker hebt gestart en start de worker opnieuw:

node worker.js

De uitvoer zal overeenkomen met het volgende, wat je laat weten dat de taak is gestart:

Output
Worker started!
Starting job: processUploadedImages

Nadat de taak is voltooid en de uitvoer een regel bevat die luidt Finished job: processUploadedImages, vernieuw de browser. De afbeeldingen worden nu geladen:

Stop de server en de worker.

Je kunt nu een tijdsintensieve taak naar de achtergrond verplaatsen en deze asynchroon uitvoeren met behulp van bullmq. In de volgende stap stel je een dashboard in om de status van de wachtrij te controleren.

Stap 4 – Een dashboard toevoegen om bullmq-wachtrijen te controleren

In deze stap gebruik je het bull-board pakket om de taken in de Redis-wachtrij te monitoren vanuit een visueel dashboard. Dit pakket zal automatisch een gebruikersinterface (UI) dashboard maken dat de informatie over de bullmq-taken die zijn opgeslagen in de Redis-wachtrij weergeeft en organiseert. Met je browser kun je de taken monitoren die zijn voltooid, in afwachting zijn, of zijn mislukt zonder de Redis CLI in de terminal te openen.

Open het index.js bestand in je teksteditor:

nano index.js

Voeg de gemarkeerde code toe om bull-board te importeren:

...
const { Queue } = require("bullmq");
const { createBullBoard } = require("@bull-board/api");
const { BullMQAdapter } = require("@bull-board/api/bullMQAdapter");
const { ExpressAdapter } = require("@bull-board/express");
...

In de voorafgaande code importeer je de methode createBullBoard() uit bull-board. Je importeert ook BullMQAdapter, waarmee bull-board toegang krijgt tot bullmq-wachtrijen, en ExpressAdapter, die functionaliteit biedt voor Express om het dashboard weer te geven.

Voeg vervolgens de gemarkeerde code toe om bull-board te verbinden met bullmq:

...
async function addJob(job) {
  ...
}

const serverAdapter = new ExpressAdapter();
const bullBoard = createBullBoard({
  queues: [new BullMQAdapter(imageJobQueue)],
  serverAdapter: serverAdapter,
});
serverAdapter.setBasePath("/admin");

const app = express();
...

Als eerste stel je de serverAdapter in op een instantie van ExpressAdapter. Vervolgens roep je createBullBoard() aan om het dashboard te initialiseren met de gegevens van de bullmq-wachtrij. Je geeft de functie een objectargument met de eigenschappen queues en serverAdapter. De eerste eigenschap, queues, accepteert een array van de wachtrijen die je hebt gedefinieerd met bullmq, wat hier de imageJobQueue is. De tweede eigenschap, serverAdapter, bevat een object dat een instantie van de Express server adapter accepteert. Daarna stel je het pad /admin in om toegang te krijgen tot het dashboard met de methode setBasePath().

Voeg vervolgens de middleware van de serverAdapter toe voor de route /admin:

app.use(express.static("public"))

app.use("/admin", serverAdapter.getRouter());

app.get("/", function (req, res) {
  ...
});

Het volledige bestand index.js zal overeenkomen met het volgende:

const path = require("path");
const fs = require("fs");
const express = require("express");
const bodyParser = require("body-parser");
const fileUpload = require("express-fileupload");
const { Queue } = require("bullmq");
const { createBullBoard } = require("@bull-board/api");
const { BullMQAdapter } = require("@bull-board/api/bullMQAdapter");
const { ExpressAdapter } = require("@bull-board/express");

const redisOptions = { host: "localhost", port: 6379 };

const imageJobQueue = new Queue("imageJobQueue", {
  connection: redisOptions,
});

async function addJob(job) {
  await imageJobQueue.add(job.type, job);
}

const serverAdapter = new ExpressAdapter();
const bullBoard = createBullBoard({
  queues: [new BullMQAdapter(imageJobQueue)],
  serverAdapter: serverAdapter,
});
serverAdapter.setBasePath("/admin");

const app = express();
app.set("view engine", "ejs");
app.use(bodyParser.json());
app.use(
  bodyParser.urlencoded({
    extended: true,
  })
);
app.use(fileUpload());

app.use(express.static("public"));

app.use("/admin", serverAdapter.getRouter());

app.get("/", function (req, res) {
  res.render("form");
});

app.get("/result", (req, res) => {
  const imgDirPath = path.join(__dirname, "./public/images");
  let imgFiles = fs.readdirSync(imgDirPath).map((image) => {
    return `images/${image}`;
  });
  res.render("result", { imgFiles });
});

app.post("/upload", async function (req, res) {
  const { image } = req.files;

  if (!image) return res.sendStatus(400);

  await addJob({
    type: "processUploadedImages",
    image: {
      data: Buffer.from(image.data).toString("base64"),
      name: image.name,
    },
  });

  res.redirect("/result");
});

app.listen(3000, function () {
  console.log("Server running on port 3000");
});

Nadat je klaar bent met het aanbrengen van wijzigingen, sla je het bestand op en sluit je het.

Voer het bestand index.js uit:

node index.js

Keer terug naar je browser en bezoek http://localhost:3000/admin. Het dashboard zal laden:

In het dashboard kun je het type taak, de verbruikte gegevens en meer informatie over de taak bekijken. Je kunt ook overschakelen naar andere tabbladen, zoals het Voltooide tabblad voor informatie over voltooide taken, het Mislukte tabblad voor meer informatie over mislukte taken, en het Gepauzeerd tabblad voor meer informatie over gepauzeerde taken.

Je kunt nu het bull-board dashboard gebruiken om wachtrijen te monitoren.

Conclusie

In dit artikel heb je een tijdsintensieve taak uitbesteed aan een takenwachtrij met behulp van bullmq. Eerst heb je, zonder bullmq te gebruiken, een app gemaakt met een tijdsintensieve taak die een trage verzoek/respons-cyclus heeft. Vervolgens heb je bullmq gebruikt om de tijdsintensieve taak uit te besteden en asynchroon uit te voeren, wat de verzoek/respons-cyclus verbetert. Daarna heb je bull-board gebruikt om een dashboard te maken om bullmq-wachtrijen in Redis te monitoren.

Je kunt de bullmq documentatie bezoeken om meer te weten te komen over functies van bullmq die niet in deze tutorial worden behandeld, zoals plannen, prioriteren of opnieuw proberen van taken, en het configureren van concurrency-instellingen voor werkers. Je kunt ook de bull-board documentatie bezoeken om meer te weten te komen over de dashboardfuncties.