理解 RXJS 可观察输出的问题

我对 rxjs 相当陌生，正在尝试学习。

我正在读这篇媒体文章 https://medium.com/javascript-in-plain-english/reactive-programming-in-javascript-8373201a6618并尝试理解这个 RXJS 输出：

import {Observable} from 'rxjs-es';
let output = Observable.interval(500)
             .map(i => [1,2,3,4,5,6][i]);

let result = output.map(num1 => num1)
    .filter(num1 => num1 > 4)
    .reduce((num1, num2) => num1 + num2);

result.subscribe(number => console.log(number));

输出：27

我不明白输出是如何变成 27 的，也不明白这个reduce 是如何工作的（以及reducer 的这两个参数是什么）。

有人可以告诉我发生了什么事吗？ （我尝试在codesandbox上运行它，但是当我运行此代码时它会抛出错误）

理解Filter and Reduce运营商

正如其他人在评论中提到的，您引用的文章使用的是具有不同语法的旧版本 rxjs。对于此示例，我将使用从版本 6 开始的较新语法。

在 rxjs 中，有各种运算符可用于转换通过流发出的值。通常这些是这样导入的：

import { filter, reduce } from 'rxjs/operators';

还有许多生成器函数可用于创建值流。interval是这些函数之一，它将创建一个流，该流每隔n毫秒。导入如：

import { interval } from 'rxjs';

让我们创建一个简单的流：

number$ = interval(1000); // emit number every 1 second
// output: 0, 1, 2, 3, 4, 5...

我们可以将运算符应用于此流来转换排放：

的用法filter很简单。它只是发出通过给定真值测试的值（与Array.filter()方法）。

numbersLessThan4$ = numbers$.pipe(
    filter(number => number < 4)
);
// output: 0, 1, 2, 3

The reduce运算符有点复杂，其行为就像Array.reduce()方法。函数应用于每个发射值，并且能够存储在评估下一次发射时可以引用的值。

reduce有两个参数。第一个是接收当前发射的函数（cur) 和之前的累计结果 (acc）并返回一个新的累计值。第二个是初始值acc.

example:

sumOfNumbers$ = numbers$.pipe(
    reduce((acc, cur) => acc + cur, 0)
);

那么，让我们看看什么reduce何时做numbers$发出前 3 个数字：

• 0
  • cur接收当前排放值0
  • acc从提供的默认值开始0
  • 表达方式acc + cur回报0
• 1
  • cur接收当前排放值1
  • acc接收之前返回的值0
  • 表达方式acc + cur回报1
• 2
  • cur接收当前排放值2
  • acc接收之前返回的值1
  • 表达方式acc + cur回报3

所以这很酷。我们可以将相当多的逻辑放入一行简单的代码中。一件重要的事情reduce是在源可观察对象完成之前它不会发出。现在，numbers$永远不会完成（interval()无限期地发出连续整数）。

我们可以使用take()运算符在发出一定数量的值后完成流。

Example:

numbers$ = interval(1000).pipe(take(5)); // completes after 5 emissions

sumOfNumbers$ = numbers$.pipe(
    // receives 5 values (0, 1, 2, 3, 4) and performs the logic described above.
    reduce((acc, cur) => acc + cur, 0) 
);
// output: 10

可以使用多个运营商来转换排放。只需在里面提供多个pipe():

sumOfNumbersLessThan4$ = numbers$.pipe(
    filter(number => number < 4),
    reduce((acc, cur) => acc + cur, 0) 
);
// output: 6