当您构建大型 Web 应用程序时，速度是首要考虑的因素。用户不想再等待很长时间才能得到响应，他们也不应该这样做。但有些流程需要时间，而且无法加快或消除。

消息队列通过为通常的请求-响应过程提供额外的分支来帮助解决这个问题。这个额外的分支有助于确保用户能够立即得到响应，并且可以暂时完成耗时的流程。大家都高高兴兴回家了。

本文将重点介绍什么是消息队列以及如何通过构建一个非常简单的应用程序来开始使用它们。您应该熟悉 Node.js 的基础知识，并且应该在本地或云实例上安装 Redis。在此处了解如何安装 Redis 。

（更|多优质内|容：java567 点 c0m）

什么是队列？

队列是一种数据结构，允许您按顺序存储实体。队列使用先进先出 (FIFO) 原则。

计算机科学中的队列概念与日常生活中人们排队取东西的队列概念相同。您从后面加入队列，等到轮到您时，然后在有人照顾后从前面离开队列。

在计算机科学中，当 API 请求等流程正在运行时，您需要从当前流程中删除某个任务（例如发送电子邮件），您可以将其推送到队列并继续该流程。

什么是工作？

作业是队列上使用的任何数据，通常是类似 JSON 的对象。

正如本文封面图片所示，您可以将工作视为机场队列中的每个人。每个人都拿着一个公文包，其中包含特定数据和其他说明（护照，可能还需要医疗文件），这些说明在轮到他们时会有所帮助。

加入队列的新人将从后面进入（作为最后一个人），人们将从前面受到照顾。这也是处理作业的方式，每个作业都包含将用于其处理的数据。新的工作是从后面添加的，而工作是从前面删除的。

什么是工作生产者？

作业生产者是将作业添加到队列的任何代码段。在现实生活中，这将是机场的保安人员，为人们指引方向，告诉他们为了不同的目的而加入哪个队列。

作业生产者可以独立于作业消费者而存在。这意味着在微服务设置中，特定服务可能只关心将作业添加到队列中，而不关心之后如何处理它们。

什么是工人（工作消费者）？

工作人员或作业使用者是可以执行作业的进程或函数。将工作人员想象为银行出纳员，负责在银行排队的人。当第一个人进来时，他们作为队列中唯一的人加入队列。然后收银员招呼他们，队列就被清空了。

收银员要求该人提供用于处理交易的具体详细信息。当收银员接待该顾客时，可能还有另外四名顾客在排队。他们将一直排队，直到收银员处理完第一位顾客，然后再叫下一位顾客。这与队列工作人员的过程相同 - 他们选择队列中的第一个作业并处理它。

什么是失败的工作？

很多时候，某些作业可能会在处理过程中失败。

以下是作业可能失败的一些原因：

输入数据无效或丢失：当要处理的作业所需的数据丢失时，作业将失败。例如，如果没有收件人的电子邮件地址，发送电子邮件的作业将会失败。超时：如果作业花费的时间比平常长，队列机制可能会导致作业失败。这可能是由于作业的依赖性问题或其他原因造成的，但通常您不希望单个作业永远运行。网络或基础设施问题：这些问题几乎超出您的控制范围，但它们确实会发生。例如，数据库连接错误会导致作业失败。依赖性问题：有时一项工作需要依赖一些外部资源才能正常运行。每当这些其他资源不可用或不成功时，作业就会失败。

当作业失败时，您可以配置队列机制来重试它们。您可以立即重试该作业，也可以在计算出的时间后重试。您可以设置最大尝试次数，建议这样做。如果没有，您最终运行的作业将永远失败。

为什么使用队列？

队列对于在微服务之间创建强大的通信通道非常有用。多个服务可以使用同一个队列。不同的服务可能负责解决不同的问题。当服务完成其任务时，它可以将作业推送到另一个有工作人员等待该作业的服务。该服务将接收它并对数据执行任何需要的操作。

队列对于从进程中卸载繁重的任务也很有用。正如您将在本文中看到的，可以将发送电子邮件等耗时的任务放入队列中，以避免减慢响应时间。

队列有助于避免单点故障。能够失败并可以重试的进程最好使用队列进行处理，在一段时间后可以重试。

如何构建使用队列的简单应用程序

在本文中，我们将使用 Node.js 和Redis构建一个简单的项目。我们将使用Bull库，因为它简化了构建队列系统所涉及的许多复杂性。该项目将有一个端点来发送电子邮件。

创建一个新的 Node.js 项目

 mkdir nodejs-queue-project cd nodejs-queue-project npm init -y

上面的命令将创建一个名为 的新文件夹nodejs-queue-project并package.json在其中创建一个文件。该package.json文件应如下所示：

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

 npm i express @types/express @types/node body-parser ts-node ts-lint typescript nodemon nodemailer @types/nodemailer

上面的命令将安装项目所需的不同包和依赖项。

安装后，您可以更新scripts您的部分package.json以获取dev命令。您的整个package.json文件现在应该如下所示：

 {   "name": "nodejs-queue-project",   "version": "1.0.0",   "description": "",   "main": "index.js",   "scripts": {     "dev": "nodemon src/app.ts"   },   "keywords": [],   "author": "",   "license": "ISC",   "dependencies": {     "@types/express": "^4.17.17",     "@types/node": "^20.3.3",     "@types/nodemailer": "^6.4.8",     "body-parser": "^1.20.2",     "express": "^4.18.2",     "nodemailer": "^6.9.3",     "nodemon": "^2.0.22",     "ts-lint": "^4.5.1",     "ts-node": "^10.9.1",     "typescript": "^5.1.6"   } }

上面的文件显示了所有已安装的依赖项。当您使用脚本时，该npm run dev命令将运行dev。

如何构建端点

要做的第一件事是创建一个名为 的新文件夹src。该文件夹将包含您的所有代码文件。它将包含的第一个文件是应用程序的根文件 -文件app.ts中定义的文件package.json。

我们将使用该app.ts文件导入所需的包并创建一个具有单个端点的简单服务器来发送电子邮件，如下所示：

 import express from "express"; import bodyParser from "body-parser"; import nodemailer from "nodemailer";  const app = express();  app.use(bodyParser.json());  app.post("/send-email", async (req, res) => {   const { from, to, subject, text } = req.body;    // Use a test account as this is a tutorial   const testAccount = await nodemailer.createTestAccount();    const transporter = nodemailer.createTransport({     host: "smtp.ethereal.email",     port: 587,     secure: false,     auth: {       user: testAccount.user,       pass: testAccount.pass,     },     tls: {       rejectUnauthorized: false,     },   });    console.log("Sending mail to %s", to);    let info = await transporter.sendMail({     from,     to,     subject,     text,     html: `<strong>${text}</strong>`,   });    console.log("Message sent: %s", info.messageId);   console.log("Preview URL: %s", nodemailer.getTestMessageUrl(info));    res.json({     message: "Email Sent",   }); });  app.listen(4300, () => {   console.log("Server started at //localhost:4300"); }); npm run dev`现在，您可以通过在终端中运行来启动服务器。您应该在终端中看到一条消息。`Server started at //localhost:4300

npm 运行开发消息

您现在可以使用 Postman 等工具测试端点：

Postman 的端点测试

如屏幕截图所示，请求花费了近 4 秒。这对于端点来说非常慢。如果您查看终端，您还应该看到一个 URL，您可以在其中预览已发送的电子邮件。

打开链接即可查看电子邮件的外观。

邮件内容

如何创建队列

为了使该过程更快，可以将电子邮件排队以便稍后发送，并立即将响应发送给用户。

为此，请安装该bull库及其@types库，因为我们将使用它来创建队列。那是：

 npm i bull @types/bull

使用创建新队列bull就像使用Bull队列名称实例化一个新对象一样简单：

 // This goes at the top of your file import Bull from 'bull';  const emailQueue = new Bull("email");

当仅使用队列名称创建队列时，它会尝试使用默认的 Redis 连接 URL：localhost:6379。如果您更喜欢使用不同的 URL，只需将第二个对象Bull作为选项对象传递给该类即可：

 const emailQueue = new Bull("email", {   redis: "localhost:6379", });

此时，您可以创建一个简单的函数来充当作业生产者，并在每次请求到来时将作业添加到队列中。

 type EmailType = {   from: string;   to: string;   subject: string;   text: string; };  const sendNewEmail = async (email: EmailType) => {   emailQueue.add({ ...email }); };

这个新创建的函数sendNewEmail接受一个对象，其中包含要发送的类型为 的新电子邮件的详细信息EmailType。包括电子邮件的发件人电子邮件地址 ( from)、收件人电子邮件地址 ( to)subject以及电子邮件的内容 ( text)。然后它将新作业推送到队列中。

您现在可以使用此功能，而不是在请求期间发送电子邮件。修改端点以执行此操作：

 app.post("/send-email", async (req, res) => {   const { from, to, subject, text } = req.body;    await sendNewEmail({ from, to, subject, text });    console.log("Added to queue");    res.json({     message: "Email Sent",   }); });

此时，代码更简单，过程也更快。该请求只需要大约 40m——比以前快了大约 100 倍。

使用 Postman 进行端点测试

此时，电子邮件已添加到队列中。它将保留在队列中直到被处理。该作业可以由同一应用程序或另一个服务（如果在微服务设置中）处理。

如何处理工作

如果邮件从未离开队列，则该循环是不完整且无用的。我们将创建一个作业使用者来处理作业并清除队列。

Job我们可以通过为接受对象并发送电子邮件的函数创建逻辑来做到这一点：

 const processEmailQueue = async (job: Job) => {   // Use a test account as this is a tutorial   const testAccount = await nodemailer.createTestAccount();    const transporter = nodemailer.createTransport({     host: "smtp.ethereal.email",     port: 587,     secure: false,     auth: {       user: testAccount.user,       pass: testAccount.pass,     },     tls: {       rejectUnauthorized: false,     },   });    const { from, to, subject, text } = job.data;    console.log("Sending mail to %s", to);    let info = await transporter.sendMail({     from,     to,     subject,     text,     html: `<strong>${text}</strong>`,   });    console.log("Message sent: %s", info.messageId);   console.log("Preview URL: %s", nodemailer.getTestMessageUrl(info));      return nodemailer.getTestMessageUrl(info); };

上面的函数接受一个Job对象。该对象具有有用的属性，可显示作业的状态和数据。在这里，我们使用该data属性。

此时，我们所拥有的只是一个函数。它不会自动获取作业，因为它不知道要使用哪个队列。

在将其连接到队列之前，您可以通过发送一些请求来继续将一些作业添加到队列中。您可以通过在以下位置运行此命令来检查当前排队的电子邮件作业redis-cli：

 LRANGE bull:email:wait 0 -1

这会检查电子邮件等待列表，并返回ids等待作业的列表。

Redis 命令行界面

我创造了一些工作只是为了展示工人的实际工作方式。

现在，通过添加以下代码行将工作线程连接到队列：

 emailQueue.process(processEmailQueue);

这就是您的app.ts文件现在应该处理的内容：

 import express from "express"; import bodyParser from "body-parser"; import nodemailer from "nodemailer"; import Bull, { Job } from "bull";  const app = express();  app.use(bodyParser.json());  const emailQueue = new Bull("email", {   redis: "localhost:6379", });  type EmailType = {   from: string;   to: string;   subject: string;   text: string; };  const sendNewEmail = async (email: EmailType) => {   emailQueue.add({ ...email }); };  const processEmailQueue = async (job: Job) => {   // Use a test account as this is a tutorial   const testAccount = await nodemailer.createTestAccount();    const transporter = nodemailer.createTransport({     host: "smtp.ethereal.email",     port: 587,     secure: false,     auth: {       user: testAccount.user,       pass: testAccount.pass,     },     tls: {       rejectUnauthorized: false,     },   });    const { from, to, subject, text } = job.data;    console.log("Sending mail to %s", to);    let info = await transporter.sendMail({     from,     to,     subject,     text,     html: `<strong>${text}</strong>`,   });    console.log("Message sent: %s", info.messageId);   console.log("Preview URL: %s", nodemailer.getTestMessageUrl(info)); };  emailQueue.process(processEmailQueue);  app.post("/send-email", async (req, res) => {   const { from, to, subject, text } = req.body;    await sendNewEmail({ from, to, subject, text });    console.log("Added to queue");    res.json({     message: "Email Sent",   }); });  app.listen(4300, () => {   console.log("Server started at //localhost:4300"); });

保存后，您会注意到服务器重新启动并立即开始发送邮件。这是因为工作人员看到队列并立即开始处理。

服务器发送排队的电子邮件

现在，生产者和工人都活跃起来。每个新的 API 请求都将被推送到队列中，并且工作线程将立即处理它，除非已经有一些待处理的作业。

概括**

我希望本文能帮助您了解什么是消息队列、如何添加作业并创建进程来运行它们，以及如何使用它们来构建更好的 Web 应用程序。

（更|多优质内|容：java567 点 c0m）