如何在Puppeteer异步任务完成后触发RabbitMQ消息Ack？

嘿，这个问题我之前也踩过坑！核心就是要确保Puppeteer的所有异步操作完全结束后，再手动给RabbitMQ发送Ack，不然消息会一直卡在Unacked状态，Worker看起来就像挂住了一样。

关键解决思路 #

RabbitMQ如果开启手动确认（noAck: false），就会等待Worker明确发送Ack才会把消息从队列中移除。如果你的Worker里异步操作没等完就结束了，或者没正确调用Ack，RabbitMQ会一直认为这个消息在处理中，Worker就会被“占用”着，没法处理下一条消息。

具体实现代码示例 #

我给你写个完整的Worker示例，用async/await来确保异步流程顺序执行：

const amqp = require('amqplib');
const puppeteer = require('puppeteer');

// 封装消息处理逻辑
async function handleUrlTask(msg) {
  if (!msg) return;
  
  const targetUrl = msg.content.toString();
  let browser;

  try {
    // 启动Puppeteer，等待浏览器实例创建完成
    browser = await puppeteer.launch({
      // 根据你的环境配置，比如无头模式、沙箱设置等
      headless: 'new',
      args: ['--no-sandbox', '--disable-setuid-sandbox']
    });
    
    const page = await browser.newPage();
    // 等待页面加载完成（可以根据需求调整waitUntil参数）
    await page.goto(targetUrl, { waitUntil: 'networkidle2' });
    
    // 这里可以添加你的业务逻辑：比如截图、提取页面内容等
    const pageTitle = await page.title();
    console.log(`Successfully processed ${targetUrl}, title: ${pageTitle}`);
    
    // 关闭浏览器
    await browser.close();
    
    // 所有异步操作完成后，发送Ack确认消息处理完成
    await msg.channel.ack(msg);
  } catch (error) {
    console.error(`Failed to process ${targetUrl}:`, error);
    // 处理错误：如果是可重试的错误，可以设置第三个参数为true让消息重新入队
    // 如果是不可重试的错误，就设为false，让消息进入死信队列
    if (browser) await browser.close();
    await msg.channel.nack(msg, false, false);
  }
}

// 启动Worker
async function startRabbitMQWorker() {
  try {
    const connection = await amqp.connect('amqp://localhost');
    const channel = await connection.createChannel();
    const queueName = 'url_processing_queue';
    
    // 声明队列（确保队列存在）
    await channel.assertQueue(queueName, { durable: true });
    // 设置prefetch，控制每个Worker同时处理的消息数，避免资源过载
    await channel.prefetch(1);
    
    console.log(`Worker is listening on queue: ${queueName}`);
    
    // 监听队列，用async回调确保等待异步处理完成
    channel.consume(queueName, async (msg) => {
      await handleUrlTask(msg);
    }, { noAck: false }); // 必须设置为false，开启手动确认
  } catch (error) {
    console.error('Failed to start worker:', error);
  }
}

startRabbitMQWorker();

几个重要的注意点 #

• noAck必须设为false：这是手动确认的前提，否则RabbitMQ会自动确认消息，不管你有没有处理完。
• 用async/await包裹所有异步操作：Puppeteer的所有方法都是异步的，必须等待它们执行完毕再发送Ack，不然消息会提前被确认，或者Worker挂在未完成的任务上。
• 错误处理不能少：如果Puppeteer操作失败，一定要调用nack来拒绝消息，避免消息一直卡在Unacked状态。根据错误类型决定是否重新入队。
• 合理设置prefetch：比如设为1，让每个Worker一次只处理一个消息，避免同时处理多个导致内存/CPU过载，也能保证消息处理的顺序性。

如果你的场景是需要并行处理多个URL，记得用Promise.all等待所有并行任务完成后再发送Ack，比如：

// 并行处理多个URL的示例
const urls = ['https://example.com', 'https://test.com'];
await Promise.all(urls.map(async (url) => {
  const page = await browser.newPage();
  await page.goto(url);
  await page.close();
}));
await browser.close();
await msg.channel.ack(msg);

这样就能确保所有异步操作都完成后，才告诉RabbitMQ可以移除这条消息，Worker也能正常处理下一条任务啦！

内容的提问来源于stack exchange，提问作者Brian Zhou