Циклирование результатов с помощью внешнего вызова API и findOneAndUpdate

Question

Я пытаюсь написать программу, которая получает документы из базы данных mongo с помощью mongoose и обрабатывает их с помощью 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 твиты не определены, поэтому он не может найти этот идентификатор.Любое решение?Спасибо

Answer 1

Основная вещь, которую вы действительно упускаете, - это то, что методы Mongoose API также используют «Обещания» , но вы, похоже, просто копируете из документации или старых примеров с использованием обратных вызовов.Решением этой проблемы является преобразование в использование только Обещаний.

Работа с Обещаниями

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() для разрешения выхода из Array.map(), обрабатываемого по результатам .find() вместо цикла for.Это на самом деле одна из самых больших проблем в вашей попытке, так как for не может фактически контролировать, когда разрешаются асинхронные функции.Другая проблема - «смешанные обратные вызовы», но это то, к чему мы обычно обращаемся, используя только обещания.

В пределах Array.map() мы возвращаем Promise из вызова API, прикованный к findOneAndUpdate(), который фактически обновляет документ.Мы также используем new: true для фактического возврата измененного документа.

Promise.all() позволяет «массиву обещаний» разрешать и возвращать массив результатов.Это вы видите как 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(), добавленным в конец объединенной цепочки, поскольку вам необходим такой метод для "сбора" результатов из Обещаний, разрешающихся в разных точках этой "цепочки".

---

Async / Await

В современных средах, начиная с 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

В качестве заключительного примечания, есливам вообще не нужны обновленные документы в ответ, тогда 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 из каждого метода выше.