Node.js中AWS SQS无限轮询接收消息失效问题求助

问题描述 #

我正在尝试使用AWS SQS建立无限连接以轮询消息，但sqs.receiveMessage函数未能在无限循环中正常调用，恳请帮忙排查问题。

参数配置 #

recieveMessageParams: {
            AttributeNames: [
                "SentTimestamp"
            ],
            MaxNumberOfMessages: 1,
            MessageAttributeNames: [
                "All"
            ],
            QueueUrl: queueURL,
            VisibilityTimeout: 43195,
            WaitTimeSeconds: 20
        }

代码实现 #

function sqsReceivemessage(params) {
  while(true) {
    sqs.receiveMessage(params, function (err, data) {
      if (err) {
        console.log("Receive Error", err);
        return err;
      } else if (data) {
        console.log("data recieved")
        let body = JSON.parse(data?.Messages[0].Body);
        let action = data?.Messages[0].MessageAttributes?.action?.StringValue;
        if(action === 'vaxxaEmailAttemptToSend'){
          vaxxaEmailAttemptToSend(body);
        }
        let deleteParams = {
          QueueUrl: queueURL,
          ReceiptHandle: data.Messages[0].ReceiptHandle
        };
        sqs.deleteMessage(deleteParams, function (err, data) {
          if (err) {
            console.log("Delete Error", err);
            return err;
          } else {
            console.log("Message Deleted", data);
            return data;
          }
        });
      }
    });
  }
}

问题分析与解决建议 #

核心问题原因 #

你用了while(true)同步循环，但sqs.receiveMessage是异步回调函数。同步循环会无限占用CPU线程，根本没机会让异步的receiveMessage回调执行，导致看起来函数没被正常调用。

正确实现方式（回调版） #

用递归调用替代同步循环，确保每次异步操作完成后再发起下一次轮询：

function sqsReceivemessage(params) {
  sqs.receiveMessage(params, function (err, data) {
    if (err) {
      console.log("Receive Error", err);
      // 出错后延迟5秒重试，避免频繁请求
      setTimeout(() => sqsReceivemessage(params), 5000);
      return;
    }

    if (data?.Messages?.length > 0) {
      console.log("data received")
      let body = JSON.parse(data.Messages[0].Body);
      let action = data.Messages[0].MessageAttributes?.action?.StringValue;
      
      if(action === 'vaxxaEmailAttemptToSend'){
        vaxxaEmailAttemptToSend(body);
      }

      let deleteParams = {
        QueueUrl: queueURL,
        ReceiptHandle: data.Messages[0].ReceiptHandle
      };

      sqs.deleteMessage(deleteParams, function (err, deleteData) {
        if (err) {
          console.log("Delete Error", err);
        } else {
          console.log("Message Deleted", deleteData);
        }
        // 无论删除成功与否，继续轮询下一条消息
        sqsReceivemessage(params);
      });
    } else {
      // 没有消息时，直接发起下一次轮询（依赖SQS长轮询机制）
      sqsReceivemessage(params);
    }
  });
}

// 启动轮询
sqsReceivemessage(recieveMessageParams);

更易读的实现方式（Async/Await版） #

async function sqsReceivemessage(params) {
  try {
    const data = await sqs.receiveMessage(params).promise();
    
    if (data?.Messages?.length > 0) {
      console.log("data received")
      const body = JSON.parse(data.Messages[0].Body);
      const action = data.Messages[0].MessageAttributes?.action?.StringValue;
      
      if(action === 'vaxxaEmailAttemptToSend'){
        await vaxxaEmailAttemptToSend(body); // 若该函数为异步，需加await
      }

      const deleteParams = {
        QueueUrl: queueURL,
        ReceiptHandle: data.Messages[0].ReceiptHandle
      };
      
      await sqs.deleteMessage(deleteParams).promise();
      console.log("Message Deleted");
    }
  } catch (err) {
    console.log("Error occurred", err);
    await new Promise(resolve => setTimeout(resolve, 5000)); // 出错延迟重试
  } finally {
    // 无论成功失败，继续轮询
    sqsReceivemessage(params);
  }
}

// 启动轮询
sqsReceivemessage(recieveMessageParams);

额外优化点 #

• 变量名修正：recieveMessageParams拼写错误，应为receiveMessageParams（少了一个'e'），规范命名避免混淆。
• 空消息处理：增加data.Messages为空的判断，避免SQS长轮询超时无消息时触发报错。
• 错误重试：接收消息失败时添加延迟，防止短时间内大量失败请求。

内容的提问来源于stack exchange，提问作者Aleem Sheikh