Wie man asynchrone Aufgaben mit Node.js und BullMQ behandelt

Der Autor hat die Gesellschaft der Ingenieurinnen ausgewählt, um im Rahmen des Write for Donations-Programms eine Spende zu erhalten.

Einführung

Webanwendungen haben Anfrage/Antwort-Zyklen. Wenn Sie eine URL besuchen, sendet der Browser eine Anfrage an den Server, auf dem eine App läuft, die Daten verarbeitet oder Abfragen in der Datenbank ausführt. Währenddessen wartet der Benutzer, bis die App eine Antwort zurückgibt. Für einige Aufgaben kann der Benutzer schnell eine Antwort erhalten; für zeitintensive Aufgaben wie die Bearbeitung von Bildern, die Analyse von Daten, das Erstellen von Berichten oder das Versenden von E-Mails dauern diese Aufgaben jedoch lange und können den Anfrage-/Antwort-Zyklus verlangsamen. Angenommen, Sie haben eine Anwendung, in der Benutzer Bilder hochladen. In diesem Fall müssen Sie möglicherweise das Bild verkleinern, komprimieren oder in ein anderes Format konvertieren, um den Speicherplatz Ihres Servers zu erhalten, bevor Sie das Bild dem Benutzer anzeigen. Die Verarbeitung eines Bildes ist eine CPU-intensive Aufgabe, die einen Node.js-Thread blockieren kann, bis die Aufgabe abgeschlossen ist. Das kann einige Sekunden oder Minuten dauern. Benutzer müssen auf die Fertigstellung der Aufgabe warten, um eine Antwort vom Server zu erhalten.

Um die Anfrage/Antwort-Zyklus nicht zu verlangsamen, können Sie bullmq verwenden, eine verteilte Aufgabenwarteschlange, die es Ihnen ermöglicht, zeitaufwändige Aufgaben von Ihrer Node.js-App auf bullmq auszulagern und den Anfrage/Antwort-Zyklus freizugeben. Dieses Tool ermöglicht es Ihrer App, schnell auf Benutzeranfragen zu antworten, während bullmq die Aufgaben im Hintergrund und unabhängig von Ihrer App asynchron ausführt. Um Jobs zu verfolgen, verwendet bullmq Redis, um eine kurze Beschreibung jedes Jobs in einer Warteschlange zu speichern. Ein bullmq-Worker holt dann jeden Job aus der Warteschlange und führt ihn aus, wobei er ihn als erledigt markiert.

In diesem Artikel werden Sie bullmq verwenden, um eine zeitaufwändige Aufgabe im Hintergrund auszulagern, was es einer Anwendung ermöglicht, schnell auf Benutzeranfragen zu reagieren. Zunächst werden Sie eine App mit einer zeitaufwändigen Aufgabe erstellen, ohne bullmq zu verwenden. Anschließend werden Sie bullmq verwenden, um die Aufgabe asynchron auszuführen. Schließlich werden Sie ein visuelles Dashboard installieren, um bullmq-Jobs in einer Redis-Warteschlange zu verwalten.

Voraussetzungen

Um diesem Tutorial zu folgen, benötigen Sie folgendes:

• Einrichtung der Node.js-Entwicklungsumgebung. Für Ubuntu 22.04 folgen Sie unserem Tutorial zu So installieren Sie Node.js unter Ubuntu 22.04. Für andere Systeme siehe So installieren Sie Node.js und erstellen eine lokale Entwicklungsumgebung.

• Redis auf Ihrem System installiert. Auf Ubuntu 22 folgen Sie den Schritten 1 bis 3 in unserem Tutorial zu So installieren und sichern Sie Redis unter Ubuntu 22.04. Für andere Systeme sehen Sie unser Tutorial zu So installieren und sichern Sie Redis.

• Vertrautheit mit Versprechungen und async/await Funktionen, die Sie in unserem Tutorial Verständnis der Ereignisschleife, Rückrufe, Versprechungen und Async/Await in JavaScript entwickeln können.

• Grundkenntnisse darüber, wie man Express verwendet. Siehe unser Tutorial zu Wie man mit Node.js und Express startet.

• Vertrautheit mit Embedded JavaScript (EJS). Erfahren Sie mehr in unserem Tutorial zu Node-Anwendung mit EJS als Vorlage.

• Grundverständnis für die Bildverarbeitung mit sharp, das Sie in unserem Tutorial zu Bildverarbeitung in Node.js mit Sharp erlernen können.

Schritt 1 – Einrichten des Projektverzeichnisses

In diesem Schritt erstellen Sie ein Verzeichnis und installieren die erforderlichen Abhängigkeiten für Ihre Anwendung. Die Anwendung, die Sie in diesem Tutorial erstellen werden, ermöglicht es Benutzern, ein Bild hochzuladen, das dann mit dem sharp-Paket verarbeitet wird. Die Bildverarbeitung ist zeitintensiv und kann den Anfrage-/Antwortzyklus verlangsamen, weshalb diese Aufgabe sich gut dafür eignet, von bullmq im Hintergrund abgeladen zu werden. Die Technik, die Sie verwenden werden, um die Aufgabe abzuladen, funktioniert auch für andere zeitintensive Aufgaben.

Um zu beginnen, erstellen Sie ein Verzeichnis mit dem Namen image_processor und wechseln Sie in das Verzeichnis:

mkdir image_processor && cd image_processor

Dann initialisieren Sie das Verzeichnis als npm-Paket:

npm init -y

Der Befehl erstellt eine package.json-Datei. Die Option -y teilt npm mit, alle Standards zu akzeptieren.

Nach Ausführung des Befehls stimmt Ihre Ausgabe mit der folgenden überein:

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"
}

Die Ausgabe bestätigt, dass die package.json-Datei erstellt wurde. Wichtige Eigenschaften sind der Name Ihrer App (name), die Versionsnummer Ihrer Anwendung (version) und der Startpunkt Ihres Projekts (main). Wenn Sie mehr über die anderen Eigenschaften erfahren möchten, können Sie die Dokumentation von npm zu package.json überprüfen.

Die Anwendung, die Sie in diesem Tutorial erstellen werden, erfordert die folgenden Abhängigkeiten:

• express: ein Webframework zum Erstellen von Webanwendungen.
• express-fileupload: ein Middleware, die es Ihren Formularen ermöglicht, Dateien hochzuladen.
• sharp: eine Bildverarbeitungsbibliothek.
• ejs: eine Vorlagensprache, die es Ihnen ermöglicht, HTML-Markup mit Node.js zu generieren.
• bullmq: eine verteilte Task-Warteschlange.
• bull-board: ein Dashboard, das auf bullmq aufbaut und den Status der Aufgaben mit einer benutzerfreundlichen Benutzeroberfläche (UI) anzeigt.

Um alle diese Abhängigkeiten zu installieren, führen Sie den folgenden Befehl aus:

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

Neben den installierten Abhängigkeiten verwenden Sie später in diesem Tutorial auch das folgende Bild:

Verwenden Sie curl, um das Bild an den von Ihnen gewählten Speicherort auf Ihrem lokalen Computer herunterzuladen

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

Sie haben die notwendigen Abhängigkeiten, um eine Node.js-Anwendung zu erstellen, die kein bullmq hat, was Sie als nächstes tun werden.

Schritt 2 – Umsetzung einer zeitintensiven Aufgabe ohne bullmq

In diesem Schritt erstellen Sie eine Anwendung mit Express, mit der Benutzer Bilder hochladen können. Die App startet eine zeitintensive Aufgabe mit sharp, um das Bild in mehrere Größen zu ändern, die dem Benutzer nach dem Senden einer Antwort angezeigt werden. Dieser Schritt hilft Ihnen zu verstehen, wie sich zeitintensive Aufgaben auf den Anfrage-/Antwortzyklus auswirken.

Mit nano oder Ihrem bevorzugten Texteditor erstellen Sie die Datei index.js:

nano index.js

Fügen Sie in Ihrer Datei index.js den folgenden Code hinzu, um Abhängigkeiten zu importieren:

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 der ersten Zeile importieren Sie das Modul path zum Berechnen von Dateipfaden mit Node. In der zweiten Zeile importieren Sie das Modul fs für die Interaktion mit Verzeichnissen. Dann importieren Sie das Web-Framework express. Sie importieren das Modul body-parser, um Middleware zum Parsen von Daten in HTTP-Anfragen hinzuzufügen. Anschließend importieren Sie das Modul sharp für die Bildverarbeitung. Schließlich importieren Sie express-fileupload für die Verarbeitung von Uploads aus einem HTML-Formular.

Fügen Sie als Nächstes den folgenden Code hinzu, um Middleware in Ihrer App zu implementieren:

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

Zuerst setzen Sie die Variable app auf eine Instanz von Express. Zweitens konfigurieren Sie mit der Methode set() von Express die Verwendung der Template-Sprache ejs. Dann fügen Sie das Middleware-Modul body-parser mit der Methode use() hinzu, um JSON-Daten in HTTP-Anfragen in Variablen umzuwandeln, auf die mit JavaScript zugegriffen werden kann. In der nächsten Zeile tun Sie dasselbe mit URL-codierten Eingaben.

Als Nächstes fügen Sie die folgenden Zeilen hinzu, um mehr Middleware hinzuzufügen, um Datei-Uploads zu handhaben und statische Dateien bereitzustellen:

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

Sie fügen Middleware hinzu, um hochgeladene Dateien zu analysieren, indem Sie die Methode fileUpload() aufrufen, und Sie legen ein Verzeichnis fest, in dem Express nach und nach statischen Dateien wie Bildern und CSS sucht und diese bereitstellt.

Mit den Middleware-Einstellungen erstellen Sie eine Route, die ein HTML-Formular zum Hochladen eines Bildes anzeigt:

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

Hier verwenden Sie die Methode get() des Express-Moduls, um die Route / und die Callback-Funktion anzugeben, die ausgeführt werden soll, wenn der Benutzer die Homepage oder die Route / besucht. In der Callback-Funktion rufen Sie res.render() auf, um die Datei form.ejs im Verzeichnis views zu rendern. Sie haben die Datei form.ejs oder das Verzeichnis views noch nicht erstellt.

Um dies zu erstellen, speichern und schließen Sie zuerst Ihre Datei. Geben Sie dann in Ihrem Terminal den folgenden Befehl ein, um das Verzeichnis views im Stammverzeichnis Ihres Projekts zu erstellen:

mkdir views

Wechseln Sie in das Verzeichnis views:

cd views

Erstellen Sie die Datei form.ejs in Ihrem Editor:

nano form.ejs

Fügen Sie in Ihrer Datei form.ejs den folgenden Code hinzu, um das Formular zu erstellen:

<!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>

Zuerst verweisen Sie auf die Datei head.ejs, die Sie noch nicht erstellt haben. Die Datei head.ejs wird das HTML head-Element enthalten, auf das Sie in anderen HTML-Seiten verweisen können.

Im body-Tag erstellen Sie ein Formular mit den folgenden Attributen:

• action gibt den Pfad an, wohin die Formulardaten gesendet werden sollen, wenn das Formular übermittelt wird.
• method gibt die HTTP-Methode zum Senden von Daten an. Die POST-Methode bindet die Daten in einer HTTP-Anforderung ein.
• encytype gibt an, wie die Formulardaten codiert werden sollen. Der Wert multipart/form-data ermöglicht es den HTML-input-Elementen, Dateidaten hochzuladen.

In dem form-Element erstellen Sie ein input-Tag, um Dateien hochzuladen. Dann definieren Sie das button-Element mit dem Attribut type, das auf submit gesetzt ist, was es Ihnen ermöglicht, Formulare zu übermitteln.

Speichern Sie abschließend Ihre Datei und schließen Sie sie.

Als Nächstes erstellen Sie eine head.ejs-Datei:

nano head.ejs

In Ihrer head.ejs-Datei fügen Sie den folgenden Code hinzu, um den Kopfbereich der App zu erstellen:

<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 verweisen Sie auf die Datei main.css, die Sie später im Schritt im Verzeichnis public erstellen werden. Diese Datei wird die Styles für diese Anwendung enthalten. Vorläufig werden Sie jedoch weiterhin die Prozesse für statische Assets einrichten.

Speichern und schließen Sie die Datei.

Um mit den aus dem Formular übermittelten Daten umzugehen, müssen Sie eine post-Methode in Express definieren. Gehen Sie dazu zurück zum Stammverzeichnis Ihres Projekts:

cd ..

Öffnen Sie Ihre Datei index.js erneut:

nano index.js

In Ihrer Datei index.js fügen Sie die hervorgehobenen Zeilen hinzu, um eine Methode zur Behandlung von Formulareinsendungen auf der Route /upload zu definieren:

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

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

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

});

Sie verwenden die Variable app, um die Methode post() aufzurufen, die das eingereichte Formular auf der Route /upload behandeln wird. Anschließend extrahieren Sie die hochgeladenen Bilddaten aus der HTTP-Anfrage in die Variable image. Danach setzen Sie eine Antwort, um einen Statuscode 400 zurückzugeben, wenn der Benutzer kein Bild hochlädt.

Um den Prozess für das hochgeladene Bild festzulegen, fügen Sie den folgenden hervorgehobenen Code hinzu:

...
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));
});

Diese Zeilen repräsentieren, wie Ihre App das Bild verarbeiten wird. Zuerst entfernen Sie die Bildextension vom hochgeladenen Bild und speichern den Namen in der Variable imageName. Als nächstes definieren Sie die Funktion processImage(). Diese Funktion nimmt den Parameter size entgegen, dessen Wert zur Bestimmung der Bildabmessungen während der Größenänderung verwendet wird. In der Funktion rufen Sie sharp() mit image.data auf, das ein Buffer mit den binären Daten für das hochgeladene Bild ist. sharp ändert die Größe des Bildes entsprechend dem Wert im size-Parameter. Sie verwenden die Methode webp() von sharp, um das Bild in das webp-Bildformat zu konvertieren. Anschließend speichern Sie das Bild im Verzeichnis public/images/.

Die nachfolgende Liste von Zahlen definiert die Größen, die verwendet werden, um das hochgeladene Bild zu verkleinern. Anschließend verwenden Sie die map()-Methode von JavaScript, um processImage() für jedes Element im sizes-Array aufzurufen, wonach es einen neuen Array zurückgibt. Jedes Mal, wenn die map()-Methode die processImage()-Funktion aufruft, gibt sie ein Promise für den neuen Array zurück. Sie verwenden die Promise.all()-Methode, um sie aufzulösen.

Die Computer-Verarbeitungsgeschwindigkeiten variieren, ebenso wie die Größe der Bilder, die ein Benutzer hochladen kann, was die Bildverarbeitungsgeschwindigkeit beeinflussen kann. Um diesen Code zu Demonstrationszwecken zu verzögern, fügen Sie die hervorgehobenen Zeilen ein, um eine CPU-intensive Inkrement-Schleife und eine Weiterleitung zu einer Seite hinzuzufügen, die die verkleinerten Bilder mit den hervorgehobenen Zeilen anzeigt:

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

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

Die Schleife wird 10 Milliarden Mal ausgeführt, um die Variable counter zu erhöhen. Sie rufen die Funktion res.redirect() auf, um die App zur /result-Route umzuleiten. Die Route rendert eine HTML-Seite, die die Bilder im Verzeichnis public/images anzeigt.

Die /result-Route existiert noch nicht. Um sie zu erstellen, fügen Sie den hervorgehobenen Code Ihrer index.js-Datei hinzu:

...

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) {
  ...
});

Sie definieren die Route /result mit der Methode app.get(). In der Funktion definieren Sie die Variable imgDirPath mit dem vollständigen Pfad zum Verzeichnis public/images. Sie verwenden die Methode readdirSync() des Moduls fs, um alle Dateien in dem angegebenen Verzeichnis zu lesen. Von dort aus hängen Sie die Methode map() an, um ein neues Array mit den Bildpfaden zu erstellen, die mit images/ vorangestellt sind.

Zum Schluss rufen Sie res.render() auf, um die Datei result.ejs zu rendern, die noch nicht existiert. Sie übergeben die Variable imgFiles, die ein Array aller relativen Pfade zu den Bildern enthält, an die Datei result.ejs.

Speichern Sie die Datei und schließen Sie sie.

Um die Datei result.ejs zu erstellen, kehren Sie zum Verzeichnis views zurück:

cd views

Erstellen und öffnen Sie die Datei result.ejs in Ihrem Editor:

nano result.ejs

In Ihrer Datei result.ejs fügen Sie die folgenden Zeilen hinzu, um Bilder anzuzeigen:

<!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>

Zunächst verweisen Sie auf die Datei head.ejs. Im body-Tag überprüfen Sie, ob die Variable imgFiles leer ist. Wenn sie Daten enthält, iterieren Sie über jede Datei und erstellen ein Bild für jedes Array-Element. Wenn imgFiles leer ist, wird eine Meldung ausgegeben, die den Benutzer auffordert, Nach einigen Sekunden aktualisieren, um die verkleinerten Bilder anzuzeigen..

Speichern Sie die Datei und schließen Sie sie.

Kehren Sie als nächstes zum Stammverzeichnis zurück und erstellen Sie das Verzeichnis public, das Ihre statischen Assets enthält.

cd .. && mkdir public

Bewegen Sie sich in das Verzeichnis public:

cd public

Erstellen Sie ein Verzeichnis images, das die hochgeladenen Bilder aufbewahren wird:

mkdir images

Erstellen Sie als Nächstes das Verzeichnis css und navigieren Sie dorthin:

mkdir css && cd css

In Ihrem Editor erstellen und öffnen Sie die Datei main.css, auf die Sie zuvor in der Datei head.ejs verwiesen haben:

nano main.css

In Ihrer Datei main.css fügen Sie die folgenden Styles hinzu:

body {
  background: #f8f8f8;
}

h1 {
  text-align: center;
}

p {
  margin-bottom: 20px;
}

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

/** Styles für die Schaltfläche "Datei wählen" **/
button[type="submit"] {
  background: none;
  border: 1px solid orange;
  padding: 10px 30px;
  border-radius: 30px;
  transition: all 1s;
}

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

/** Styles für die Schaltfläche "Bild hochladen" **/
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;
}

Diese Zeilen werden Elemente in der App stylen. Mit HTML-Attributen gestalten Sie die Hintergrundfarbe der Schaltfläche Datei wählen mit dem Hex-Code #2196f3 (ein Blauton) und den Rand der Schaltfläche Bild hochladen mit orange. Sie gestalten auch die Elemente auf der Route /result, um sie ansprechender zu machen.

Speichern und schließen Sie Ihre Datei, wenn Sie fertig sind.

Kehren Sie zum Stammverzeichnis des Projekts zurück:

cd ../..

Öffnen Sie index.js in Ihrem Editor:

nano index.js

In Ihrem index.js fügen Sie den folgenden Code hinzu, der den Server startet:

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

Die vollständige Datei index.js entspricht nun folgendem:

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");
});

Wenn Sie mit den Änderungen fertig sind, speichern und schließen Sie Ihre Datei.

Führen Sie die App mit dem Befehl node aus:

node index.js

Sie erhalten eine Ausgabe wie folgt:

Output
Server running on port 3000

Diese Ausgabe bestätigt, dass der Server ohne Probleme läuft.

Öffnen Sie Ihren bevorzugten Browser und besuchen Sie http://localhost:3000/.

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

Wenn die Seite geladen ist, wird sie dem folgenden entsprechen:

Drücken Sie als nächstes die Datei auswählen Schaltfläche und wählen Sie das underwater.png Bild auf Ihrem lokalen Computer aus. Die Anzeige wechselt von Keine Datei ausgewählt zu underwater.png. Drücken Sie anschließend die Bild hochladen Schaltfläche. Die App lädt eine Weile, während sie das Bild verarbeitet und die Inkrementierungsschleife ausführt.

Sobald die Aufgabe abgeschlossen ist, wird die /result Route mit den skalierten Bildern geladen:

Sie können den Server jetzt mit CTRL+C stoppen. Node.js lädt den Server nicht automatisch neu, wenn Dateien geändert werden, daher müssen Sie den Server stoppen und neu starten, wann immer Sie die Dateien aktualisieren.

Sie wissen jetzt, wie sich eine zeitaufwändige Aufgabe auf den Anfrage-/Antwortzyklus einer Anwendung auswirken kann. Als nächstes werden Sie die Aufgabe asynchron ausführen.

Schritt 3 — Ausführen zeitintensiver Aufgaben asynchron mit bullmq

In diesem Schritt werden Sie eine zeitintensive Aufgabe im Hintergrund mit bullmq auslagern. Diese Anpassung wird den Anfrage-/Antwortzyklus freigeben und Ihrer App ermöglichen, sofort auf Benutzer zu reagieren, während das Bild verarbeitet wird.

Um das zu tun, müssen Sie eine präzise Beschreibung der Aufgabe erstellen und sie mit bullmq einer Warteschlange hinzufügen. Eine Warteschlange ist eine Datenstruktur, die ähnlich funktioniert wie eine Warteschlange im wirklichen Leben. Wenn sich Menschen anstellen, um einen Raum zu betreten, wird die erste Person in der Schlange die erste Person sein, die den Raum betritt. Jeder, der später kommt, stellt sich am Ende der Schlange an und betritt den Raum nach allen, die vor ihnen in der Schlange stehen, bis die letzte Person den Raum betritt. Mit dem Prozess der Warteschlangendatenstruktur First-In, First-Out (FIFO) wird das erste Element, das der Warteschlange hinzugefügt wird, als erstes entfernt (dequeue). Mit bullmq wird ein Produzent eine Aufgabe in eine Warteschlange einfügen, und ein Verbraucher (oder Arbeiter) wird eine Aufgabe aus der Warteschlange entfernen und ausführen.

Die Warteschlange in bullmq befindet sich in Redis. Wenn Sie eine Aufgabe beschreiben und sie zur Warteschlange hinzufügen, wird ein Eintrag für die Aufgabe in einer Redis-Warteschlange erstellt. Eine Aufgabenbeschreibung kann ein String oder ein Objekt mit Eigenschaften sein, die minimale Daten oder Verweise auf die Daten enthalten, die es bullmq ermöglichen, die Aufgabe später auszuführen. Nachdem Sie die Funktionalität zum Hinzufügen von Aufgaben zur Warteschlange definiert haben, verschieben Sie den zeitintensiven Code in eine separate Funktion. Später wird bullmq diese Funktion mit den Daten aufrufen, die Sie in der Warteschlange gespeichert haben, wenn die Aufgabe aus der Warteschlange entfernt wird. Sobald die Aufgabe abgeschlossen ist, markiert bullmq sie als erledigt, zieht eine weitere Aufgabe aus der Warteschlange und führt sie aus.

Öffnen Sie index.js in Ihrem Editor:

nano index.js

Fügen Sie in Ihrer Datei index.js die hervorgehobenen Zeilen hinzu, um eine Warteschlange in Redis mit bullmq zu erstellen:

...
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);
}
...

Sie beginnen, die Queue-Klasse aus bullmq zu extrahieren, die verwendet wird, um eine Warteschlange in Redis zu erstellen. Dann setzen Sie die Variable redisOptions auf ein Objekt mit Eigenschaften, die die Instanz der Queue-Klasse verwenden wird, um eine Verbindung mit Redis herzustellen. Sie setzen den Wert der Eigenschaft host auf localhost, da Redis auf Ihrem lokalen Rechner läuft.

Als nächstes setzen Sie die Variable imageJobQueue auf eine Instanz der Klasse Queue, wobei der Name der Warteschlange als erstes Argument und ein Objekt als zweites Argument übergeben werden. Das Objekt hat eine Eigenschaft connection mit dem Wert, der auf ein Objekt in der Variablen redisOptions gesetzt ist. Nach der Instanziierung der Klasse Queue wird eine Warteschlange namens imageJobQueue in Redis erstellt.

Zuletzt definieren Sie die Funktion addJob(), die Sie verwenden werden, um einen Job in der imageJobQueue hinzuzufügen. Die Funktion nimmt einen Parameter job entgegen, der die Informationen über den Job enthält (Sie werden die Funktion addJob() mit den Daten aufrufen, die Sie in einer Warteschlange speichern möchten). In der Funktion rufen Sie die Methode add() der imageJobQueue auf, wobei der Name des Jobs als erstes Argument und die Jobdaten als zweites Argument übergeben werden.

Fügen Sie den hervorgehobenen Code hinzu, um die Funktion addJob() aufzurufen und einen Job in die Warteschlange hinzuzufügen:

...
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 rufen Sie die Funktion addJob() mit einem Objekt auf, das den Job beschreibt. Das Objekt hat das Attribut type mit einem Wert, der dem Namen des Jobs entspricht. Die zweite Eigenschaft, image, ist auf ein Objekt gesetzt, das die Bilddaten enthält, die der Benutzer hochgeladen hat. Da die Bilddaten in image.data als Puffer (binäre Form) vorliegen, rufen Sie die Methode toString() von JavaScript auf, um sie in einen String umzuwandeln, der in Redis gespeichert werden kann. Dies setzt die Eigenschaft data als Ergebnis. Die Eigenschaft image ist auf den Namen des hochgeladenen Bildes (einschließlich der Bildextension) gesetzt.

Sie haben nun die Informationen definiert, die bullmq benötigt, um diesen Job später auszuführen. Je nach Ihrem Job können Sie mehr oder weniger Jobinformationen hinzufügen.

Speichern Sie Ihre Datei und schließen Sie sie.

Erstellen Sie anschließend die Datei utils.js und öffnen Sie sie, die den Code für die Bildverarbeitung enthält:

nano utils.js

In Ihrer Datei utils.js fügen Sie den folgenden Code hinzu, um die Funktion zur Verarbeitung eines Bildes zu definieren:

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

function processUploadedImages(job) {
}

module.exports = { processUploadedImages };

Sie importieren die Module, die zum Verarbeiten von Bildern und Berechnen von Pfaden erforderlich sind, in den ersten beiden Zeilen. Dann definieren Sie die processUploadedImages()-Funktion, die die zeitintensive Bildverarbeitungsaufgabe enthält. Diese Funktion nimmt einen job-Parameter entgegen, der befüllt wird, wenn der Worker die Jobdaten aus der Warteschlange abruft, und ruft dann die processUploadedImages()-Funktion mit den Warteschlangendaten auf. Außerdem exportieren Sie die processUploadedImages()-Funktion, damit Sie darauf in anderen Dateien verweisen können.

Speichern und schließen Sie Ihre Datei.

Kehren Sie zur Datei index.js zurück:

nano index.js

Kopieren Sie die markierten Zeilen aus der Datei index.js und löschen Sie sie dann aus dieser Datei. Sie benötigen den kopierten Code gleich, also speichern Sie ihn in die Zwischenablage. Wenn Sie nano verwenden, können Sie diese Zeilen markieren und mit der rechten Maustaste klicken, um sie zu kopieren:

...
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");
});

Die post-Methode für die upload-Route wird nun wie folgt aussehen:

...
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");
});
...

Speichern und schließen Sie diese Datei, und öffnen Sie dann die Datei utils.js:

nano utils.js

In Ihrer Datei utils.js fügen Sie die gerade kopierten Zeilen für den /upload-Route-Callback in die processUploadedImages-Funktion ein:

...
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++;
  };
}
...

Da Sie den Code zur Verarbeitung eines Bildes verschoben haben, müssen Sie ihn aktualisieren, um die Bilddaten aus dem job-Parameter der zuvor definierten processUploadedImages()-Funktion zu verwenden.

Um das zu tun, fügen Sie die markierten Zeilen unten hinzu und aktualisieren Sie sie:

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`);
  ...
}

Du wandelst die stringifizierte Version der Bilddaten mit der Buffer.from()-Methode zurück in Binärdaten um. Dann aktualisierst du path.parse() mit einem Verweis auf den im Queue gespeicherten Bildnamen. Danach aktualisierst du die sharp()-Methode, um die im imageFileData-Variablen gespeicherten Bilddaten zu verwenden.

Die vollständige utils.js-Datei wird jetzt wie folgt aussehen:

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 };

Speichere und schließe deine Datei und kehre dann zu index.js zurück:

nano index.js

Die sharp-Variable wird nicht mehr als Abhängigkeit benötigt, da das Bild jetzt in der utils.js-Datei verarbeitet wird. Lösche die markierte Zeile aus der Datei:

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

Speichere und schließe deine Datei.

Du hast nun die Funktionalität definiert, um eine Warteschlange in Redis zu erstellen und einen Job hinzuzufügen. Du hast auch die processUploadedImages()-Funktion definiert, um hochgeladene Bilder zu verarbeiten.

Die verbleibende Aufgabe besteht darin, einen Consumer (oder Worker) zu erstellen, der einen Job aus der Warteschlange abruft und die processUploadedImages()-Funktion mit den Jobdaten aufruft.

Erstelle eine worker.js-Datei in deinem Editor:

nano worker.js

Füge in deiner worker.js-Datei den folgenden Code hinzu:

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 der ersten Zeile importierst du die Worker-Klasse aus bullmq; beim Instanziieren wird dies einen Worker starten, der Jobs aus der Warteschlange in Redis abruft und ausführt. Anschließend verweist du auf die processUploadedImages()-Funktion aus der utils.js-Datei, damit der Worker die Funktion mit den Daten in der Warteschlange aufrufen kann.

Sie definieren eine workerHandler()-Funktion, die einen job-Parameter enthält, der die Jobdaten in der Warteschlange enthält. In der Funktion protokollieren Sie, dass der Job gestartet wurde, rufen dann processUploadedImages() mit den Jobdaten auf. Danach protokollieren Sie eine Erfolgsmeldung und geben null zurück.

Um dem Arbeiter zu erlauben, eine Verbindung zu Redis herzustellen, einen Job aus der Warteschlange abzurufen und workerHandler() mit den Jobdaten aufzurufen, fügen Sie die folgenden Zeilen zur Datei hinzu:

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

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

console.log("Worker started!");

Hier setzen Sie die Variable workerOptions auf ein Objekt, das die Verbindungseinstellungen von Redis enthält. Sie setzen die Variable worker auf eine Instanz der Klasse Worker, die die folgenden Parameter annimmt:

• imageJobQueue: der Name der Jobwarteschlange.
• workerHandler: die Funktion, die ausgeführt wird, nachdem ein Job aus der Redis-Warteschlange abgerufen wurde.
• workerOptions: die Redis-Konfigurationseinstellungen, die der Arbeiter verwendet, um eine Verbindung mit Redis herzustellen.

Zuletzt protokollieren Sie eine Erfolgsmeldung.

Nachdem Sie die Zeilen hinzugefügt haben, speichern und schließen Sie Ihre Datei.

Sie haben jetzt die Funktionalität des bullmq-Arbeiters definiert, um Jobs aus der Warteschlange abzurufen und auszuführen.

In Ihrem Terminal entfernen Sie die Bilder im Verzeichnis public/images, damit Sie für das Testen Ihrer Anwendung von vorne beginnen können:

rm public/images/*

Als nächstes führen Sie die Datei index.js aus:

node index.js

Die Anwendung wird gestartet:

Output
Server running on port 3000

Sie starten jetzt den Arbeiter. Öffnen Sie eine zweite Terminal-Sitzung und navigieren Sie zum Projektverzeichnis:

cd image_processor/

Starten Sie den Arbeiter mit folgendem Befehl:

node worker.js

Der Arbeiter wird starten:

Output
Worker started!

Öffnen Sie http://localhost:3000/ in Ihrem Browser. Drücken Sie die Datei auswählen-Schaltfläche und wählen Sie die Datei underwater.png von Ihrem Computer aus, drücken Sie dann die Schaltfläche Bild hochladen.

Sie erhalten möglicherweise eine sofortige Antwort, die Ihnen mitteilt, dass Sie die Seite nach einigen Sekunden aktualisieren sollen:

Alternativ erhalten Sie möglicherweise eine sofortige Antwort mit einigen verarbeiteten Bildern auf der Seite, während andere noch verarbeitet werden:

Sie können die Seite einige Male aktualisieren, um alle skalierten Bilder zu laden.

Kehren Sie zum Terminal zurück, in dem Ihr Arbeiter läuft. Dieses Terminal wird eine Nachricht haben, die der folgenden entspricht:

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

Die Ausgabe bestätigt, dass bullmq den Job erfolgreich ausgeführt hat.

Ihre App kann auch dann zeitintensive Aufgaben auslagern, wenn der Arbeiter nicht läuft. Um dies zu demonstrieren, stoppen Sie den Arbeiter im zweiten Terminal mit CTRL+C.

In Ihrer initialen Terminal-Sitzung stoppen Sie den Express-Server und entfernen die Bilder in public/images:

rm public/images/*

Danach starten Sie den Server erneut:

node index.js

In Ihrem Browser besuchen Sie http://localhost:3000/ und laden das Bild underwater.png erneut hoch. Wenn Sie auf den Pfad /result umgeleitet werden, werden die Bilder nicht auf der Seite angezeigt, weil der Arbeiter nicht läuft:

Kehren Sie zum Terminal zurück, in dem Sie den Arbeiter ausgeführt haben, und starten Sie den Arbeiter erneut:

node worker.js

Die Ausgabe wird der folgenden entsprechen, was Ihnen mitteilt, dass der Job gestartet wurde:

Output
Worker started!
Starting job: processUploadedImages

Nachdem der Job abgeschlossen wurde und die Ausgabe eine Zeile enthält, die Fertiger Job: processUploadedImages lautet, aktualisieren Sie den Browser. Die Bilder werden nun geladen:

Stoppen Sie den Server und den Worker.

Sie können nun eine zeitintensive Aufgabe im Hintergrund auslagern und sie asynchron mit bullmq ausführen. Im nächsten Schritt werden Sie ein Dashboard einrichten, um den Status der Warteschlange zu überwachen.

Schritt 4 — Hinzufügen eines Dashboards zur Überwachung von bullmq-Warteschlangen

In diesem Schritt werden Sie das Paket bull-board verwenden, um die Jobs in der Redis-Warteschlange über ein visuelles Dashboard zu überwachen. Dieses Paket erstellt automatisch eine Benutzeroberfläche (UI)-Dashboard, das Informationen über die bullmq-Jobs anzeigt und organisiert, die in der Redis-Warteschlange gespeichert sind. Mit Ihrem Browser können Sie die Jobs überwachen, die abgeschlossen, wartend oder fehlgeschlagen sind, ohne die Redis-Befehlszeilenschnittstelle im Terminal öffnen zu müssen.

Öffnen Sie die Datei index.js in Ihrem Texteditor:

nano index.js

Fügen Sie den hervorgehobenen Code hinzu, um bull-board zu importieren:

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

Im vorherigen Code importieren Sie die Methode createBullBoard() aus bull-board. Sie importieren auch BullMQAdapter, der bull-board den Zugriff auf bullmq-Queues ermöglicht, sowie ExpressAdapter, der Express die Anzeige des Dashboards ermöglicht.

Fügen Sie anschließend den markierten Code hinzu, um bull-board mit bullmq zu verbinden:

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

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

const app = express();
...

Zuerst setzen Sie den serverAdapter auf eine Instanz des ExpressAdapter. Danach rufen Sie createBullBoard() auf, um das Dashboard mit den Daten der bullmq-Queue zu initialisieren. Sie übergeben der Funktion ein Objektargument mit den Eigenschaften queues und serverAdapter. Die erste Eigenschaft, queues, akzeptiert ein Array der von Ihnen mit bullmq definierten Queues, hier die imageJobQueue. Die zweite Eigenschaft, serverAdapter, enthält ein Objekt, das eine Instanz des Express Server Adapters akzeptiert. Danach setzen Sie den Pfad /admin, um auf das Dashboard mit der Methode setBasePath() zuzugreifen.

Fügen Sie anschließend das Middleware für den serverAdapter für den Pfad /admin hinzu:

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

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

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

Die vollständige Datei index.js sieht wie folgt aus:

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");
});

Nachdem Sie die Änderungen abgeschlossen haben, speichern und schließen Sie Ihre Datei.

Führen Sie die index.js-Datei aus:

node index.js

Kehren Sie zu Ihrem Browser zurück und besuchen Sie http://localhost:3000/admin. Das Dashboard wird geladen:

Im Dashboard kannst du den Jobtyp, die verbrauchten Daten und weitere Informationen zum Job überprüfen. Du kannst auch zu anderen Tabs wechseln, wie zum Beispiel zum Abgeschlossen-Tab für Informationen zu abgeschlossenen Jobs, zum Fehlgeschlagen-Tab für mehr Informationen zu fehlgeschlagenen Jobs und zum Pausiert-Tab für weitere Informationen zu pausierten Jobs.

Du kannst jetzt das bull-board-Dashboard verwenden, um Warteschlangen zu überwachen.

Fazit

In diesem Artikel hast du eine zeitaufwändige Aufgabe mithilfe von bullmq in eine Job-Warteschlange ausgelagert. Zuerst hast du, ohne bullmq zu verwenden, eine App mit einer zeitaufwändigen Aufgabe erstellt, die einen langsamen Anfrage-/Antwortzyklus hat. Dann hast du bullmq verwendet, um die zeitaufwändige Aufgabe auszulagern und asynchron auszuführen, was den Anfrage-/Antwortzyklus beschleunigt. Danach hast du bull-board verwendet, um ein Dashboard zu erstellen und bullmq-Warteschlangen in Redis zu überwachen.

Du kannst die bullmq-Dokumentation besuchen, um mehr über Funktionen von bullmq zu erfahren, die in diesem Tutorial nicht behandelt wurden, wie z.B. das Planen, Priorisieren oder Wiederholen von Jobs sowie das Konfigurieren von Konkurrenz-Einstellungen für Arbeiter. Du kannst auch die bull-board-Dokumentation besuchen, um mehr über die Dashboard-Funktionen zu erfahren.