如何使用 Bluebird 承诺 AWS SQS

Question

我有以下有效的代码。我开始使用 Bluebird 对它进行承诺，但是，我不确定如何承诺对消息数组的处理。

var s3 = new AWS.S3();
var sqs = new AWS.SQS();
 // This notification call is triggered by the latest message but there may
 // be earlier unprocessed messages. So, we request the maximum number of
 // messages (10) from the queue and process and then remove from the queue
 // all of them.
sqs.receiveMessage({
  QueueUrl: settings.sqsQueueUrl[prdOrDev],
  /* required */
  WaitTimeSeconds: 20, // to enable long polling, which polls all servers for any unprocessed SQS messages
  VisibilityTimeout: 120, // without this longpolling didn't work.
  MaxNumberOfMessages: 10
}, function(err, data) {
  if (err) {
    console.error('SQS receiveMessage failed: ', err, err.stack);
    return res.status(400).json({
      success: false
    });
  } else {
    var messages = data.Messages;
    messages.forEach(function(message) {
      var body = JSON.parse(message.Body);
      var sesMsg = JSON.parse(body.Message);
      s3.getObject({
        Bucket: sesMsg.receipt.action.bucketName,
        Key: sesMsg.receipt.action.objectKey
      }, function(err, data2) {
        if (err) {
          console.error('S3 getObject failed: ', err, err.stack);
        } else {
          sqs.deleteMessage({
            QueueUrl: settings.sqsQueueUrl[prdOrDev],
            /* required */
            ReceiptHandle: message.ReceiptHandle
          }, function(err, data) {
            if (err) {
              console.error('SQS deleteMessage failed: ', err, err.stack);
            }
          });
        }
      });
    });
  }
});

这是我对上面代码的承诺：

var Promise = require('bluebird');
var s3 = new AWS.S3();
var sqs = new AWS.SQS();
Promise.promisifyAll(Object.getPrototypeOf(s3));
Promise.promisifyAll(Object.getPrototypeOf(sqs));

sqs.receiveMessageAsync({
  QueueUrl: settings.sqsQueueUrl[prdOrDev],
  /* required */
  WaitTimeSeconds: 20, // to enable long polling, which polls all servers for any unprocessed SQS messages
  VisibilityTimeout: 120, // without this longpolling didn't work.
  MaxNumberOfMessages: 10
}).then(function(data) {
  var messages = data.Messages;
  messages.forEach(function(message) {
    var body = JSON.parse(message.Body);
    var sesMsg = JSON.parse(body.Message);
    s3.getObjectAsync({
      Bucket: sesMsg.receipt.action.bucketName,
      Key: sesMsg.receipt.action.objectKey
    }).then(function(data2) {
      return sqs.deleteMessageAsync({
        QueueUrl: settings.sqsQueueUrl[prdOrDev],
        /* required */
        ReceiptHandle: message.ReceiptHandle
      }).catch(function(err) {
        console.log('SQS deleteMessage failed: ', err, err.stack);
      });
    }).catch(function(err) {
      console.log('S3 getObject failed: ', err, err.stack);
    });
  });
}).catch(function(err) {
  notifyAdmin('SQS receiveMessage failed: ', err, err.stack);
});

我猜这不是使用 Promises 的最佳方式。我特别好奇是否有更好的方法来处理 forEach 循环，类似于 Bluebird 主页中的以下示例：

mongoClient.connectAsync('mongodb://localhost:27017/mydb')
    .then(function(db) {
        return db.collection('content').findAsync({})
    })
    .then(function(cursor) {
        return cursor.toArrayAsync();
    })
    .then(function(content) {
        res.status(200).json(content);
    })
    .catch(function(err) {
        throw err;
    });

那么，我如何最好地使用 Bluebird 来保证顶部的代码 sn-p？

Answer 1

在 forEach 循环中，您的 then() 函数打破了链条，您正在创建 Promise，但您不会“等待”它们。通常的方法是将所有的 Promise 存储在一个数组中并使用 Promise.all()。所以用你的代码：

sqs.receiveMessageAsync({
  QueueUrl: settings.sqsQueueUrl[prdOrDev],
  /* required */
  WaitTimeSeconds: 20, // to enable long polling, which polls all servers for any unprocessed SQS messages
  VisibilityTimeout: 120, // without this longpolling didn't work.
  MaxNumberOfMessages: 10
}).then(function(data) {
  var messages = data.Messages;
  var promises = [];
  messages.forEach(function(message) {
    var body = JSON.parse(message.Body);
    var sesMsg = JSON.parse(body.Message);

    var promise = s3.getObjectAsync({
      Bucket: sesMsg.receipt.action.bucketName,
      Key: sesMsg.receipt.action.objectKey
    }).then(function(data2) {
      return sqs.deleteMessageAsync({
        QueueUrl: settings.sqsQueueUrl[prdOrDev],
        /* required */
        ReceiptHandle: message.ReceiptHandle
      }).catch(function(err) {
        console.log('SQS deleteMessage failed: ', err, err.stack);
      });
    }).catch(function(err) {
      console.log('S3 getObject failed: ', err, err.stack);
    });

    promises.push(promise);
  });

  return Promise.all(promises);
}).then(function(result) {
  console.log('all done');
}).catch(function(err) {
  notifyAdmin('SQS receiveMessage failed: ', err, err.stack);
});

您还可以将 promisify 代码简化为：

var s3 = Promise.promisifyAll(new AWS.S3());
var sqs = Promise.promisifyAll(new AWS.SQS());