响应式编程（rxjs）学习分享

1.学习网址

文档：https://rxjs-cn.github.io/learn-rxjs-operators/operators/combination/concatall.html
图示理解：https://rxmarbles.com/

2.rxjs

rxjs和promise的区别：

promise： then获取值，resolve传值，只能触发一次，所以只能传一个值；
rxjs： subscribe获取值，next传值，可以触发很多次，所以可以传多个值

var p = new Promise((resolve, reject) => {resolve('ainini');resolve('ainini22');reject('ainini33');console.log('lll');
});
p.then(res => {console.log(res);
}).catch((err) => {console.log(err);
});
// 输出:  lll     ainini

// 创建Observable，每隔一秒会向观察者发送字符串 'hi'
var observable = new Observable(function subscribe(observer) {observer.next('hello wmz');var id = setInterval(() => {observer.next('hi')}, 1000);
});observable.subscribe(x => console.log(x));

输出：

hello wmz
hi
hi
hi
hi
// 之后每隔一秒输出一个hi

2.1 Observable (被观察对象)

2.1.1 创建Observable

var observable = new Observable(function subscribe(observer) {observer.next('hello wmz');var id = setInterval(() => {observer.next('hi')}, 1000);
});

2.1.2 执行 Observable（传递next通知）

Observable Execution 可以提供三种类型的值：

“下一个”通知：发送一个值，如数字、字符串、对象等。
“错误”通知：发送 JavaScript 错误或异常。
“完成”通知：不发送值。

const observable = new Observable(function(subscriber) {try {subscriber.next(1);subscriber.next(2);subscriber.next(3);subscriber.complete(); // complete之后对值不会传递了，即之后的代码都不会执行了；subscriber.next(4);} catch (err) {subscriber.error(err); // delivers an error if it caught one}
});

输出：

1
2
3
complete

2.1.3 订阅 Observables

observable示例中的 Observable可以订阅，如下所示：

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

2.1.4 取消执行

const observable = new Observable(function(subscriber) {try {setInterval(() => subscriber.next('hi'), 1000);subscriber.next(4);} catch (err) {subscriber.error(err); // delivers an error if it caught one}// 取消订阅会执行return里面的函数。return () => { console.log('kkk'); };
});
const subscription = observable.subscribe({next: (x) => { console.log(x); },complete: () => { console.log('complete'); },error: (err) => { console.log('err-err', err); }
});setTimeout(() => { subscription.unsubscribe(); }, 4000);

输出：

4
hi
hi
hi
kkk

2.2 Observer（观察者）

观察者是由 Observable 发送的值的消费者。观察者只是一组回调函数的集合，每个回调函数对应一种 Observable 发送的通知类型：next、error 和 complete 。

下面的示例是一个典型的观察者对象：

// 可观察对象和观察者之间的关系
observable.subscribe(observer);

const observer = {next: x => console.log('Observer got a next value: ' + x),error: err => console.error('Observer got an error: ' + err),complete: () => console.log('Observer got a complete notification'),
};

要使用观察者，需要把它提供给 Observable 的 subscribe 方法:

observable.subscribe(observer);

观察者只是有三个回调函数的对象，每个回调函数对应一种 Observable 发送的通知类型。

RxJS 中的观察者也可能是部分的。如果你没有提供某个回调函数，Observable 的执行也会正常运行，只是某些通知类型会被忽略，因为观察者中没有相对应的回调函数下面的示例是没有 complete 回调函数的观察者：

const observer = {next: x => console.log('Observer got a next value: ' + x),error: err => console.error('Observer got an error: ' + err),
};

当订阅 Observable 时，你可能只提供了一个回调函数作为参数，而并没有将其附加到观察者对象上，例如这样：

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

在 observable.subscribe 内部，它会创建一个观察者对象并使用第一个回调函数参数作为 next 的处理方法。三种类型的回调函数都可以直接作为参数来提供：

observable.subscribe(x => console.log('Observer got a next value: ' + x),err => console.error('Observer got an error: ' + err),() => console.log('Observer got a complete notification'),
);

2.3 Subscription（订阅）

const subscription = observable.subscribe({next: (x) => { console.log(x); },complete: () => { console.log('complete'); },error: (err) => { console.log('err-err', err); }
});

Subscription 基本上只有一个 unsubscribe() 函数，这个函数用来释放资源或去取消 Observable 执行。

订阅可以合并，然后一起取消

subscription.add(childSubscription);
// 这样就同时取消了两个订阅
subscription.unsubscribe();

撤销一个已添加的子 Subscription

subscription.remove(childSubscription);

2.4 Subject（主体）

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

Subject即是观察者Observer，也是被观察对象Observable，同时实现了这两个接口。

2.4.1 subject作为Observable

var subject = new Subject();let count = 1;
setInterval(() => {subject.next(count++);
}, 1000);

2.4.2 subject作为观察者

可以实现Observable的单播变成多播

subject.subscribe({next: (v) => console.log(`observerA: ${v}`)
});
setTimeout(() => {subject.subscribe({next: (v) => console.log(`observerB: ${v}`)});
}, 3000);const observable = new Observable((o) => {let count = 1;setInterval(() => {o.next(count++);}, 1000);
})// subject作为观察者
observable.subscribe(subject);

输出：

observerA: 1
observerA: 2
observerA: 3
observerB: 3
observerA: 4
observerB: 4
observerA: 5
observerB: 5
observerA: 6
observerB: 6

2.5 BehaviorSubject

Subject 的其中一个变体就是 BehaviorSubject，它有一个当前值的概念。它保存了发送给消费者的最新值。并且当有新的观察者订阅时，会立即从 BehaviorSubject 那接收到当前值。

2.6 ReplaySubject

ReplaySubject 类似于 BehaviorSubject，它可以发送旧值给新的订阅者，但它还可以记录 Observable 执行的一部分

ReplaySubject 记录 Observable 执行中的多个值并将其回放给新的订阅者

// 在订阅后可以拿到订阅之前的三个值
const subject = new ReplaySubject(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);

输出：

observerA: 1
observerA: 2
observerA: 3
observerA: 4
observerB: 2
observerB: 3
observerB: 4
observerA: 5
observerB: 5

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

// 最大缓冲数量100，500毫秒之前的值可以拿到
const subject = new ReplaySubject(100, 500);

2.7 AsyncSubject

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

// 只有当 Observable 执行完成时(执行 complete())，它才会将执行的最后一个值发送给观察者。
const subject = new AsyncSubject(3);subject.subscribe({next: v => console.log(`observerA: ${v}`),
});let count = 1;
setInterval(() => {subject.next(count++);if (count === 5) subject.complete();
}, 1000);

输出：

observerA: 4

2.8 observable的热（多播）和冷（单播）

Observable 热：多播。所有的观察者，无论进来的早还是晚，看到的是同样内容的同样进度，订阅的时候得到的都是最新时刻发送的值。
Observable 冷：单播。新的订阅者每次从头开始。

例如：普通的Observable的订阅就是冷的

const observable = new Observable(function(subscriber) {try {let count = 1;setInterval(() => subscriber.next(count++), 1000);} catch (err) {subscriber.error(err); }
});
const observer =  {next: (x) => { console.log(x); },complete: () => { console.log('complete'); },error: (err) => { console.log('err-err', err); }
}
observable.subscribe(x => console.log(x));
// 四秒后开始订阅,不是实时订阅，而是从头开始订阅，即从1开始
setTimeout(() => {observable.subscribe(x => console.log('hi ' + x));
}, 4000);

输出：

1
2
3
4
hi 1
5
hi 2
6
hi 3

例如：subject的订阅是热的

var subject = new Subject();
let count = 1;setInterval(() => {subject.next(count++);
}, 1000);subject.next('hello world');// Subject只有在订阅之后，才能收到数据源发出的值
subject.subscribe(x => console.log(x));// 四秒之后开始订阅，但订阅到的是实时消息，即从4开始
setTimeout(() => {subject.subscribe(x => console.log('hi ' + x));
}, 4000);

输出：

1
2
3
4
hi 4
5
hi 5
6
hi 6
7
hi 7

3.rxjs Operator(操作符)

操作符是 Observable 类型上的方法，比如 .map(…)、.filter(…)、.merge(…)，等等。当操作符被调用时，它们不会改变已经存在的 Observable 实例。相反，它们返回一个新的 Observable ，它的 subscription 逻辑基于第一个 Observable 。

操作符是函数，它基于当前的 Observable 创建一个新的 Observable。这是一个无副作用的操作（就是纯函数）：前面的 Observable 保持不变。

observableInstance.pipe(operator(scheduler))

3.1 管道pipe

obs.pipe(op1(), op2(), op3(), op4());  // op4()(op3()(op2()(op1()(obs))))

其实用到的就是函数式编程发的函数组合的pipe方式，从左到右依次执行操作符。

3.2 Creation Operators（创建操作符）

3.2.1 empty（立即完成的 observable ）

// 输出: 'Complete!'
const subscribe = empty().subscribe({next: () => console.log('Next'),complete: () => console.log('Complete!')
});

输出：Complete！

3.2.2 from(将数组、promise 或迭代器转换成 observable )

// from可以将数组，promise，或者迭代器转换成Observable
let ob_arr = from([1, '9', 10]);
ob_arr.subscribe(x => {console.log(x)});// promise不能直接订阅，需要转换成Observable才能订阅。
let pro = Promise.resolve('hello-pro');
let ob_promise = from(pro);
ob_promise.subscribe(x => {console.log(x)});// 使用 js 的集合
const map = new Map();
map.set(1, 'Hi');
map.set(2, 'Bye');
const ob_map = from(map);
// 输出: [1, 'Hi'], [2, 'Bye']
ob_map.subscribe(val => console.log(val));// 字符串转换成的Observable
const ob_str = from('wumengzhao');
ob_str.subscribe(x => {console.log(x)});

输出：

1
9
10
[ 1, 'Hi' ]
[ 2, 'Bye' ]
w
u
m
e
n
g
hello-pro

3.2.3 fromEvent（将事件转换成 observable 序列）

// 创建发出点击事件的 observable
const source = fromEvent(document, 'click');
// 映射成给定的事件时间戳
const example = source.pipe(map(event => `Event time: ${event.timeStamp}`));
// 输出 (示例中的数字以运行时为准): 'Event time: 7276.390000000001'
const subscribe = example.subscribe(val => console.log(val));

3.2.4 fromPromise(创建由 promise 转换而来的 observable，并发出 promise 的结果。)

promise通过reject发出消息

const pro = new Promise((resolve, reject) => {reject('wmz');
});from(pro).subscribe({next: x => console.log('next ' + x),error: err => console.log('error ' + err),complete: x => console.log('complete ' + x),
});

输出：

error wmz

promise通过resolve发出消息

const pro = new Promise((resolve, reject) => {resolve('wmz');
});from(pro).subscribe({next: x => console.log('next ' + x),error: err => console.log('error ' + err),complete: x => console.log('complete ' + x),
});

输出：

next wmz
complete undefined

3.2.5 interval

// 每1秒发出数字序列中的值
const source = interval(1000);
// 数字: 0,1,2,3,4,5....
const subscribe = source.subscribe(val => console.log(val));

3.2.6 of(按顺序发出任意数量的值)。

// 出任意类型值
const source = of({ name: 'Brian' }, [1, 2, 3], function hello() {return 'Hello';
});
// 输出: {name: 'Brian}, [1,2,3], function hello() { return 'Hello' }
const subscribe = source.subscribe(val => console.log(val));

3.2.7 range(依次发出给定区间内的数字)

//  输出: 1,2,3,4,5,6,7,8,9,10
var rangeOb = range(1, 10);
rangeOb.subscribe({next: x => console.log(x),
})

3.2.8 throw（在订阅上发出错误）

var te = throwError('wmz, error');
te.subscribe({catchError: err => console.log(err)
})

3.2.9 timer(initialDelay: number | Date, period: number, scheduler: Scheduler): Observable

// 延迟一秒发送0，之后每隔三秒发送一个数字
var te = timer(1000, 3000);
te.subscribe({next: x => console.log(x)
})；

3.3 转换操作符

3.3.1 map

对源 observable 的每个值应用投射函数生成新的值。

3.3.2 mapTo

mapTo是把传进来的值改写成为一个固定值;

var te = of(1, 2, 3);
te.pipe(mapTo('sss')).subscribe({next: x => console.log(x)
})

输出：

sss
sss
sss

3.3.3 mergeMap（不会取消之前的订阅,并且产生的多个observable发出的值按时间顺序订阅）(相当于map+mergeAll)

对源observable 的每个值映射产生新的Observable，然后对这些新的observable产生的值按顺序发送（打平）

// 立即发出值， 然后每2秒发出值
const source = timer(0, 2000);
// 当 source 发出值时并不会取消前一个内部observal的订阅
const example = source.pipe(mergeMap(() => interval(500)));
// 输出: 0,1,2,3,0,4,1,5,2,6,3,7,0,4,8...
const subscribe = example.subscribe(val => console.log(val));

3.3.4 mergeMapTo

对源Observable发出的值映射成固定的observable，并按顺序打平输出

// 立即发出值， 然后每z秒发出值
const source = timer(0, 2000);
// 当 source 发出值时并不会取消前一个内部observal的订阅
const example = source.pipe(mergeMapTo(interval(500)));
// 输出: 0,1,2,3,0,4,1,5,2,6,3,7,0,4,8...
const subscribe = example.subscribe(val => console.log(val));

3.3.5 switchMap（重点是会取消之前的订阅）（相当于map+switchAll）

// 立即发出值， 然后每5秒发出值
const source = timer(0, 5000);// 当 source 发出值时切换到新的内部 observable，发出新的内部 observable 所发出的值,取消以前的订阅const example = source.pipe(switchMap(() => interval(500)));// 输出: 0,1,2,3,4,5,6,7,8,9...0,1,2,3,4,5,6,7,8const subscribe = example.subscribe(val => console.log(val));

3.3.6 switchMapTo

和switchMap很相似，只不过只能映射成固定的Observable输出。

3.3.7 concatMap（按observable产生的顺序来发出值，只有上一个observable发出的值订阅完成，才会开始下一个订阅）（相当于map+concatAll）

// 立即发出值， 然后每z秒发出值
const source = timer(0, 2000);
// 当 source 发出值时,只有完成当前的订阅，才能开始下一个Observable的订阅，想数组一样concat
const example = source.pipe(concatMap(x => interval(500)));
// 输出: 0,1,2,3,4,5,6,7,8,9,20,....
const subscribe = example.subscribe(val => console.log(val));

3.3.8 concatMapTo

和concatMap很相似，只不过只能映射成固定的Observable输出。

3.4 组合操作符

3.4.1 concat

// 发出 1,2,3
const sourceOne = of(1, 2, 3);
// 发出 4,5,6
const sourceTwo = of(4, 5, 6);// 作为静态方法使用，输出: 1,2,3,4,5,6
concat(sourceOne, sourceTwo).subscribe (val => console.log(val));

3.4.2 concatAll(收集 observables，当前一个完成时订阅下一个)

把内部产生的多个observables合并成一个observable，传递的值串联起来
高阶Observable：就是observable内部会产生新的observable。

// 创建并解析一个基础的 promise
const samplePromise = val => new Promise(resolve => resolve(val));
// 每2秒发出值
const source = interval(2000);const example = source.pipe(map(val => samplePromise(val)),// 合并解析过的 promise 的值// concatAll()
);
// 输出: 'Example with Promise 0', 'Example with Promise 1'...
const subscribe = example.subscribe(val =>console.log('Example with Promise:', val)
);

没有使用（concatAll)输出：

Example with Promise: Promise { 0 }
Example with Promise: Promise { 1 }
Example with Promise: Promise { 2 }
Example with Promise: Promise { 3 }

使用（concatAll)输出：

Example with Promise: 0
Example with Promise: 1
Example with Promise: 2
Example with Promise: 3

注意内部产生的多个observable,只有上一个complete了，才会订阅下一个

3.4.3 merge

// 每2.5秒发出值
const first = interval(2500);
// 每2秒发出值
const second = interval(2000);
// 每1.5秒发出值
const third = interval(1500);
// 每1秒发出值
const fourth = interval(1000);// 从一个 observable 中发出输出值
const example = merge(first.pipe(mapTo('FIRST!')),second.pipe(mapTo('SECOND!')),third.pipe(mapTo('THIRD')),fourth.pipe(mapTo('FOURTH'))
);
// 输出: "FOURTH", "THIRD", "SECOND!", "FOURTH", "FIRST!", "THIRD", "FOURTH"
const subscribe = example.subscribe(val => console.log(val));

3.4.4 mergeAll

const source = of (1000, 2000, 3000);
const example = source.pipe(map(val => interval(val)),mergeAll()
).subscribe(val => console.log(val));

和mergeMap效果一样

3.4.5 combineLatest

[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-YXaGqL5V-1629689242026)(https://ceph-dev-pub.dz11.com/fed-doc/1628843244862.png)]

// timerOne 在1秒时发出第一个值，然后每4秒发送一次
const timerOne = timer(1000, 4000);
// timerTwo 在2秒时发出第一个值，然后每4秒发送一次
const timerTwo = timer(2000, 4000);
// timerThree 在3秒时发出第一个值，然后每4秒发送一次
const timerThree = timer(3000, 4000);// 当一个 timer 发出值时，将每个 timer 的最新值作为一个数组发出
const combined = combineLatest(timerOne, timerTwo, timerThree);const subscribe = combined.subscribe(latestValues => {// 从 timerValOne、timerValTwo 和 timerValThree 中获取最新发出的值const [timerValOne, timerValTwo, timerValThree] = latestValues;/*输出:'Timer One Latest: 0, Timer Two Latest:0, Timer Three Latest: 0'Timer One Latest: 1, Timer Two Latest:0, Timer Three Latest: 0'Timer One Latest: 1, Timer Two Latest:1, Timer Three Latest: 0'Timer One Latest: 1, Timer Two Latest:1, Timer Three Latest: 1*/console.log(`Timer One Latest: ${timerValOne},Timer Two Latest: ${timerValTwo},Timer Three Latest: ${timerValThree}`);
});

3.4.6 combineAll

当源 observable 完成时，对收集的 observables 使用 combineLatest 。

// timerOne 在1秒时发出第一个值，然后每4秒发送一次
const timero = val => timer(val, 4000);const subscribe = of(1000, 2000, 3000).pipe(map(val => timero(val)),combineAll()
).subscribe(latestValues => {const [timerValOne, timerValTwo, timerValThree] = latestValues;console.log(`Timer One Latest: ${timerValOne},Timer Two Latest: ${timerValTwo},Timer Three Latest: ${timerValThree}`);combineLatess
});

和上面的combineLatest输出结果一样

3.4.7 zip

// timerOne 在1秒时发出第一个值，然后每4秒发送一次
const timerOne = timer(1000, 4000);
// timerTwo 在2秒时发出第一个值，然后每4秒发送一次
const timerTwo = timer(2000, 4000);
// timerThree 在3秒时发出第一个值，然后每4秒发送一次
const timerThree = timer(3000, 4000);const combined = zip(timerOne, timerTwo, timerThree).subscribe(latestValues => {const [timerValOne, timerValTwo, timerValThree] = latestValues;console.log(`Timer One Latest: ${timerValOne},Timer Two Latest: ${timerValTwo},Timer Three Latest: ${timerValThree}`);
});

3.4.8 zipAll

const subscribe = of(1000, 2000, 3000).pipe(map(val => timero(val)),zipAll()
).subscribe(latestValues => {//  等待timerValOne、timerValTwo 和 timerValThree各发出一个值之后组合，如何再等待下一次都新发完一个值组合const [timerValOne, timerValTwo, timerValThree] = latestValues;console.log(`Timer One Latest: ${timerValOne},Timer Two Latest: ${timerValTwo},Timer Three Latest: ${timerValThree}`);
});

3.5 过滤操作符

3.5.1 filter（对源Observable发出的值进行过滤）

var te = of(1, 2, 3);
te.pipe(filter(x => {return x >= 2} )).subscribe({next: x => console.log(x)
})

输出：

2
3

3.5.2 first(订阅且仅订阅第一个满足某个条件的值)

const source = (val) =>  timer(val, 1000);
const ob = of(2000, 1000, 3000);
// 当 source 发出值时并不会取消前一个内部observal的订阅
const example = ob.pipe(mergeMap(val => source(val)),first(x => x > 2)
);
// 输出: 3
const subscribe = example.subscribe(val => console.log(val));

3.5.3 take(订阅固定前n个值)

const ob = of(2000, 1000, 3000);
const example = ob.pipe(take(2)
);
// 输出: 2000,1000
const subscribe = example.subscribe(val => console.log(val));

3.5.4 last（订阅满足条件的最后一个值）

const ob = of(2000, 1000, 3000);
const example = ob.pipe(// 订阅满足条件的最后一个值last(x => x === 1000)
);
// 输出: 2000,1000
const subscribe = example.subscribe(val => console.log(val));

3.5.5 single( 订阅第一个满足条件的值，如果有两个值满足条件，则会报错)

const ob = of(2000, 1000, 3000);
const example = ob.pipe(// 订阅满足条件的最后一个值single(x => x === 1000)
);
// 输出: 2000,1000
const subscribe = example.subscribe(val => console.log(val));

3.5.6 debounce(防抖)

const ob = interval(1000);
const example = ob.pipe(// 只有订阅的下一个值是在前一个值的2秒之后发出的，前一个值才会被订阅debounce(x => timer(2000))
);
// 输出: 空
const subscribe = example.subscribe(val => console.log(val));

3.5.7 debounceTime(防抖)

const ob = interval(1000);
const example = ob.pipe(// 只有订阅的下一个值是在前一个值的2秒之后发出的，前一个值才会被订阅debounceTime(2000)
);
// 输出: 空
const subscribe = example.subscribe(val => console.log(val));

3.5.8 throttle (节流)

const ob = interval(500);
const example = ob.pipe(// 忽略1秒之内的订阅值，超过1秒才开始订阅下一个值throttle(x => timer(1000))
);
// 输出: 0, 3, 6, 9
const subscribe = example.subscribe(val => console.log(val));

3.5.9 throttleTime（节流）

const ob = interval(500);
const example = ob.pipe(// 忽略1秒之内的订阅值，超过1秒才开始订阅下一个值throttleTime(1000)
);
// 输出: 0, 3, 6
const subscribe = example.subscribe(val => console.log(val));

3.5.10 takeUntil

// 每1秒发出值
const source = interval(1000);
// 5秒后发出值
const timer$ = timer(5000);
// 当5秒后 timer 发出值时， source 则完成
const example = source.pipe(takeUntil(timer$));
// 输出: 0,1,2,3
const subscribe = example.subscribe(val => console.log(val));

3.5.11 takeWhile

const ob = interval(500);
const example = ob.pipe(// 一旦false就不再订阅takeWhile(x => x < 5)
);
// 输出: 0, 1, 2, 3, 4
const subscribe = example.subscribe(val => console.log(val));