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/ ", // 实际路径 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.jspm2 list列出所有由 PM2 管理的应用程序，显示应用程序的状态、进程 ID、名称等信息。pm2 listpm2 stop <app_name_or_id>停止指定名称或 ID 的应用程序，但不会从 PM2 的列表中删除。pm2 stop my_app 或 pm2 stop 0pm2 stop all停止所有由 PM2 管理的应用程序。pm2 stop allpm2 restart <app_name_or_id>重启指定名称或 ID 的应用程序。pm2 restart my_app 或 pm2 restart 0pm2 restart all重启所有由 PM2 管理的应用程序。pm2 restart allpm2 delete <app_name_or_id>从 PM2 的管理列表中删除指定名称或 ID 的应用程序，同时停止该应用程序。pm2 delete my_app 或 pm2 delete 0pm2 delete all从 PM2 的管理列表中删除所有应用程序。pm2 delete allpm2 show <app_name_or_id>显示指定名称或 ID 的应用程序的详细信息，如环境变量、执行模式等。pm2 show my_app 或 pm2 show 0pm2 logs <app_name_or_id>显示指定名称或 ID 的应用程序的日志信息。若不指定则显示所有应用日志。pm2 logs my_app 或 pm2 logs 0pm2 monit实时监控由 PM2 管理的所有应用程序的 CPU 和内存使用情况。pm2 monitpm2 save保存当前 PM2 管理的应用程序列表，以便在系统重启后自动恢复这些应用。pm2 savepm2 resurrect恢复之前使用 pm2 save 保存的应用程序列表。pm2 resurrect 5、 宝塔中部署

启动选项

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