RxJS 入门

什么是RxJS？

RxJS代表响应式扩展JavaScript（Reactive Extensions Library for JavaScript）。它是一个用于处理事件流和异步数据流的库，可以将这些流组合起来以产生更复杂的结果。

响应式编程

响应式编程（Reactive Programming）是一种编程范式，它关注于数据流和变化的传播。这种编程方式特别适用于处理异步数据流

• 事件驱动(addEventListener/dispatchEvent)

容易产生“回调地狱”（Callback Hell），使代码难以阅读和维护。

// 注册事件
addEventListener('click',(e)=>{
    console.log(e)
    console.log('click')
})

// 触发事件
dispatchEvent(new Event('click'))

• Promise (Async await)

const p = new Promise((resolve, reject) => {
    resolve('Success')
})

p.then((value) => {
    console.log(value)
})

• RxJS

const a = new Observable((observer) => {
    observer.next(1);
    observer.next(2);
    observer.next(3);
    setTimeout(() => {
        observer.next(4);
        observer.complete();
    }, 3);
});
a.subscribe((x) => console.log(x))

RxJS的主要核心概念

• Observable: 表示一个可调用的未来值或事件的集合。Observable并不是实际的数据，而是产生数据的源头。
• Observer: 一个回调函数的集合，用于处理由Observable发出的值，包括next、error和complete。
• Subscription: 表示Observable和Observer之间订阅关系的对象。它提供了一种机制来管理这种关系，包括取消订阅以释放资源。当一个Observer订阅了一个Observable时，会返回一个Subscription对象。这个对象允许Observer取消订阅，或者管理多个订阅。

RxJS：一个返回值是数据流的函数

Observables 像是没有参数, 但可以泛化为多个值的函数。

考虑如下代码：

function foo() {
  console.log('Hello');
  return 42;
}

var x = foo.call(); // 等同于 foo()
console.log(x);
var y = foo.call(); // 等同于 foo()
console.log(y);

我们期待看到的输出：

"Hello"
42
"Hello"
42

你可以使用Observables重写上面的代码：

var foo = new Observable(function (observer) {
  console.log('Hello');
  observer.next(42);
});

foo.subscribe(function (x) {
  console.log(x);
});
foo.subscribe(function (y) {
  console.log(y);
});

输出是一样的:

"Hello"
42
"Hello"
42

区别是Observables的返回值可以是不同时间产生的多个值。

函数和Observables都是惰性计算，如果你不调用函数，console.log('Hello') 就不会执行。Observables也是如此，如果你不“调用”它(使用subscribe)，console.log('Hello') 也不会执行。此外，“调用”或“订阅”是独立的操作：两个函数调用会触发两个单独的副作用，两个Observable 订阅同样也是触发两个单独的副作用。EventEmitters共享副作用并且无论是否存在订阅者都会尽早执行，Observables与之相反，不会共享副作用并且是延迟执行。

Observables 传递值可以是同步的，也可以是异步的。

Observable 可以随着时间的推移“返回”多个值，这是函数所做不到的。你无法这样：

function foo() {
  console.log('Hello');
  return 42;
  return 100; // 死代码，永远不会执行
}

函数只能返回一个值。但 Observables 可以这样：

var foo = new Observable(function (observer) {
  console.log('Hello');
  observer.next(42);
  observer.next(100);
  observer.next(200);
  setTimeout(() => {
    observer.next(300); // 异步执行
  }, 1000);
});

console.log('before');
foo.subscribe(function (x) {
  console.log(x);
});
console.log('after');

输出

"before"
"Hello"
42
100
200
"after"
300

Observables不仅返回了多个值，其中还在1秒后异步返回了一个300 我们可以理解这是一个同步或异步的数据流，当你完成订阅后，就能够获得一个持续的数据流动。

RxJS：Operators操作数据流的函数

我们通过订阅一个Observable可以得到一个数据流，如果我们在订阅后，再通过一个新的Observable将订阅获得的数据流进行一定操作再发送出去，这样如果有新的Observer订阅。

import { Observable } from "rxjs";

const a = new Observable((observer) => {
    observer.next(1);
    observer.next(2);
    observer.next(3);
    setTimeout(() => {
        observer.next(4);
        observer.complete();
    }, 3);
});

const b = new Observable<number>((observer) => {
    // 想要每个值+1
    a.subscribe((x) => {
        observer.next(x + 1);
    });
});

console.log("just before subscribe");
b.subscribe((x) => console.log(x));
console.log("just after subscribe");

得到输出

just before subscribe
2
3
4
just after subscribe
5

可以看到，我们通过订阅b得到了对订阅a后的数据流+1的数据流。 像这样的操作，我将其理解为对数据流的操作，如果有多个这样操作，比如c订阅b再进行操作，这样就形成了订阅链。

正常情况下，数据流的处理都是相当复杂的，在RxJS中提供了很多的操作符(Operator)来操作数据流。

如果要实现类似上面+1的逻辑，可以使用map操作符：

import { Observable, map } from "rxjs";

const a = new Observable((observer) => {
    observer.next(1);
    observer.next(2);
    observer.next(3);
    setTimeout(() => {
        observer.next(4);
        observer.complete();
    }, 3);
});

const b = a.pipe(
    map((x) => x + 1)
)

console.log("just before subscribe");
b.subscribe((x) => console.log(x));
console.log("just after subscribe");

map操作符的作用是将每一个数据流中的数据操作一遍，以上代码执行会获得一样的输出结果。 我们通过pipe去使用RxJS给我们提供的操作符，如字面意思一样，可以理解为管道。 pipe函数可以接受多个操作符，并且按顺序处理数据流：

import { Observable, map } from "rxjs";

const a = new Observable((observer) => {
    observer.next(1);
    observer.next(2);
    observer.next(3);
    setTimeout(() => {
        observer.next(4);
        observer.complete();
    }, 3);
});

const b = a.pipe(
    map((x) => x + 1),
    map((x) => x * 2 + "2")
)

console.log("just before subscribe");
b.subscribe((x) => console.log(x));
console.log("just after subscribe");

以上代码在pipe中连续使用了两个map操作符，这使得数据流会按顺序先给第一个map处理，再给第二个map处理，得到的输出结果是：

just before subscribe
42
62
82
just after subscribe
102

更多操作符

https://rxjs.dev/guide/operators

https://segmentfault.com/a/1190000044727567