在 Node.js 中用 pipe 處理數(shù)組的實現(xiàn)

mikyou 發(fā)布于2019-08-21 15:26 / 1943人閱讀

摘要：總結(jié)雖然我沒有像老師一樣進(jìn)行性能測試，但是猜測也知道的方式在數(shù)量比較小的時候肯定要弱于正常方式，的好處在于數(shù)據(jù)量比較大的時候，可以使用比較小的內(nèi)存，盡快的處理數(shù)組中前置的數(shù)據(jù)。

TLDR;
這篇文章的風(fēng)格是在致敬 Jim 老師；致敬，致敬，懂嗎，不是抄襲，程序員的事怎么能叫抄襲。
當(dāng)然我對 Node.js 的 stream 也是現(xiàn)學(xué)現(xiàn)賣，有使用不當(dāng)?shù)牡胤剑凑堉赋觥?br>原文鏈接 歡迎 star。

寫這篇文章的初衷是年前看 SICP 的時候，第二章介紹構(gòu)造數(shù)據(jù)抽象的時候有提到 Lisp 對序列的處理采用類似『信號流』的方式。所以很自然的就想到了 Node.js 中的 pipe 方式，于是就一直想用 pipe 的方式嘗試一下。

同 Jim 老師的這篇文章中描述的一樣，我也是懶癌發(fā)作，從年尾拖到今年年初，然后在年初又看到了 Jim 老師的博客，深受啟發(fā)，終于下定決心要開始碼了...... 然后，嗯，又拖到昨天。促使我下定決心要寫的主要原因是昨天部門的年會！反正年會跟我這種死肥宅也沒多大關(guān)系，在大家 happy 的時候構(gòu)思了下代碼實現(xiàn)，回家用了一晚上的時候補(bǔ)上了代碼。

Jim 老師在他的文章里面也說了，JS 的那些數(shù)組操作 (map/ reduce/filter) 啥的，每次調(diào)用的時候都會進(jìn)行一次完整的遍歷。試想一下如果有一個第一個數(shù)是1，長度是 1億的遞增為 1 的數(shù)組，需要把所有的數(shù)組都乘 3，再排除其中的奇數(shù)，如果用 (map/filter) 的方法，只要也需要循環(huán) 一億五千萬次；那么如果有其他辦法能只循環(huán)一億次，是不是節(jié)省了大量的內(nèi)存資源和循環(huán)消耗的時間。

廢話不多說，直接上代碼吧。

pipe

在編寫代碼時，我們應(yīng)該有一些方法將程序像連接水管一樣連接起來 -- 當(dāng)我們需要獲取一些數(shù)據(jù)時，可以去通過"擰"其他的部分來達(dá)到目的。這也應(yīng)該是IO應(yīng)有的方式。 -- Doug McIlroy. October 11, 1964

關(guān)于 node 的 stream 可以看看這篇文章。

下面是代碼部分，整個代碼我是在邊學(xué) pipe 邊用一晚上的時間倉促寫就的，懶癌發(fā)作，也不想再重構(gòu)了，各位相公講究看吧，求別噴代碼。

入口

const stream = require("stream")

const last = Symbol()

// 在 selfArray 中接收一個真正的數(shù)組
// 返回一個可讀流
// 如果再做的精細(xì)點，可以做成可讀可寫流，這樣就能通過控制流的大小，來控制內(nèi)存的大小，別幾億條數(shù)據(jù)直接撐爆內(nèi)存了
// 不過對后面 reduce 的處理就比較麻煩
function selfArray(a) {
  const rs = new stream.Readable({
    objectMode: true
  })

  a.forEach((v, index) => {
    rs.push(v)
  })
  rs.push(last)
  rs.push(null)
  return rs
}

上面的 selfArray 在流的最后面 push 了一個 Symbol 對象來標(biāo)志整個流的輸入結(jié)束，留待為之后 reduce 的使用。

Map/Filter/Reduce 的實現(xiàn)

function forEach(callback) {
  const ws = new stream.Writable({
    objectMode: true
  })
  let index = 0

  ws._write = function (chunk, enc, next) {
    if (chunk !== last) {
      callback(chunk, index++)
      next()
    }
  }

  return ws
}

function filter(callback) {
  const trans = new stream.Transform({
    readableObjectMode: true,
    writableObjectMode: true
  })

  let index = 0

  trans._transform = function (chunk, enc, next) {
    if (chunk === last) {
      next(null, last)
    } else {
      let condition = callback(chunk, index++)
      if (condition) {
        this.push(chunk)
      }
      next()
    }
  }
  return trans
}

function map(callback) {
  const trans = new stream.Transform({
    readableObjectMode: true,
    writableObjectMode: true
  })
  let index = 0
  trans._transform = function (chunk, enc, next) {
    if (chunk === last) {
      next(null, last)
    } else {
      next(null, callback(chunk, index++))
    }
  }
  return trans
}

function reduce(callback, initial) {
  const trans = new stream.Transform({
    readableObjectMode: true,
    writableObjectMode: true
  })

  let index = 0,
    current = initial,
    prev = initial


  trans._transform = function (chunk, enc, next) {

    if (chunk === last) {
      if (index > 1) {
        prev = callback(prev, current, index - 1)
      }
      this.push(prev)
      this.push(last)
      return next(null, last)
    }

    if (initial === void 0 && index === 0) {
      prev = chunk
    }

    if (index > 0) {
      prev = callback(prev, current, index - 1)
    }

    current = chunk
    index++
    next()
  }

  return trans
}

上面的代碼在 reduce 的實現(xiàn)稍微麻煩了一些，reduce 對沒有初始值，原始數(shù)組為空的條件下有各種不同的處理情況，翻看了下 MDN 的解釋又自己實現(xiàn)了下。

使用

selfArray([9, 2, 6, 3, 5, 6, 7, 1, 4, 4])
  .pipe(map(v => v * 3))
  .pipe(filter(v => v % 2))
  .pipe(reduce((p, c) => p + c, 0))
  .pipe(forEach(v => {
    console.log("pipe 計算最后的結(jié)果是:", v)
  }))

為了好看我故意把各種括號都刪掉了。嗯，看起來還挺完美，我們來測試下

selfArray([9, 2, 6, 3, 5, 6, 7, 1, 4, 4])
  .pipe(map(v => {
    console.log("map:", v)
    return v * 3
  }))
  .pipe(filter(v => {
    console.log("filter:", v)
    return v % 2
  }))
  .pipe(reduce((p, c) => {
    console.log("reduce:", p, c)
    return p + c
  }, 0))
  .pipe(forEach(v => {
    console.log("pipe 計算最后的結(jié)果是:", v)
  }))
  
  
加上 log 之后可以看到結(jié)算結(jié)果是:
  
map: 9
filter: 27
map: 2
filter: 6
map: 6
filter: 18
map: 3
filter: 9
reduce: 0 27
map: 5
filter: 15
reduce: 27 9
map: 6
filter: 18
map: 7
filter: 21
reduce: 36 15
map: 1
filter: 3
reduce: 51 21
map: 4
filter: 12
map: 4
filter: 12
reduce: 72 3
pipe 計算最后的結(jié)果是: 75

從上面的 log 可以看到，第一個數(shù) 9 先執(zhí)行了 map，然后在 3 之后就直接進(jìn)入了 filter，此時第 2 個數(shù) 2 也開始被 map 處理，然后被 filter 處理，但是由于 3 之后是偶數(shù)不會被 reduce 接收， reduce 會一直等到第二個奇數(shù)，也就是 3 進(jìn)入之后才會被處理... 嗯，直到最終的計算結(jié)果是 75，被 forEach 消耗。

總結(jié)

雖然我沒有像 Jim 老師一樣進(jìn)行性能測試，但是猜測也知道 pipe 的方式在數(shù)量比較小的時候肯定要弱于正常方式，pipe 的好處在于數(shù)據(jù)量比較大的時候，可以使用比較小的內(nèi)存，盡快的處理數(shù)組中前置的數(shù)據(jù)。