readable.reduce(fn[, initial[, options]])

• fn <Function> | <AsyncFunction> 一个 reducer 函数来调用流中的每个块。

  ¥fn <Function> | <AsyncFunction> a reducer function to call over every chunk in the stream.

  • previous <any> 从最后一次调用 fn 获得的值或 initial 值（如果指定）或流的第一个块。

    ¥previous <any> the value obtained from the last call to fn or the initial value if specified or the first chunk of the stream otherwise.

  • data <any> 来自流的数据块。

    ¥data <any> a chunk of data from the stream.

  • options <Object>

    • signal <AbortSignal> 如果流被销毁则中止，允许提前中止 fn 调用。

      ¥signal <AbortSignal> aborted if the stream is destroyed allowing to abort the fn call early.

• initial <any> 用于减少的初始值。

  ¥initial <any> the initial value to use in the reduction.

• options <Object>

  • signal <AbortSignal> 如果信号中止，允许销毁流。

    ¥signal <AbortSignal> allows destroying the stream if the signal is aborted.

• 返回：<Promise> 对最终减少值的 promise。

  ¥Returns: <Promise> a promise for the final value of the reduction.

此方法按顺序在流的每个块上调用 fn，将前一个元素的计算结果传递给它。它返回对减少的最终值的 promise。

¥This method calls fn on each chunk of the stream in order, passing it the result from the calculation on the previous element. It returns a promise for the final value of the reduction.

如果没有提供 initial 值，则流的第一个块将用作初始值。如果流为空，则使用 TypeError 和 ERR_INVALID_ARGS 代码属性拒绝 promise。

¥If no initial value is supplied the first chunk of the stream is used as the initial value. If the stream is empty, the promise is rejected with a TypeError with the ERR_INVALID_ARGS code property.

import { Readable } from 'node:stream';
import { readdir, stat } from 'node:fs/promises';
import { join } from 'node:path';

const directoryPath = './src';
const filesInDir = await readdir(directoryPath);

const folderSize = await Readable.from(filesInDir)
  .reduce(async (totalSize, file) => {
    const { size } = await stat(join(directoryPath, file));
    return totalSize + size;
  }, 0);

console.log(folderSize); 拷贝

reducer 函数逐个元素地迭代流，这意味着没有 concurrency 参数或并行性。要同时执行 reduce，你可以将异步函数提取到 readable.map 方法。

¥The reducer function iterates the stream element-by-element which means that there is no concurrency parameter or parallelism. To perform a reduce concurrently, you can extract the async function to readable.map method.

import { Readable } from 'node:stream';
import { readdir, stat } from 'node:fs/promises';
import { join } from 'node:path';

const directoryPath = './src';
const filesInDir = await readdir(directoryPath);

const folderSize = await Readable.from(filesInDir)
  .map((file) => stat(join(directoryPath, file)), { concurrency: 2 })
  .reduce((totalSize, { size }) => totalSize + size, 0);

console.log(folderSize); 拷贝