使用外部 API 调用和 findOneAndUpdate 循环结果

Question

我正在尝试编写一个程序，使用 mongoose 从 mongo 数据库获取文档，并使用 API 处理它们，然后使用处理结果编辑数据库中的每个文档。我的问题是我遇到问题是因为我不完全理解nodejs和异步。这是我的代码：

Model.find(function (err, tweets) {
    if (err) return err;
    for (var i = 0; i < tweets.length; i++) {
        console.log(tweets[i].tweet);
        api.petition(tweets[i].tweet)
            .then(function(res) {
                TweetModel.findOneAndUpdate({_id: tweets[i]._id}, {result: res}, function (err, tweetFound) {
                    if (err) throw err;
                    console.log(tweetFound);
                });
            })
            .catch(function(err) {
                console.log(err);
            })
    }
})

问题是在 findOneAndUpdate 中，tweets 未定义，因此无法找到该 id。有什么解决办法吗？谢谢

Answer 1

您真正缺少的核心是 Mongoose API 方法也使用"Promises"，但您似乎只是使用回调从文档或旧示例中复制。解决方案是转换为仅使用 Promise。

与 Promise 一起工作

Model.find({},{ _id: 1, tweet: 1}).then(tweets => 
  Promise.all(
    tweets.map(({ _id, tweet }) => 
      api.petition(tweet).then(result =>   
       TweetModel.findOneAndUpdate({ _id }, { result }, { new: true })
         .then( updated => { console.log(updated); return updated })
      )
    )
  )
)
.then( updatedDocs => {
  // do something with array of updated documents
})
.catch(e => console.error(e))

除了回调的一般转换之外，主要的变化是使用解析结果处理的Promise.all()输出而不是循环。这实际上是您尝试中最大的问题之一，因为实际上无法控制异步函数何时解析。另一个问题是“混合回调”，但这就是我们通常仅使用 Promise 来解决的问题。Array.map().find()forfor

在 中，我们从 API 调用Array.map()返回，链接到实际更新文档的 。我们还用于实际返回修改后的文档。PromisefindOneAndUpdate()new: true

Promise.all()允许“Promise 数组”解析并返回结果数组。您将这些视为updatedDocs. 这里的另一个优点是内部方法将以“并行”而不是串行方式触发。这通常意味着更快的分辨率，尽管它需要更多的资源。

另请注意，我们使用 的“投影”{ _id: 1, tweet: 1 }仅从结果中返回这两个字段Model.find()，因为它们是其余调用中唯一使用的字段。当您不使用其他值时，这可以节省为每个结果返回整个文档的时间。

Promise您可以简单地从 中返回findOneAndUpdate()，但我只是添加了 ，console.log()以便您可以看到此时正在触发输出。

正常的生产使用应该不需要它：

Model.find({},{ _id: 1, tweet: 1}).then(tweets => 
  Promise.all(
    tweets.map(({ _id, tweet }) => 
      api.petition(tweet).then(result =>   
       TweetModel.findOneAndUpdate({ _id }, { result }, { new: true })
      )
    )
  )
)
.then( updatedDocs => {
  // do something with array of updated documents
})
.catch(e => console.error(e))

另一个“调整”可能是使用 的“bluebird”实现Promise.map()，它结合了通用Array.map()的Promise(s) 实现和控制运行并行调用的“并发性”的能力：

const Promise = require("bluebird");

Model.find({},{ _id: 1, tweet: 1}).then(tweets => 
  Promise.map(tweets, ({ _id, tweet }) => 
    api.petition(tweet).then(result =>   
      TweetModel.findOneAndUpdate({ _id }, { result }, { new: true })
    ),
    { concurrency: 5 }
  )
)
.then( updatedDocs => {
  // do something with array of updated documents
})
.catch(e => console.error(e))

“并行”的替代方案是按顺序执行。如果太多结果导致太多 API 调用和写回数据库的调用，则可以考虑这样做：

Model.find({},{ _id: 1, tweet: 1}).then(tweets => {
  let updatedDocs = [];
  return tweets.reduce((o,{ _id, tweet }) => 
    o.then(() => api.petition(tweet))
      .then(result => TweetModel.findByIdAndUpdate(_id, { result }, { new: true })
      .then(updated => updatedDocs.push(updated))
    ,Promise.resolve()
  ).then(() => updatedDocs);
})
.then( updatedDocs => {
  // do something with array of updated documents
})
.catch(e => console.error(e))

在那里，我们可以使用Array.reduce()将承诺“链接”在一起，使它们能够按顺序解决。请注意，结果数组保留在范围内，并与.then()附加到连接链末尾的最终结果进行交换，因为您需要这样的技术来“收集”在该“链”中不同点解析的 Promise 的结果。

异步/等待

在现代环境中，从 NodeJS V8.x（实际上是当前的 LTS 版本）开始，已经有一段时间了，您实际上支持async/await. 这使您可以更自然地编写流程

try {
  let tweets = await Model.find({},{ _id: 1, tweet: 1});

  let updatedDocs = await Promise.all(
    tweets.map(({ _id, tweet }) => 
      api.petition(tweet).then(result =>   
        TweetModel.findByIdAndUpdate(_id, { result }, { new: true })
      )
    )
  );

  // Do something with results
} catch(e) {
  console.error(e);
}

如果资源是个问题，甚至可能按顺序处理：

try {
  let cursor = Model.collection.find().project({ _id: 1, tweet: 1 });

  while ( await cursor.hasNext() ) {
    let { _id, tweet } = await cursor.next();
    let result = await api.petition(tweet);
    let updated = await TweetModel.findByIdAndUpdate(_id, { result },{ new: true });
    // do something with updated document
  }

} catch(e) {
  console.error(e)
}

还请注意，findByIdAndUpdate()也可以用作匹配_id已经暗示的内容，因此您不需要整个查询文档作为第一个参数。

批量写入

最后一点，如果您实际上根本不需要更新的文档作为响应，那么这bulkWrite()是更好的选择，并且允许写入通常在单个请求中在服务器上处理：

Model.find({},{ _id: 1, tweet: 1}).then(tweets => 
  Promise.all(
    tweets.map(({ _id, tweet }) => api.petition(tweet).then(result => ({ _id, result }))
  )
).then( results =>
  Tweetmodel.bulkWrite(
    results.map(({ _id, result }) => 
      ({ updateOne: { filter: { _id }, update: { $set: { result } } } })
    )
  )
)
.catch(e => console.error(e))

或者通过async/await语法：

try {
  let tweets = await Model.find({},{ _id: 1, tweet: 1});

  let writeResult = await Tweetmodel.bulkWrite(
    (await Promise.all(
      tweets.map(({ _id, tweet }) => api.petition(tweet).then(result => ({ _id, result }))
    )).map(({ _id, result }) =>
      ({ updateOne: { filter: { _id }, update: { $set: { result } } } })
    )
  );
} catch(e) {
  console.error(e);
}

几乎所有上面显示的组合都可以变成这样，因为该bulkWrite()方法采用指令“数组”，因此您可以从上面每个方法的已处理 API 调用中构造该数组。