sqs-consumer重复执行函数致S3大量子文件夹生成问题求助

问题核心错误分析 #

你的代码存在几个关键问题，直接导致了重复执行和重复创建文件夹：

• 重复拉取消息：sqs-consumer的核心功能就是自动轮询接收消息，你在handleMessage里又手动调用sqs.receiveMessage，会导致同一消息被多次拉取处理。
• 异步操作未正确处理：createSubDirectory中s3.putObject是异步方法，你没有等待它完成就结束了处理逻辑，sqs-consumer会因为无法感知处理完成时间而重复触发处理。
• 重复删除消息：已经配置了shouldDeleteMessages: true，sqs-consumer会自动在处理成功后删除消息，手动调用deleteMessage会造成逻辑冲突。
• 错误的异常处理逻辑：createSubDirectory的try/catch块逻辑颠倒，try块里打印未定义的err，catch块反而打印成功信息。

修复后的代码 #

1. 修复S3文件夹创建函数

// 改成异步函数，正确处理s3.putObject的异步操作和异常
const createSubDirectory = async (s3BucketName, s3ObjectKey) => {
  const params = { 
    Bucket: s3BucketName, 
    Key: s3ObjectKey, 
    ACL: "public-read", 
    Body: "" // 空内容即可，不需要无效文本
  };

  try {
    await s3.putObject(params).promise(); // 用promise形式等待操作完成
    console.log(`子文件夹创建成功: ${s3BucketName}/${s3ObjectKey}`);
  } catch (err) {
    console.error(`创建文件夹出错: `, err);
    throw err; // 抛出错误让sqs-consumer感知到处理失败，避免自动删消息
  }
};

2. 修复主消费逻辑

const app = Consumer.create({
  queueUrl: process.env.SQS_QUEUE_URL,
  WaitTimeSeconds: 20,
  VisibilityTimeout: 600,
  shouldDeleteMessages: true, // 保持自动删消息配置
  batchSize: 1,
  pollingWaitTimeMs: 10000,
  messageAttributeNames: ["All"],
  // handleMessage改成异步函数，确保consumer能感知处理完成状态
  handleMessage: async (message) => {
    const event = JSON.parse(message.Body);
    console.log("收到消息:", event);
    
    // 直接使用consumer传入的message，不要再手动拉取
    const s3ObjectKey = `${parentDirectory}/${AWS.util.uuid.v4()}/`;
    // 等待文件夹创建完成
    await createSubDirectory(s3BucketName, s3ObjectKey);
    
    // 不需要手动删消息，shouldDeleteMessages: true会自动处理
    console.log(`消息处理完成，将自动删除: ${message.MessageId}`);
  },
  sqs: new AWS.SQS({
    apiVersion: "2012-11-05",
    region: "us-east-1",
    httpOptions: {
      agent: new https.Agent({ keepAlive: true }),
    },
  }),
});

app.on("error", (err) => {
  console.error("消费者全局错误:", err.message);
});

app.on("processing_error", (err) => {
  console.error("消息处理错误:", err.message);
});

app.start();

关键修复点说明 #

• 让handleMessage成为异步函数，并等待所有异步操作完成，确保sqs-consumer能准确判断处理状态，不会重复触发。
• 移除手动调用的sqs.receiveMessage和sqs.deleteMessage，完全依赖sqs-consumer的内置逻辑管理消息的接收和删除。
• 修正createSubDirectory的异步处理和异常逻辑，确保失败时抛出错误，让consumer不会自动删除消息，便于后续重新处理。

内容的提问来源于stack exchange，提问作者kaise white