Eine kurze Analyse beobachtbarer Objekte, Beobachter und RxJS-Operatoren in Angular

Dieser Artikel stellt Ihnen die Operatoren Observable, Observer und RxJS inAngularvor. Ich hoffe, er wird Ihnen hilfreich sein!

Observable (observable object)

Observable(observable object) ist ein Objekt in derRxJS-Bibliothek, das zur Verarbeitung asynchroner Ereignisse verwendet werden kann, z HTTP-Anfrage (eigentlich geben in Angular alle HTTP-Anfragen Observable zurück). [Verwandte Tutorial-Empfehlung: „Observable(可观察对象)，是RxJS库里面的一个对象，可以用来处理异步事件，例如HTTP请求（实际上，在Angular中，所有的HTTP请求返回的都是Observable）。【相关教程推荐：《angular教程》】

或许，你以前接触过一个叫promise的东西，它们本质上面是相同的：都是生产者主动向消费者“push”产品，而消费者是被动接收的，但是他们两者还是有很大区别的：Observable可以发送任意多值，并且，在被订阅之前，它是不会执行的！这是promise不具备的特点。

• Observable用于在发送方和接收方之间传输消息，你可以将这些消息看成是流
• 在创建Observable对象时，需要传入一个函数作为构造函数的参数，这个函数叫订阅者函数，这个函数也就是生产者向消费者推送消息的地方
• 在被消费者subscribe（订阅）之前，订阅者函数不会被执行，直到subscribe()函数被调用，该函数返回一个subscription对象，里面有一个unsubscribe()函数，消费者可以随时拒绝消息的接收！
• subscribe()函数接收一个observer（观察者）对象作为入参
• 消息的发送可以是同步的，也可以是异步的

observer（观察者）

有了可观察对象（发送方），就需要一个观察者（接收方）来观察可观察对象，观察者要实现observerAngular-Tutorial

• Observablewird verwendet, um Nachrichten zwischen dem Sender und dem Empfänger zu übertragen. Sie können diese Nachrichten als Streams betrachten
• Beim Erstellen vonObservableObjekt, Sie müssen eine Funktion als Parameter des Konstruktors übergeben. Diese Funktion wird alsAbonnentenfunktionbezeichnet. Bei dieser Funktion sendet der Produzent Nachrichten an den Verbraucher
Da der Verbraucher subscribe(abonniert) hat, wird die Abonnentenfunktion erst ausgeführt, wenn die Funktion subscribe()aufgerufen wird, die ein subscriptionzurückgibt Objekt, das eine unsubscribe()-Funktion enthält, kann der Verbraucher den Empfang der Nachricht jederzeit verweigern!
• Die Funktionsubscribe()erhält einobserver-Objekt als Eingabeparameter
• Das Versenden von Nachrichten kann synchron erfolgen, oder es kann asynchron sein

Abonnements

Nur wenn jemand eine Instanz vonObservableabonniert, werden Werte veröffentlicht. Beim Abonnieren müssen Sie zunächst die Methodesubscribe()des beobachtbaren Objekts aufrufen und ihm ein Beobachterobjekt übergeben, um Benachrichtigungen zu erhalten. Wie folgt:Observable的实例时，它才会开始发布值。 订阅时要先调用可观察对象的subscribe()方法，并把一个观察者对象传给它，用来接收通知。如下：

为了展示订阅的原理，需要先创建新的可观察对象。它有一个构造函数可以用来创建新实例，但是为了更简明，也可以使用Observable上定义的一些静态方法来创建一些常用的简单可观察对象：

• of(...items)：返回一个Observable实例，它用同步的方式把参数中提供的这些值一个一个发送出来。
• from(iterable)： 把它的参数转换成一个Observable实例。 该方法通常用于把一个数组转换成一个（发送多个值的）可观察对象。

import { of } from "rxjs"; // 1、通过 of() 方法返回一个可观察对象，并准备将1，2，3三个数据发送出去 const observable = of(1, 2, 3); // 2、实现 observer 接口，观察者 const observer = { next: (num: number) => console.log(num), error: (err: Error) => console.error('Observer got an error: ' + err), complete: () => console.log('Observer got a complete notification'), } // 3、订阅。调用可观察对象的 subscribe() 方法订阅，subscribe() 方法中传入的对象就是一个观察者 observable.subscribe(observer);

运行结果如下：

上面订阅的写法可以直接改为如下：参数不是对象

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

订阅者函数

在上面的例子中使用的是of()方法来创建可观察对象，这节使用构造函数创建可观察对象。

Observable构造函数可以创建任何类型的可观察流。 当执行可观察对象的subscribe()方法时，这个构造函数就会把它接收到的参数作为订阅函数来运行。 订阅函数会接收一个Observer对象，并把值发布给观察者的next()方法。

// 1、自定义订阅者函数 function sequenceSubscriber(observer: Observer) { observer.next(1); // 发送数据 observer.next(2); // 发送数据 observer.next(3); // 发送数据 observer.complete(); return {unsubscribe() {}}; } // 2、通过构造函数创建一个新的可观察对象，参数就是一个订阅者函数 const sequence = new Observable(sequenceSubscriber); // 3、订阅 sequence.subscribe({ next(num) { console.log(num); }, // 接受数据 complete() { console.log('Finished sequence'); } });

运行结果如下：

上面一个例子演示了如何自定义订阅函数，那么既然可以自定义订阅者函数，我们就可以将异步代码封装进可观察对象的订阅者函数中，待异步代码执行完再发送数据。如下：

import { Observable } from 'rxjs' // 异步函数 function fn(num) { return new Promise((reslove, reject) => { setTimeout(() => { num++ reslove(num) }, 1000) }) } // 创建可观察对象，并传入订阅者函数 const observable = new Observable((x) => { let num = 1 fn(num).then( res => x.next(res) // 异步代码执行完成，发送数据 ) }) // 订阅，接收数据，可以改为链式调用 observable.subscribe(data => console.log(data)) // 2

多播

https://angular.cn/guide/observables#multicasting

RxJS操作符

我们可以使用一系列的RxJS操作符，在这些消息被接收方接收之前，对它们进行一系列的处理、转换，因为这些操作符都是纯函数。

import { of } from 'rxjs'; import { map } from 'rxjs/operators'; // 1、创建可观察对象，并发送数据 const nums = of(1, 2, 3); // 2、创建函数以接受可观察对象 const squareValues = map((val: number) => val * val); const squaredNums = squareValues(nums); squaredNums.subscribe(x => console.log(x));

上面的方式我看不懂且难以接受，一般常用下面这种，使用pipe把多个操作符链接起来

import { map, Observable, filter } from 'rxjs' // 创建可观察对象，并传入订阅者函数 const observable = new Observable((x) => { x.next(1) x.next(2) x.next(3) x.next(4) }).pipe( map(value => value*100), // 操作符 filter(value => value == 200) // 操作符 ) .subscribe(data => console.log(data)) // 200

错误处理

RxJS还提供了catchError

Um das Prinzip des Abonnements zu demonstrieren, müssen Sie zunächst ein neues beobachtbares Objekt erstellen. Es verfügt über einen Konstruktor, der zum Erstellen neuer Instanzen verwendet werden kann. Der Einfachheit halber können Sie jedoch auch einige auf Observabledefinierte statische Methoden verwenden, um einige häufig verwendete einfache beobachtbare Objekte zu erstellen:

< li > of(...items): Gibt eine Observable-Instanz zurück, die die in den Parametern bereitgestellten Werte eins nach dem anderensendet eine synchrone Art und Weise herauskommen.
• from(iterable): Konvertiert sein Argument in eineObservable-Instanz. Diese Methode wird normalerweise verwendet, um ein Array in ein beobachtbares Objekt umzuwandeln (das mehrere Werte sendet).

import { map, Observable, filter, catchError, of } from 'rxjs' const observable = new Observable((x) => { x.next(1) // 发送数据 1 和 2 x.next(2) }).pipe( map(value => { if (value === 1) { // 1、当发送的数据为 1 时，将其乘以 100 return value*100 } else { // 2、否则抛出错误 throw new Error('抛出错误'); } }), // 3、此处捕获错误并处理错误，对外发送数据 0 catchError((err) => { console.log(err) return of(0) }) ) .subscribe( data => console.log(data), // 4、由于上面抛出的错误被 catchError 操作符处理（重新发送数据）了，所以这里能顺利订阅到数据而不报错 err => console.log('接受不到数据：', err) )

Die laufenden Ergebnisse sind wie folgt:

rrreeeAbonnentenfunktion