rxjs简介

摘要：类似于函数声明类似于函数调用类似于调用时传入回调函数对象来执行执行后的返回值称为可以停止后续的执行还可以合在一起一起终止多个的执行是多个值的惰性推送集合。

Observable类似于函数声明

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

subscribe类似于函数调用(类似于function.call),调用时传入回调函数observer对象来执行

observable.subscribe({next:x => console.log(x)})

执行后的返回值称为Subscription,可以停止后续Observable的执行,Subscription还可以合在一起,一起终止多个Observable的执行

var subscription = observable.subscribe({next:x => console.log(x)})
subscription.unsubscribe()

Observable是多个值的惰性推送集合。

Observable和普通函数的区别:

普通的function会同步的返回一个值,属于一种拉取值的操作,你可以确定这个值是存在的,而Observable可以同步或异步的返回一个或多个值,属于推送值的操作,类似于Promise,你并不确定这个值是否存在,也不知道何时会接收到数据,但是可以通过回调函数的方式,明确当值可用的时候做什么操作.

为什么要用Observable呢,因为它可以完成一些普通function完成不了的工作,比如:

纯净性使你的代码无需引入外部状态,更不容易出错。

var count = 0;
var button = document.querySelector("button");
button.addEventListener("click", () => console.log(`Clicked ${++count} times`));

// 使用rxjs
var button = document.querySelector("button");
Rx.Observable.fromEvent(button, "click")
  .scan(count => count + 1, 0)
  .subscribe(count => console.log(`Clicked ${count} times`));

RxJS 提供了一整套操作符来帮助你控制事件如何流经 observables ,代码更简洁,更可读。

var count = 0;
var rate = 1000;
var lastClick = Date.now() - rate;
var button = document.querySelector("button");
button.addEventListener("click", () => {
  if (Date.now() - lastClick >= rate) {
    console.log(`Clicked ${++count} times`);
    lastClick = Date.now();
  }
});

var button = document.querySelector("button");
Rx.Observable.fromEvent(button, "click")
  .throttleTime(1000)
  .scan(count => count + 1, 0)
  .subscribe(count => console.log(`Clicked ${count} times`));

Subject类似于事件监听,会在内部维护多个observer

var subject = new Rx.Subject();

subject.subscribe({
  next: (v) => console.log("observerA: " + v)
});
subject.subscribe({
  next: (v) => console.log("observerB: " + v)
});

调用Subject.next时发送事件通知,将依次执行Subject内部维护的全部observer

var subject = new Rx.Subject();

subject.subscribe({
  next: (v) => console.log("observerA: " + v)
});
subject.subscribe({
  next: (v) => console.log("observerB: " + v)
});

subject.next(1)
subject.next(2)

在某个Observable声明之后,进行subscribe调用时也可以传入Subject,这将依次执行Subject内部维护的全部observer

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); 

BehaviorSubject是Subject的一个变体,它有一个“当前值”的概念。当前值是发送给消费者的最后一个值。在new BehaviorSubject时可以初始化一个当前值。当有新的观察者订阅时，会立即从 BehaviorSubject 那接收到“当前值”。

var subject = new Rx.BehaviorSubject(0); // 0是初始的当前值

subject.subscribe({
  next: (v) => console.log("observerA: " + v)
});

subject.next(1);
subject.next(2);

subject.subscribe({
  next: (v) => console.log("observerB: " + v)
});

subject.next(3);

ReplaySubject类似于BehaviorSubject,它有一个“缓冲值”的概念,可以想象成多个"当前值"。在new ReplaySubject时可以初始化一个"缓冲值"的数量,当有新的观察者订阅时，会立即从 ReplaySubject 那接收到每一个“缓冲值”。

除了缓冲数量，你还可以指定 window time (以毫秒为单位)来确定多久之前的值可以记录。

var subject = new Rx.ReplaySubject(3); // 为新的订阅者缓冲3个值

subject.subscribe({
  next: (v) => console.log("observerA: " + v)
});

subject.next(1);
subject.next(2);
subject.next(3);
subject.next(4);

subject.subscribe({
  next: (v) => console.log("observerB: " + v)
});

subject.next(5);

AsyncSubject 是另一个 Subject 变体，只有当 Observable 执行完成时(执行 complete())，它才会将执行的最后一个值发送给观察者。

var subject = new Rx.AsyncSubject();

subject.subscribe({
  next: (v) => console.log("observerA: " + v)
});

subject.next(1);
subject.next(2);
subject.next(3);
subject.next(4);

subject.subscribe({
  next: (v) => console.log("observerB: " + v)
});

subject.next(5);
subject.complete();

observable的执行类似函数调用,subject的执行类似事件监听.

observer是一些待执行的函数,Observable可以执行observer,Subject也可以执行observer,只不过是两种不同的运行模式,究竟用哪种视情况而定

操作符就是以特定的observer来订阅原observable,然后返回一个以特定操作结果作为next值的新的Observable

function multiplyByTen(input) {
    // 原observable被封装在新的observable内部,只有外部的observable被订阅时才会执行内部的订阅
    const output = Observable.create(function subscribe(observer) {
        input.subscribe(val => {
            observer.next(val * 10)
        })
    })
    // 返回新的observable
    return output;
}

var source = Observable.create(function subscribe(observer) {
    observer.next(1)
    observer.next(2)
});

multiplyByTen(source).subscribe(val => console.log(val))