了解RXJS可观察到的输出问题

如何解决了解RXJS可观察到的输出问题

我对rxjs相当陌生，想学习。

我正在阅读this medium article，并试图理解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和Reduce运算符

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

在rxjs中，可以使用各种运算符来转换通过流发出的值。通常，这些导入方式如下：

import { filter,reduce } from 'rxjs';

还有许多生成器功能可用于创建值流。 interval是这些函数之一，它将创建一个流，该流每n毫秒发出一个连续的整数。像这样导入：

import { interval } from 'rxjs/operators';

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

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,3

reduce运算符稍微复杂一点，其行为类似于Array.reduce()方法。将一个函数应用于每个发射值，并且可以存储一个值，该值可以在评估下一个发射时参考。

reduce采用两个参数。第一个函数是接收当前发射（cur）和先前的累积结果（acc）并返回新的累积值的函数。第二个是acc的初始值。

示例：

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

所以，让我们看一下reduce发出前三个数字时的情况：

• 0
  • numbers$接收当前排放值cur
  • 0以提供的默认acc
  开头
  • 表达式0返回acc + cur
• 1
  • 0接收当前排放值cur
  • 1收到先前返回的值acc
  • 表达式0返回acc + cur
• 2
  • 1接收当前排放值cur
  • 2收到先前返回的值acc
  • 表达式1返回acc + cur

这很酷。我们可以在一条简单的代码行中获得很多逻辑。 3的一件事很重要，那就是它在源可观察者完成之前不会发射。当前，reduce从未完成（numbers$无限期地发出连续的整数）。

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

示例：

take()

可以使用多个运算符来转换排放。只需在numbers$ = interval(1000).pipe(take(5)); // completes after 5 emissions sumOfNumbers$ = numbers$.pipe( // receives 5 values (0,4) and performs the logic described above. reduce((acc,0) ); // output: 10 内提供多个即可：

pipe()