[[356203]]

目錄

在 Events 中使用 asyncIterator

• events.on() 示例 1
• events.on() 示例 2
• events.on() 開啟一個 Node.js 服務(wù)器
• 解析 Node.js 源碼對 events.on 異步迭代器的實現(xiàn)

在 Stream 中使用 asyncIterator

• 異步迭代器 與 Readable
• 從 Node.js 源碼看 readable 是如何實現(xiàn)的 asyncIterator
• 異步迭代器與 Writeable

在 MongoDB 中使用 asyncIterator

• MongoDB 中的 cursor
• MongoDB 異步迭代器實現(xiàn)源碼分析
• 使用 for await...of 遍歷可迭代對象 cursor
• 傳送 cursor 到可寫流

在 Events 中使用 asyncIterator

Node.js v12.16.0 中新增了 events.on(emitter, eventName) 方法，返回一個迭代 eventName 事件的異步迭代器。

events.on() 示例 1

如下例所示， for await...of 循環(huán)只會輸出 Hello 當(dāng)觸發(fā) error 事件時會被 try catch 所捕獲。

const { on, EventEmitter } = require('events'); 
 
(async () => { 
  const ee = new EventEmitter(); 
  const ite = on(ee, 'foo'); 
 
  process.nextTick(() => { 
    ee.emit('foo', 'Hello'); 
    ee.emit('error', new Error('unknown mistake.')) 
    ee.emit('foo', 'Node.js'); 
  }); 
 
  try { 
    for await (const event of ite) { 
      console.log(event); // prints ['Hello'] 
    } 
  } catch (err) { 
    console.log(err.message); // unknown mistake. 
  } 
})(); 

上述示例，如果 EventEmitter 對象實例 ee 觸發(fā)了 error 事件，錯誤信息會被拋出并且退出循環(huán)，該實例注冊的所有事件偵聽器也會一并移除。

events.on() 示例 2

for await...of 內(nèi)部塊的執(zhí)行是同步的，每次只能處理一個事件，即使你接下來還有會立即執(zhí)行的事件，也是如此。如果是需要并發(fā)執(zhí)行的則不建議使用，這個原因會在下面解析 events.on() 源碼時給出答案。

如下所示，雖然事件是按順序同時觸發(fā)了兩次，但是在內(nèi)部塊模擬了 2s 的延遲，下一次事件的處理也會得到延遲。

const ite = on(ee, 'foo'); 
 
process.nextTick(() => { 
  ee.emit('foo', 'Hello'); 
  ee.emit('foo', 'Node.js'); 
  // ite.return(); // 調(diào)用后可以結(jié)束 for await...of 的遍歷 
  // ite.throw() // 迭代器對象拋出一個錯誤 
}); 
 
try { 
  for await (const event of ite) { 
    console.log(event); // prints ['Hello'] ['Node.js'] 
    await sleep(2000); 
  } 
} catch (err) { 
  console.log(err.message); 
} 
 
// Unreachable here 
console.log('這里將不會被執(zhí)行'); 

上例中最后一句代碼是不會執(zhí)行的，此時的迭代器會一直處于遍歷中，雖然上面兩個事件 emit 都觸發(fā)了，但是迭代器并沒有終止，什么時候終止呢?也就是當(dāng)內(nèi)部出現(xiàn)一些錯誤或我們手動調(diào)用可迭代對象的 return() 或 throw() 方法時迭代器才會終止。

events.on() 開啟一個 Node.js 服務(wù)器

之前一篇文章《“Hello Node.js” 這一次是你沒見過的寫法》寫過一段使用 events.on() 開啟一個 HTTP 服務(wù)器的代碼，在留言中當(dāng)時有小伙伴對此提出疑惑，基于本章對異步迭代器在 events.on() 中使用的學(xué)習(xí)，可以很好的解釋。

相關(guān)代碼如下所示：

import { createServer as server } from 'http'; 
import { on } from 'events'; 
const ee = on(server().listen(3000), 'request'); 
for await (const [{ url }, res] of ee) 
  if (url === '/hello') 
    res.end('Hello Node.js!'); 
  else 
    res.end('OK!'); 

以上代碼看似新穎，其核心實現(xiàn)就是使用 events.on() 返回 createServer() 對象 request 事件的異步可迭代對象，之后用 for await...of 語句遍歷，客戶端每一次請求，就相當(dāng)于做了一次 ee.emit('request', Req, Res)。

由于內(nèi)部塊的執(zhí)行是同步的，下一次事件處理需要依賴上次事件完成才可以執(zhí)行，對于一個 HTTP 服務(wù)器需要考慮并發(fā)的，請不要使用上面這種方式!

解析 Node.js 源碼對 events.on 異步迭代器的實現(xiàn)

events 模塊直接導(dǎo)出了 on() 方法，這個 on() 方法主要是將異步迭代器與事件的 EventEmitter 類的實例對象做了結(jié)合，實現(xiàn)還是很巧妙的，以下對核心源碼做下解釋，理解之后你完全也可以自己實現(xiàn)一個 events.on()。

• 行 {1} ObjectSetPrototypeOf 是為對象設(shè)置一個新的原型，這個對象包含了 next()、return()、throw() 三個方法。
• 行 {2} 根據(jù)異步可迭代協(xié)議，可迭代對象必須要包含一個 Symbol.asyncIterator 屬性，該屬性是一個無參數(shù)的函數(shù)，返回可迭代對象本身，也就是下面代碼中 SymbolAsyncIterator。
• 行 {3} 新的原型就是 ObjectSetPrototypeOf 的第二個參數(shù) AsyncIteratorPrototype。
• 行 {4} eventTargetAgnosticAddListener 是對事件注冊監(jiān)聽器，里面還是用的事件觸發(fā)器對象的 on() 方法 emitter.on(name, listener) 。
• 行 {5} addErrorHandlerIfEventEmitter 判斷事件名如果不等于 'error' 同時注冊一個 error 事件的監(jiān)聽器，具體實現(xiàn)同行 {4}。
• 行 {6} eventHandler() 函數(shù)就是上面注冊的監(jiān)聽器函數(shù) listener 當(dāng)有事件觸發(fā)時執(zhí)行該監(jiān)聽器函數(shù)，與異步迭代器的結(jié)合就在這里，當(dāng)有新事件觸發(fā)時會從 unconsumedPromises 數(shù)組里取出第一個元素執(zhí)行，如果理解異步迭代器實現(xiàn)標(biāo)準(zhǔn)你會發(fā)現(xiàn) PromiseResolve(createIterResult(args, false)) 就是異步迭代器對象 next() 方法返回值的標(biāo)準(zhǔn)定義。

下面繼續(xù)看 unconsumedPromises 從何而來。

module.exports = EventEmitter; 
module.exports.on = on; 
 
function on(emitter, event) { 
  const unconsumedEvents = []; 
  const unconsumedPromises = []; 
  const iterator = ObjectSetPrototypeOf({ // {1} 
    next() { .... }, 
    return() { ... }, 
    throw(err) { ... }, 
    [SymbolAsyncIterator]() { // {2} 
      return this; 
    } 
  }, AsyncIteratorPrototype); // {3} 
  eventTargetAgnosticAddListener(emitter, event, eventHandler); // {4} 
  if (event !== 'error') { 
    addErrorHandlerIfEventEmitter(emitter, errorHandler); // {5} 
  } 
  return iterator; 
               
  function eventHandler(...args) { // {6} 
    const promise =  .shift(); 
    if (promise) { 
      // 以下等價于 promise.resolve({ value: args, done: false }); 
      PromiseResolve(createIterResult(args, false)); 
    } else { 
      // for await...of 遍歷器內(nèi)部塊的執(zhí)行是同步的，所以每次只能處理 1 個事件，如果同時觸發(fā)多個事件，上次事件未完成剩下的事件會被保存至 unconsumedEvents 中，待上次事件完成后，遍歷器會自動調(diào)用 iterator 對象的 next() 方法，消費所有未處理的事件。 
      unconsumedEvents.push(args); 
    } 
  } 
} 
 
function eventTargetAgnosticAddListener(emitter, name, listener, flags) { 
  ... 
  emitter.on(name, listener); 
} 

以下是 iterator 對象的 next() 方法實現(xiàn)：

• 行 {1} 首先消費未讀消息
• 行 {2} 判斷如果是發(fā)生錯誤則拋出錯誤信息，例如 iterator 對象的 throw() 方法被調(diào)用后就會對 error 做賦值待下次遍歷器調(diào)用 next() 此處代碼就會被執(zhí)行。
• 行 {3} 如果迭代器對象完成，返回的 Promise 對象 done 屬性設(shè)置為 true，遍歷器也就結(jié)束了，變量 finished 是由 iterator 對象的 return() 方法被調(diào)用之后設(shè)置的。
• 行 {4} 這個是上面提到的 unconsumedPromises 數(shù)據(jù)來源處，例如當(dāng)我們執(zhí)行 for await...of 語句遍歷異步迭代器對象時就會自動觸發(fā) iterator 對象的 next() 方法，執(zhí)行到行 {4} 處會創(chuàng)建一個 Promise 對象但是 resolve 并沒有被立即執(zhí)行，而是先存放在 unconsumedPromises 數(shù)組中，所以在上面 #events.on() 示例 2# 提到一個問題，for await...of 遍歷事件的異步迭代器對象時后面的代碼塊并不會被執(zhí)行， 當(dāng)我們觸發(fā)一個事件時才會在監(jiān)聽器函數(shù)里執(zhí)行這個 resolve 函數(shù)，此時才會被釋放，之后 for await...of 遍歷器會自動再次執(zhí)行 next() 方法，然后 new 一個新的 Promise 反復(fù)循環(huán)，直到事件對象拋出 error 事件或執(zhí)行 iterator 對象的 return() 方法。

const iterator = ObjectSetPrototypeOf({ 
  next() { 
    // {1} 首先，我們會消費所有未讀消息 
    const value = unconsumedEvents.shift(); 
    if (value) { 
      return PromiseResolve(createIterResult(value, false)); 
    } 
 
    // {2} 如果發(fā)生一次 error 就會執(zhí)行 Promise.reject 拋出一個錯誤，在這個錯誤發(fā)生后也會停止事件監(jiān)聽。 
    if (error) { 
      const p = PromiseReject(error); 
      // Only the first element errors 
      error = null; 
      return p; 
    } 
 
    // {3} 如果迭代器對象完成，Promise.resolve done 設(shè)置為 true 
    if (finished) { 
      return PromiseResolve(createIterResult(undefined, true)); 
    } 
 
    // {4} 等待直到一個事件發(fā)生 
    return new Promise(function(resolve, reject) { 
      unconsumedPromises.push({ resolve, reject }); 
    }); 
  } 
  ... 
} 

在 Stream 中使用 asyncIterator

Node.js Stream 模塊的可讀流對象在 v10.0.0 版本試驗性的支持了 [Symbol.asyncIterator] 屬性，可以使用 for await...of 語句遍歷可讀流對象，在 v11.14.0 版本以上已 LTS 支持。

異步迭代器 與 Readable

借助 fs 模塊創(chuàng)建一個可讀流對象 readable。

const fs = require('fs'); 
const readable = fs.createReadStream('./hello.txt', { 
  encoding: 'utf-8', 
  highWaterMark: 1 
}); 

以往當(dāng)我們讀取一個文件時，需要監(jiān)聽 data 事件，拼接數(shù)據(jù)，在 end 事件里判斷完成，如下所示：

function readText(readable) { 
  let data = ''; 
  return new Promise((resolve, reject) => { 
    readable.on('data', chunk => { 
      data += chunk; 
    }) 
    readable.on('end', () => { 
      resolve(data); 
    }); 
    readable.on('error', err => { 
      reject(err); 
    }); 
  }) 
} 

現(xiàn)在通過異步迭代器能以一種更簡單的方式實現(xiàn)，如下所示：

async function readText(readable) { 
  let data = ''; 
  for await (const chunk of readable) { 
    data += chunk; 
  } 
  return data; 
} 

現(xiàn)在我們可以調(diào)用 readText 做測試。

(async () => { 
  try { 
    const res = await readText(readable); 
    console.log(res); // Hello Node.js 
  } catch (err) { 
    console.log(err.message); 
  } 
})(); 

使用 for await...of 語句遍歷 readable，如果循環(huán)中因為 break 或 throw 一個錯誤而終止，則這個 Stream 也將被銷毀。

上述示例中 chunk 每次接收的值是根據(jù)創(chuàng)建可讀流時 highWaterMark 這個屬性決定的，為了能清晰的看到效果，在創(chuàng)建 readable 對象時我們指定了 highWaterMark 屬性為 1 每次只會讀取一個字符。

從 Node.js 源碼看 readable 是如何實現(xiàn)的 asyncIterator

與同步的迭代器遍歷語句 for...of 類似，用于 asyncIterator 異步迭代器遍歷的 for await...of 語句在循環(huán)內(nèi)部會默認(rèn)調(diào)用可迭代對象 readable 的 Symbol.asyncIterator() 方法得到一個異步迭代器對象，之后調(diào)用迭代器對象的 next() 方法獲取結(jié)果。

本文以 Node.js 源碼 v14.x 為例來看看源碼是如何實現(xiàn)的。當(dāng)我們調(diào)用 fs.createReadStream() 創(chuàng)建一個可讀流對象時，對應(yīng)的該方法內(nèi)部會調(diào)用 ReadStream 構(gòu)造函數(shù)

// https://github.com/nodejs/node/blob/v14.x/lib/fs.js#L2001 
function createReadStream(path, options) { 
  lazyLoadStreams(); 
  return new ReadStream(path, options); 
} 

其實在 ReadStream 這個構(gòu)造函數(shù)里沒有我們要找的，重點是它通過原型的方式繼承了 Stream 模塊的 Readable 構(gòu)造函數(shù)。

function ReadStream(path, options) { 
  ... 
  Readable.call(this, options); 
} 

那么現(xiàn)在我們重點來看看 Readable 這個構(gòu)造函數(shù)的實現(xiàn)。

Readable 原型上定義了 SymbolAsyncIterator 屬性，該方法返回了一個由生成器函數(shù)創(chuàng)建的迭代器對象。

// for await...of 循環(huán)會調(diào)用 
Readable.prototype[SymbolAsyncIterator] = function() { 
  let stream = this; 
  ... 
  const iter = createAsyncIterator(stream); 
  iter.stream = stream; 
  return iter; 
}; 
 
// 聲明一個創(chuàng)建異步迭代器對象的生成器函數(shù) 
async function* createAsyncIterator(stream) { 
  let callback = nop; 
 
  function next(resolve) { 
    if (this === stream) { 
      callback(); 
      callback = nop; 
    } else { 
      callback = resolve; 
    } 
  } 
 
  const state = stream._readableState; 
 
  let error = state.errored; 
  let errorEmitted = state.errorEmitted; 
  let endEmitted = state.endEmitted; 
  let closeEmitted = state.closeEmitted; 
  
  // error、end、close 事件控制了什么時候結(jié)束迭代器遍歷。 
  stream 
    .on('readable', next) 
    .on('error', function(err) { 
      error = err; 
      errorEmitted = true; 
      next.call(this); 
    }) 
    .on('end', function() { 
      endEmitted = true; 
      next.call(this); 
    }) 
    .on('close', function() { 
      closeEmitted = true; 
      next.call(this); 
    }); 
 
  try { 
    while (true) { 
      // stream.read() 從內(nèi)部緩沖拉取并返回數(shù)據(jù)。如果沒有可讀的數(shù)據(jù)，則返回 null 
      // readable 的 destroy() 方法被調(diào)用后 readable.destroyed 為 true，readable 即為下面的 stream 對象 
      const chunk = stream.destroyed ? null : stream.read(); 
      if (chunk !== null) { 
        yield chunk; // 這里是關(guān)鍵，根據(jù)迭代器協(xié)議定義，迭代器對象要返回一個 next() 方法，使用 yield 返回了每一次的值 
      } else if (errorEmitted) { 
        throw error; 
      } else if (endEmitted) { 
        break; 
      } else if (closeEmitted) { 
        break; 
      } else { 
        await new Promise(next); 
      } 
    } 
  } catch (err) { 
    destroyImpl.destroyer(stream, err); 
    throw err; 
  } finally { 
    if (state.autoDestroy || !endEmitted) { 
      // TODO(ronag): ERR_PREMATURE_CLOSE? 
      destroyImpl.destroyer(stream, null); 
    } 
  } 
} 

通過上面源碼可以看到可讀流的異步迭代器實現(xiàn)使用了生成器函數(shù) Generator yield，那么對于 readable 對象遍歷除了 for await...of 遍歷之外，其實也是可以直接使用調(diào)用生成器函數(shù)的 next() 方法也是可以的。

const ret = readable[Symbol.asyncIterator]() 
console.log(await ret.next()); // { value: 'H', done: false } 
console.log(await ret.next()); // { value: 'e', done: false } 

異步迭代器與 Writeable

通過上面講解，我們知道了如何遍歷異步迭代器從 readable 對象獲取數(shù)據(jù)，但是你有沒有想過如何將一個異步迭代器對象傳送給可寫流?正是此處要講的。

從迭代器中創(chuàng)建可讀流

Node.js 流對象提供了一個實用方法 stream.Readable.from()，對于符合 Symbol.asyncIterator 或 Symbol.iterator 協(xié)議的可迭代對象(Iterable)會先創(chuàng)建一個可讀流對象 readable 之后從迭代器中構(gòu)建 Node.js 可讀流。

以下是 從理解到實現(xiàn)輕松掌握 ES6 中的迭代器 一文中曾講解過的例子，r1 就是我們創(chuàng)建的可迭代對象。使用 stream.Readable.from() 方法則可以將可迭代對象構(gòu)造為一個可讀流對象 readable。

function Range(start, end) { 
  this.id = start; 
  this.end = end; 
} 
Range.prototype[Symbol.asyncIterator] = async function* () { 
  while (this.id <= this.end) { 
    yield this.id++; 
  } 
} 
const r1 = new Range(0, 3); 
const readable = stream.Readable.from(r1); 
readable.on('data', chunk => { 
  console.log(chunk); // 0 1 2 3 
}); 

傳送異步迭代器到可寫流

使用 pipeline 可以將一系列的流和生成器函數(shù)通過管道一起傳送，并在管道完成時獲取通知。

使用 util.promisify 將 pipeline 轉(zhuǎn)化為 promise 形式。

const util = require('util'); 
const pipeline = util.promisify(stream.pipeline); // 轉(zhuǎn)為 promise 形式 
 
(async () => { 
  try { 
    const readable = stream.Readable.from(r1); 
    const writeable = fs.createWriteStream('range.txt'); 
    await pipeline( 
      readable, 
      async function* (source) { 
        for await (const chunk of source) { 
          yield chunk.toString(); 
        } 
      }, 
      writeable 
    ); 
    console.log('Pipeline 成功'); 
  } catch (err) { 
    console.log(err.message); 
  } 
})() 

在寫入數(shù)據(jù)時，傳入的 chunk 需是 String、Buffer、Uint8Array 類型，否則 writeable 對象在寫入數(shù)據(jù)時會報錯。由于我們自定義的可迭代對象 r1 里最終返回的值類型為 Number 在這里需要做次轉(zhuǎn)換，管道中間的生成器函數(shù)就是將每次接收到的值轉(zhuǎn)為字符串。

在 MongoDB 中使用 asyncIterator

除了上面我們講解的 Node.js 官方提供的幾個模塊之外，在 MongoDB 中也是支持異步迭代的，不過介紹這點的點資料很少，MongoDB 是通過一個游標(biāo)的概念來實現(xiàn)的。

MongoDB 中的 cursor

本處以 Node.js 驅(qū)動 mongodb 模塊來介紹，當(dāng)我們調(diào)用 db.collection.find() 這個方法返回的是一個 cursor(游標(biāo))，如果想要訪問文檔那么我們需要迭代這個游標(biāo)對象來完成，但是通常我們會直接使用 toArray() 這個方法來完成。

下面讓我們通過一段示例來看，現(xiàn)在我們有一個數(shù)據(jù)庫 example，一個集合 books，表里面有兩條記錄，如下所示：

image.png

查詢 books 集合的所有數(shù)據(jù)，以下代碼中定義的 myCursor 變量就是游標(biāo)對象，它不會自動進(jìn)行迭代，可以使用游標(biāo)對象的 hasNext() 方法檢測是否還有下一個，如果有則可以使用 next() 方法訪問數(shù)據(jù)。

通過以下日志記錄可以看到在第三次調(diào)用 hasNext() 時返回了 false，如果此時在調(diào)用 next() 就會報錯，游標(biāo)已關(guān)閉，也就是已經(jīng)沒有數(shù)據(jù)可遍歷了。

const MongoClient = require('mongodb').MongoClient; 
const dbConnectionUrl = 'mongodb://127.0.0.1:27017/example'; 
 
(async () => { 
  const client = await MongoClient.connect(dbConnectionUrl, { useUnifiedTopology: true }); 
  const bookColl = client.db('example').collection('books'); 
  const myCursor = await bookColl.find(); 
  
  console.log(await myCursor.hasNext()); // true 
  console.log((await myCursor.next()).name); // 深入淺出Node.js 
  console.log(await myCursor.hasNext()); // true 
  console.log((await myCursor.next()).name); // Node.js實戰(zhàn) 
  console.log(await myCursor.hasNext()); // false 
  console.log((await myCursor.next()).name); // MongoError: Cursor is closed 
})() 

直接調(diào)用 next() 也可檢測，如果還有值則返回該條記錄，否則 next() 方法返回 null。

console.log((await myCursor.next()).name); 
console.log((await myCursor.next()).name); 
console.log((await myCursor.next())); 

MongoDB 異步迭代器實現(xiàn)源碼分析

MongoDB 中游標(biāo)是以 hasNext() 返回 false 或 next() 返回為 null 來判斷是否達(dá)到游標(biāo)尾部，與之不同的是在我們的 JavaScript 可迭代協(xié)議定義中是要有一個 Symbol.asyncIterator 屬性的迭代器對象，且迭代器對象是 { done, value } 的形式。

幸運(yùn)的是 MongoDB Node.js 驅(qū)動已經(jīng)幫助我們實現(xiàn)了這一功能，通過一段源碼來看在 MongoDB 中的實現(xiàn)。

• find 方法

find 方法返回的是一個可迭代游標(biāo)對象。

// https://github.com/mongodb/node-mongodb-native/blob/3.6/lib/collection.js#L470 
 
Collection.prototype.find = deprecateOptions( 
  { 
    name: 'collection.find', 
    deprecatedOptions: DEPRECATED_FIND_OPTIONS, 
    optionsIndex: 1 
  }, 
  function(query, options, callback) { 
    const cursor = this.s.topology.cursor( 
      new FindOperation(this, this.s.namespace, findCommand, newOptions), 
      newOptions 
    ); 
 
    return cursor; 
  } 
); 

• CoreCursor

核心實現(xiàn)就在這里，這是一個游標(biāo)的核心類，MongoDB Node.js 驅(qū)動程序中所有游標(biāo)都是基于此，如果當(dāng)前支持異步迭代器，則在 CoreCursor 的原型上設(shè)置 Symbol.asyncIterator 屬性，返回基于 Promise 實現(xiàn)的異步迭代器對象，這符合 JavaScript 中關(guān)于異步可迭代對象的標(biāo)準(zhǔn)定義。

// https://github.com/mongodb/node-mongodb-native/blob/3.6/lib/core/cursor.js#L610 
 
if (SUPPORTS.ASYNC_ITERATOR) { 
  CoreCursor.prototype[Symbol.asyncIterator] = require('../async/async_iterator').asyncIterator; 
} 

// https://github.com/mongodb/node-mongodb-native/blob/3.6/lib/async/async_iterator.js#L16 
 
// async function* asyncIterator() { 
//   while (true) { 
//     const value = await this.next(); 
//     if (!value) { 
//       await this.close(); 
//       return; 
//     } 
 
//     yield value; 
//   } 
// } 
 
// TODO: change this to the async generator function above 
function asyncIterator() { 
  const cursor = this; 
 
  return { 
    next: function() { 
      return Promise.resolve() 
        .then(() => cursor.next()) 
        .then(value => { 
          if (!value) { 
            return cursor.close().then(() => ({ value, done: true })); 
          } 
          return { value, done: false }; 
        }); 
    } 
  }; 
} 

目前是默認(rèn)使用的 Promise 的形式實現(xiàn)的，上面代碼中有段 TODO， Node.js 驅(qū)動關(guān)于異步迭代實現(xiàn)這塊可能后期會改為基于生成器函數(shù)的實現(xiàn)，這對我們使用是沒變化的.

使用 for await...of 遍歷可迭代對象 cursor

還是基于我們上面的示例，如果換成 for await...of 語句遍歷就簡單的多了。

const myCursor = await bookColl.find(); 
for await (val of myCursor) { 
  console.log(val.name); 
} 

在 MongoDB 中的聚合管道中使用也是如此，就不再做過多分析了，如下所示：

const myCursor = await bookColl.aggregate(); 
for await (val of myCursor) { 
  console.log(val.name); 
} 

對于遍歷龐大的數(shù)據(jù)集時，使用游標(biāo)它會批量加載 MongoDB 中的數(shù)據(jù)，我們也不必?fù)?dān)心一次將所有的數(shù)據(jù)存在于服務(wù)器的內(nèi)存中，造成內(nèi)存壓力過大。

傳送 cursor 到可寫流

MongoDB 游標(biāo)對象本身也是一個可迭代對象(Iterable)，結(jié)合流模塊的 Readable.from() 則可轉(zhuǎn)化為可讀流對象，是可以通過流的方式進(jìn)行寫入文件。

但是要注意 MongoDB 中的游標(biāo)每次返回的是單條文檔記錄，是一個 Object 類型的，如果直接寫入，可寫流是會報參數(shù)類型錯誤的，因為可寫流默認(rèn)是一個非對象模式(僅接受 String、Buffer、Unit8Array)，所以才會看到在 pipeline 傳輸?shù)闹虚g又使用了生成器函數(shù)，將每次接收的數(shù)據(jù)塊處理為可寫流 Buffer 類型。

const myCursor = await bookColl.find(); 
const readable = stream.Readable.from(myCursor); 
await pipeline( 
  readable, 
  async function* (source) { 
    for await (const chunk of source) { 
      yield Buffer.from(JSON.stringify(chunk)); 
    } 
  }, 
  fs.createWriteStream('books.txt') 
); 

Reference

• https://nodejs.org/dist/latest-v14.x/docs/api/stream.html#stream_readable_symbol_asynciterator
• https://nodejs.org/dist/latest-v14.x/docs/api/events.html#events_events_on_emitter_eventname
• https://docs.mongodb.com/manual/tutorial/iterate-a-cursor/index.html

本文轉(zhuǎn)載自微信公眾號「Nodejs技術(shù)?！梗梢酝ㄟ^以下二維碼關(guān)注。轉(zhuǎn)載本文請聯(lián)系Nodejs技術(shù)棧公眾號。