RabbitMQ 消息队列 优化发送邮件

express 发送邮件

• 最简单的异步发送邮件方法
• 为何要使用 RabbitMQ？
• 如何在 Node 项目中集成 RabbitMQ？

一、 不用 await 发送邮件

在实际开发之前，不妨先思考下，我们最终的目的是为了让邮件异步发送。那发送邮件这里有个await，我们干嘛不直接把这个await去掉，这不就完事了吗？这样不就是不等待邮件发送完成，直接提示成功了吗？

我要告诉大家，你想的一点也没错，这样做 100% 可以的。当然这样做，也会造成一些问题。因为不等待异步执行的结果，如果发送邮件出现错误了，会导致catch里无法捕获错误。

1、参考案例：

router.post('/sign_up', validateCaptcha, async function (req, res) {try {const body = {email: req.body.email,username: req.body.username,nickname: req.body.nickname,password: req.body.password,sex: 2,role: 0}const user = await User.create(body);delete user.dataValues.password;         // 删除密码// 请求成功，删除验证码，防止重复使用await delKey(req.body.captchaKey);const html = `您好，<span style="color: red">${user.nickname}。</span><br><br>恭喜，您已成功注册会员！<br><br>xw`await sendMail(user.email, '「xw」的注册成功通知', html);success(res, '创建用户成功。', {user}, 201);} catch (error) {failure(res, error);}
});

2、去掉sendMail前的await

router.post('/sign_up', validateCaptcha, async function (req, res) {try {const body = {email: req.body.email,username: req.body.username,nickname: req.body.nickname,password: req.body.password,sex: 2,role: 0}const user = await User.create(body);delete user.dataValues.password;         // 删除密码// 请求成功，删除验证码，防止重复使用await delKey(req.body.captchaKey);const html = `您好，<span style="color: red">${user.nickname}。</span><br><br>恭喜，您已成功注册会员！<br><br>xw`sendMail(user.email, '「xw」的注册成功通知', html);success(res, '创建用户成功。', {user}, 201);} catch (error) {failure(res, error);}
});

结果 ：可以注册成功 但是终端崩溃了，再去请求其他任何接口，都是无法访问的了

3、原因就是因为，这是一个异步操作，我们没有等待异步操作的结果，所以没法在这里捕获异常。解决方法也很简单，打开utils/mail.js，给发送邮件函数，加上try…catch即可。

/*** 发送邮件* @param email* @param subject* @param html* @returns {Promise<void>}*/
const sendMail = async (email, subject, html) => {try {await transporter.sendMail({from: process.env.MAILER_USER,to: email,subject,html,});} catch (error) {console.log('邮件发送失败：', error);}
};

最后运行：虽然现在还是报错了，但是程序自身并没有崩溃。你去访问其他接口，也是完全不受影响的。

那么现在，我们就用非常简单的方式，实现了异步发送邮件了。这样是完全可行的

二、 为何要使用 RabbitMQ

既然这么简单就能实现了，那为何还要使用RabbitMQ呢？我觉得主要有以下三点目的

• 1、解耦：将发送邮件的功能独立出去，这样即使邮件服务出现问题，也不会直接影响主应用的性能和可用性。
• 2、增加并发：对于高并发的情况，将任务放入队列中可以起到缓冲作用，增加程序的吞吐量，防止后端服务因瞬间请求过多而崩溃。
• 3、错误处理：如果消息队列中的任务处理失败，可以进行重试，确保任务最终能够成功完成。
  所以，对于非常简单的功能，你确实可以不写await来达到异步的目的。但对于大型程序来说，为了有更好的性能，增加并发处理能力，提高错误处理的可靠性，使用消息队列是一个更好的选择。

所以，对于非常简单的功能，你确实可以不写await来达到异步的目的。但对于大型程序来说，为了有更好的性能，增加并发处理能力，提高错误处理的可靠性，使用消息队列是一个更好的选择。

三. 在 Node 项目中集成 RabbitMQ

3.1. 环境变量

打开项目的.env文件，加入

RABBITMQ_URL=amqp://admin:xw@localhost

这样连接到RabbitMQ的时候，相关的信息，就可以从环境变量读取。

.env.example中加入

RABBITMQ_URL=

README.md中加入对应的说明

RABBITMQ_URL=你的连接- `RABBITMQ_URL`配置为消息队列服务器地址。

3.2.、连接到 RabbitMQ

新建utils/rabbit-mq.js，里面创建一个连接

const amqp = require('amqplib');
const sendMail = require('./mail');// 创建全局的 RabbitMQ 连接和通道
let connection;
let channel;/*** 连接到 RabbitMQ* @returns {Promise<*>}*/
const connectToRabbitMQ = async () => {if (connection && channel) return;  // 如果已经连接，直接返回try {connection = await amqp.connect(process.env.RABBITMQ_URL);channel = await connection.createChannel();await channel.assertQueue('mail_queue', { durable: true });} catch (error) {console.error('RabbitMQ 连接失败：', error);}
};

• 顶部做了相关的引用。
• 创建了全局的连接和通道。
• 如果已经连接了，就直接返回。如果没有连接就连上，并创建通道。
• 创建了队列，名字叫做：mail_queue。
• 使用了durable: true，表示队列需要持久化。

3.3、 邮件队列生产者

接着继续建一个方法，来发送邮件

/*** 邮件队列生产者（发送消息）*/
const mailProducer = async (msg) => {try {await connectToRabbitMQ(); // 确保已连接channel.sendToQueue('mail_queue', Buffer.from(JSON.stringify(msg)), { persistent: true });} catch (error) {console.error('邮件队列生产者错误：', error);}
};

• 这里就直接将消息，发送到队列中。
• 注意用JSON.stringify转了一下，说明我们传过来的msg将会是对象格式。
• 使用了persistent: true，表示消息需要持久化。

3.4、 邮件队列消费者

/*** 邮件队列消费者（接收消息）*/
const mailConsumer = async () => {try {await connectToRabbitMQ();channel.consume('mail_queue',async (msg) => {const message = JSON.parse(msg.content.toString());await sendMail(message.to, message.subject, message.html);}, {noAck: true,});} catch (error) {console.error('邮件队列消费者错误：', error);}
};module.exports = {mailProducer,mailConsumer,
};

• 我们监听了mail_queue队列。
• 如过收到消息了，就执行发送邮件。
• 使用了noAck: true，表示自动确认消息。

四. 实际运用

4.1、启用生产者

打开routes/auth.js

// const sendMail = require('../utils/mail');
const { mailProducer } = require('../utils/rabbit-mq');/*** 用户注册* POST /auth/sign_up*/
router.post('/sign_up', validateCaptcha, async function (req, res) {try {// ...// 将邮件发送请求放入队列const msg = {to: user.email,subject: '「xw」的注册成功通知',html: `您好，<span style="color: red">${user.nickname}</span>。<br><br>恭喜，您已成功注册会员！<br><br>xw`,};await mailProducer(msg);success(res, '创建用户成功。', { user }, 201);} catch (error) {failure(res, error);}
});

• 顶部将发送邮件的方法去掉，并引用一下生产者。
• 将要发送的信息，改为对象格式。
• 调用生产者方法，就将信息发送到队列中了。

4.2、 启动消费者

根目录的app.js。在里面加上：

require('dotenv').config();// 启动邮件消费者
const { mailConsumer } = require('./utils/rabbit-mq');
(async () => {await mailConsumer();console.log('邮件消费者已启动');
})();

这样只要 Node 项目一启动，消费者就会一直自动运行起来。在真实大型项目里，为了不影响主程序的稳定运行。更好的方式，应该将发送邮件解耦了，可以让消费者在专门的程序中独立启动。

解耦封装

1、封装生产者、队列/utils/rabbit-mq.js

const amqp = require('amqplib');
const sendMail = require('./mail');// 创建全局的 RabbitMQ 连接和通道
let connection;
let channel;// 封装一个重试连接的函数，增加连接的稳定性
const connectWithRetry = async (url, retries = 5, delay = 5000) => {let attempt = 0;while (attempt < retries) {try {return await amqp.connect(url);} catch (error) {attempt++;console.error(`RabbitMQ 连接尝试 ${attempt} 失败:`, error);if (attempt < retries) {await new Promise(resolve => setTimeout(resolve, delay));}}}throw new Error('无法连接到 RabbitMQ，已达到最大重试次数');
};/*** 连接到 RabbitMQ* @returns {Promise<*>}*/
const connectToRabbitMQ = async () => {if (connection && channel) return;try {connection = await connectWithRetry(process.env.RABBITMQ_URL);channel = await connection.createChannel();// 监听连接关闭事件，方便处理异常connection.on('close', () => {console.warn('RabbitMQ 连接已关闭，尝试重新连接...');connection = null;channel = null;});// 监听连接错误事件，增强错误处理能力connection.on('error', (err) => {console.error('RabbitMQ 连接发生错误:', err);});await channel.assertQueue('mail_queue', { durable: true });} catch (error) {console.error('RabbitMQ 连接失败：', error);throw error;}
};/*** 邮件队列生产者（发送消息）*/
const mailProducer = async (msg) => {try {await connectToRabbitMQ(); // 确保已连接// 消息持久化设置，提高消息可靠性const options = { persistent: true };const sent = channel.sendToQueue('mail_queue', Buffer.from(JSON.stringify(msg)), options);if (!sent) {console.warn('消息未能立即入队，等待下次机会');}} catch (error) {console.error('邮件队列生产者错误：', error);throw error;}
};/*** 邮件队列消费者（接收消息）*/
const mailConsumer = async () => {try {await connectToRabbitMQ();// 消费消息时，手动确认消息，避免消息丢失channel.consume('mail_queue', async (msg) => {if (msg) {try {const message = JSON.parse(msg.content.toString());await sendMail(message.to, message.subject, message.html);channel.ack(msg); // 手动确认消息} catch (error) {console.error('处理邮件消息时出错:', error);channel.nack(msg, false, true); // 消息处理失败，重新入队}}}, { noAck: false }); // 关闭自动确认console.log('邮件队列消费者已开始监听');} catch (error) {console.error('邮件队列消费者错误：', error);throw error;}
};module.exports = {mailProducer,mailConsumer,
};

2、封装消费者/utils/mail-consumer.js

require('dotenv').config();
const { mailConsumer } = require('./rabbit-mq');// 封装启动消费者的函数，方便后续扩展和错误处理
const startMailConsumer = async () => {try {await mailConsumer();console.log('邮件消费者已启动');} catch (error) {console.error('启动邮件消费者时出错:', error);process.exit(1);}
};// 启动消费者
startMailConsumer();// 监听进程信号，优雅关闭消费者
process.on('SIGINT', async () => {console.log('收到 SIGINT 信号，正在优雅关闭邮件消费者...');try {// 这里可以添加关闭连接和通道的逻辑process.exit(0);} catch (error) {console.error('关闭邮件消费者时出错:', error);process.exit(1);}
});

如果在app.js里面使用了消费者 把代码去掉

3、安装pm2

npm i pm2

4、根目录新建cosystem.config.js

module.exports = {apps: [{name: "express-app",script: "./bin/www", // 实际路径watch: process.env.NODE_ENV === 'development', // 根据环境变量决定是否开启监听interpreter: "node",env: {NODE_ENV: "development"},env_production: {NODE_ENV: "production"}},{name: "mail-consumer",script: "./utils/mail-consumer.js", // 实际路径interpreter: "node",env: {NODE_ENV: "development"},env_production: {NODE_ENV: "production"}}]
};

4、本地运行

pm2 start ecosystem.config.js

pm2 简单的方法

命令	说明	示例
pm2 start <app.js>	启动一个 Node.js 应用程序。可指定 JavaScript 文件、JSON 配置文件或其他可执行文件。	pm2 start app.js
pm2 list	列出所有由 PM2 管理的应用程序，显示应用程序的状态、进程 ID、名称等信息。	pm2 list
pm2 stop <app_name_or_id>	停止指定名称或 ID 的应用程序，但不会从 PM2 的列表中删除。	pm2 stop my_app 或 pm2 stop 0
pm2 stop all	停止所有由 PM2 管理的应用程序。	pm2 stop all
pm2 restart <app_name_or_id>	重启指定名称或 ID 的应用程序。	pm2 restart my_app 或 pm2 restart 0
pm2 restart all	重启所有由 PM2 管理的应用程序。	pm2 restart all
pm2 delete <app_name_or_id>	从 PM2 的管理列表中删除指定名称或 ID 的应用程序，同时停止该应用程序。	pm2 delete my_app 或 pm2 delete 0
pm2 delete all	从 PM2 的管理列表中删除所有应用程序。	pm2 delete all
pm2 show <app_name_or_id>	显示指定名称或 ID 的应用程序的详细信息，如环境变量、执行模式等。	pm2 show my_app 或 pm2 show 0
pm2 logs <app_name_or_id>	显示指定名称或 ID 的应用程序的日志信息。若不指定则显示所有应用日志。	pm2 logs my_app 或 pm2 logs 0
pm2 monit	实时监控由 PM2 管理的所有应用程序的 CPU 和内存使用情况。	pm2 monit
pm2 save	保存当前 PM2 管理的应用程序列表，以便在系统重启后自动恢复这些应用。	pm2 save
pm2 resurrect	恢复之前使用 pm2 save 保存的应用程序列表。	pm2 resurrect

5、 宝塔中部署

启动选项

 pm2 start ecosystem.config.js --no-daemon