Rxjs Observable

创建Observable：

Rx.Observable.create 是 Observable 构造函数的别名，它接收一个参数：subscribe 函数。
例子：每秒推送一个“Hi”

Rx.Observable.create(function subscribe(observer){var i=setInterval((observer)=>{observer.next("Hi")
},1000);
});

在上面的示例中，subscribe 函数是用来描述 Observable 最重要的一块。

订阅Observable：

示例中的 Observable 对象 observable 可以订阅，像这样：

observable.subscribe(x => console.log(x));

observable.subscribe 和 Observable.create(function subscribe(observer) {…}) 中的 subscribe 有着同样的名字，这并不是一个巧合。在库的实现中，这两个subscribe是不同的，但在实际使用中，你可以认为在概念上它们是等同的。

这个例子表明订阅行为在Observable 的多个观察者之间是不共享的。当使用一个观察者调用 observable.subscribe 时，Observable.create(function subscribe(observer) {…}) 中的 subscribe 函数就会为这个观察者运行一次。对 observable.subscribe 的每次调用都会触发针对给定观察者的独立设置。(每个已订阅的观察者都拥有 Observable 的独立执行)

以上简单的理解为：
订阅Observable的时候，需要传入一个observer。observer定义了自身的next、complete、error方法。observer的next方法会获得Observable的next方法推送过来的值，并按照自己的定义处理他们。当一个observer订阅Observable的时候，不会影响另外一个observer。(每个已订阅的观察者都拥有 Observable 的独立执行)
也就是说：只有Observable才可以被subscribe。只有observer才可以subscribe Observable。
订阅 Observable 像是调用函数, 并提供接收数据的回调函数。

取消订阅：

Observable.subscribe()返回一个Subscription对象。调用其unsubscribe()方法取消订阅。实现类似如下：

var observable = Rx.Observable.create(function subscribe(observer) {// 追踪 interval 资源var intervalID = setInterval(() => {observer.next('hi');}, 1000);// 提供取消和清理 interval 资源的方法return function unsubscribe() {clearInterval(intervalID);};
});//这里返回一个函数声明
var subscription=observable.subscribe((x)=>console.log(x));
//执行取消订阅
subscription.unsubscribe();

Observer (观察者)

简单的讲，观察者是用来消费（处理）Observable 推送出来的值。
它是一个包含了3个方法的对象。每个方法分别对应一个Observable发送通知的类型。
如果你只提供了一个方法，但是没有将它附加到观察者对象上，像这样：

observable.subscribe(x => console.log('Observer got a next value: ' + x));

在 observable.subscribe 内部，它会创建一个观察者对象并使用第一个回调函数参数作为 next 的处理方法。
如果你没有提供某个回调函数，Observable 的执行也会正常运行，只是某些通知类型会被忽略，因为观察者中没有相对应的回调函数。
Subscription (订阅)

subscription通常只有一个unsubscribe()方法，这个函数用来释放资源或去取消 Observable 执行。
Subscription 还可以合在一起，这样一个 Subscription 调用 unsubscribe() 方法，可能会有多个 Subscription 取消订阅。你可以通过把一个 Subscription 添加到另一个上面来做这件事：

subscription1.add（subscription2）；//添加子订阅
subscription1.unsubscribe();// 2个都会取消订阅。
subscription1.remove(subscription2); //取消添加子订阅

Subject (主题)

什么是 Subject？ - RxJS Subject 是一种特殊类型的 Observable，它允许将值多播给多个观察者，所以 Subject 是多播的，而普通的 Observables 是单播的(每个已订阅的观察者都拥有 Observable 的独立执行)。

Subject 像是 Observable，但是可以多播给多个观察者。Subject 还像是 EventEmitters，维护着多个监听器的注册表。每个 Subject 都是 Observable 。

在 Subject 的内部，subscribe 不会调用发送值的新执行。它只是将给定的观察者注册到观察者列表中，类似于其他库或语言中的 addListener 的工作方式。

每个 Subject 都是观察者。

Subjects 是将任意 Observable 执行共享给多个观察者的唯一方式。
多播 Observable 在底层是通过使用 Subject 使得多个观察者可以看见同一个 Observable 执行。
多播的 Observables

多播 Observable

在底层是通过使用 Subject 使得多个观察者可以看见同一个 Observable 执行。（共享一个Observable ）

在底层，这就是 multicast 操作符的工作原理：观察者订阅一个基础的 Subject，然后 Subject 订阅源 Observable 。

var subject = new Rx.Subject();subject.subscribe({next: (v) => console.log('observerA: ' + v)
});
subject.subscribe({next: (v) => console.log('observerB: ' + v)
});var observable = Rx.Observable.from([1, 2, 3]);observable.subscribe(subject); // 你可以提供一个 Subject 进行订阅

执行结果：
observerA: 1
observerB: 1
observerA: 2
observerB: 2
observerA: 3
observerB: 3

multicast 操作符返回一个 Observable，它看起来和普通的 Observable 没什么区别，但当订阅时就像是 Subject 。multicast 返回的是 ConnectableObservable，它只是一个有 connect() 方法的 Observable 。

var source = Rx.Observable.from([1, 2, 3]);
var subject = new Rx.Subject();
var multicasted = source.multicast(subject);// 在底层使用了 `subject.subscribe({...})`:
multicasted.subscribe({next: (v) => console.log('observerA: ' + v)
});
multicasted.subscribe({next: (v) => console.log('observerB: ' + v)
});// 在底层使用了 `source.subscribe(subject)`:
multicasted.connect();

connect() 方法十分重要，它决定了何时启动共享的 Observable 执行。因为 connect() 方法在底层执行了 source.subscribe(subject)，所以它返回的是 Subscription，你可以取消订阅以取消共享的 Observable 执行。

Subject的三种变体类型：

BehaviorSubject ：主要用来表示随着时间的流逝，值会改变的情况。比如生日是Subject，而生日就是BehaviorSubject。
ReplaySubject ：当有新的订阅者订阅时，可以指定回放的值得个数，或者以 window time (以毫秒为单位) 规定范围内发生的值。
AsyncSubject ：当被观察者的值推送完成时，才将最后一个值发送给观察者。类似于last ()操作符。

Rxjs Observable相关推荐

使用 RxJs Observable 来避免 Angular 应用中的 Promise 使用
我们通过一个具体的例子来阐述. 考虑您正在构建一个搜索输入掩码,该掩码应在您键入时立即显示结果. 如果您曾经构建过这样的东西,那么您可能会意识到该任务带来的挑战. 不要在每次击键时都点击搜索端点将搜 ...
Rxjs Observable.pipe 传入多个 operators 的执行逻辑分析
测试代码: fromEvent(this.test, 'click').pipe(map( event => event.timeStamp), mapTo(1)).subscribe((eve ...
通过一个实际例子理解Angular rxjs Observable的异步行为
源代码: ngOnInit(): void {const a = new Observable((observer) => {observer.next(1);observer.next(2); ...
如何取消Angular rxjs Observable的订阅
源代码: ngOnInit(): void {const ob = new Observable(function subscribe(observer) {// 追踪 interval 资源cons ...
RxJS 系列之二 - Observable 详解
查看新版教程,请访问前端修仙之路 RxJS 系列目录 RxJS 系列之一 - Functional Programming 简介 RxJS 系列之二 - Observable 详解 (本文) RxJS ...
redux rxjs_可观察的RxJS和Redux入门指南
redux rxjs Redux-Observable is an RxJS-based middleware for Redux that allows developers to work wit ...
angular 使用rxjs 监听同级兄弟组件数据变化
angular 的官网给出了父子组件之间数据交互的方法,如ViewChild.EventEmitter 但是如果要在同级组件之间进行数据同步,似乎并没有给出太多的信息. 有时候我们想,在一个组件中修改 ...
rxjs angular_Angular RxJS深度
rxjs angular In this tutorial, we'll learn to use the RxJS 6 library with Angular 6 or Angular 7. We ...
rxjs处理http请求超时
博客原文地址使用场景用户进行一个操作请求后台而长时间未响应,我们希望给用户一个信息展示(请求超时,网络不好-). RxJS实现关于RxJS请看这里我这个功能的实现主要使用 delay,race ...