了解一个问题要从：是什么，为什么，怎么做。逐步深入去考虑

基础知识

rxjs是什么？

官网说：

Think of RxJS as Lodash for events.

把 rxjs 当做处理事件的工具库，有的前端老师傅说，不能把 rxjs 当做库来学习，

库和框架的区别是什么呢？

学习库，你了解了这个库大概是干嘛的就好，用到什么去查 api 就完事

而学习框架，你要了解这个框架建设的一些列概念，来驱动着去了解这个框架的思想再去使用

那 rxjs 到底应该怎么学？它是提供了一些方法？还是建设了很多概念？（比如事件流）

那么流是指什么呢？举个例子，代码中每1s输出一个数字， 用户每一次对元素的点击，就像是在时间这个维度上，产生了一个数据集。这个数据集不像数组那样，它不是一开始都存在的，而是随着时间的流逝，一个一个数据被输出出来。这种异步行为产生的数据，就可以被称之为一个流，在Rxjs中，称之为observable（抛开英文，本质其实就是一个数据的集合，只是这些数据不一定是一开始就设定好的，而是随着时间而不断产生的）。而Rxjs，就是为了处理这种流而产生的工具，比如流与流的合并，流的截断，延迟，消抖等等操作。

为什么出现 rxjs 这个东西

每个新技术的出现 都是为了解决原有技术的一些痛点

那么 rxjs 呢？它和 promise 的异同优缺点 各有哪些呢？

rxjs 怎么用？

正文开始

基本概念

先了解 rx 给我们提供的基本概念

The essential concepts in RxJS which solve async event management are:

• Observable:  represents the idea of an invokable collection of future values or events.
• Observer:  is a collection of callbacks that knows how to listen to values delivered by the Observable.
• Subscription:  represents the execution of an Observable, is primarily useful for cancelling the execution.
• Operators:  are pure functions that enable a functional programming style of dealing with collections with operations like map, filter, concat, reduce, etc.
• Subject:  is equivalent to an EventEmitter, and the only way of multicasting a value or event to multiple Observers.
• Schedulers:  are centralized dispatchers to control concurrency, allowing us to coordinate when computation happens on e.g. setTimeout or requestAnimationFrame or others.

---

先上总结

Subscription = Observable.subscribe(observer)

• Observable 好似一个Function，有很多return，定义了在哪个时点return什么数据，是数据流，可产生数据，通过 subscribe 来启动流，不要看这个单词的引文翻译，直接把它理解成，一个随时间不断产生数据的一个集合，称之为流更容易理解。

const x$ = () => {
  （无条件的）： 
    return 'I'm the 1st go';
  在'500 ms'以后: 
    return 'I'm the 2nd go';
  ...
  在http请求出错时：
    return 'something went wrong';
  // 然后就没有然后了
}

• observer 就是一个Object，是怎么处理上述产生的数据，理解成回调吧？

const observer = {
  next: (value) => console.log(value),
  error: (error) => console.log(error),
  complete: () => console.log('completed')
}

• Subscription 储存已启动过的流，提供了unsubscribe，来停止这个流。

简单理解了这三个名词observable, observer, subscription后，从数据的角度来思考：observable定义了要生成一个什么样的数据，其subscribe方法，接收一个observer（定义了接收到数据如何处理），并开始产生数据，该方法的返回值，subscription, 存储了这个已经开启的流（暂时没想到啥好的中文名），同事具有unscbscribe方法，可以将这个流停止。

再上例子

import { Observable } from "rxjs";

// 流的创建
const stream$ = new Observable(subscriber => {
  setTimeout(() => {
    subscriber.next([1, 2, 3]);
  }, 500);
});

// 如何消费流产生的数据，observer
const observer = {
  complete: () => console.log("done"),
  next: v => console.log(v),
  error: () => console.log("error")
};

// 启动流
const subscription = stream$.subscribe(observer);

setTimeout(() => {
  // 停止流
  subscription.unsubscribe();
}, 1000);

new Observalbe(fn)定义了一个流

通过stream$.subscribe(obj)启动了这个流

500ms后，执行了`subsciber.next([1,2,3])

通过传入的obj.next方法输出了该值

操作符

emmm，这怎么理解呢。。。

subject

有前端大哥总结说：

subject 消费者也是生产者，有next和subscribe方法，有一个observers列表

一时间我没有理解

import { Subject } from "rxjs";

// 创建subject
const subject = new Subject();

// 订阅一个observer
subject.subscribe(v => console.log("stream 1", v));
// 再订阅一个observer
subject.subscribe(v => console.log("stream 2", v));
// 延时1s再订阅一个observer
setTimeout(() => {
  subject.subscribe(v => console.log("stream 3", v));
}, 1000);
// 产生数据1
subject.next(1);
// 产生数据2
subject.next(2);
// 延时3s产生数据3
setTimeout(() => {
  subject.next(3);
}, 3000);
// output
// stream 1 1 //立即输出
// stream 2 1 //立即输出
// stream 1 2 //立即输出
// stream 2 2 //立即输出
// stream 1 3 //3s后输出
// stream 2 3 //3s后输出
// stream 3 3 //3s后输出

参考资料

zhuanlan.zhihu.com/p/274469124 zhuanlan.zhihu.com/p/24451002