Angular 8 - 响应式编程

响应式编程是一种处理数据流和变化传播的编程范式。数据流可以是静态的或动态的。静态数据流的例子是数组或数据集合。它将具有初始数量，并且不会更改。动态数据流的例子是事件发射器。事件发射器在事件发生时发出数据。最初，可能没有事件，但随着时间的推移，事件会发生并被发出。

响应式编程使数据流能够从称为Observable的一个源发出，并通过称为订阅的过程由称为Observer的其他源捕获发出的数据流。这种 Observable/Observer 模式或简单的Observer模式极大地简化了编程环境中复杂的变化检测和必要的更新。

JavaScript 本身并不支持响应式编程。RxJs是一个使 JavaScript 能够进行响应式编程的 JavaScript 库。Angular 广泛使用RxJs库来实现以下高级概念：

组件之间的数据传输。
HTTP 客户端。
路由器。
响应式表单。

让我们在本节中使用RxJs库学习响应式编程。

Observable

如前所述，Observable是数据源，它们可以是静态的或动态的。Rxjs提供了许多从常用 JavaScript 对象创建Observable的方法。让我们看看一些常见的方法。

of - 按顺序发出任意数量的值，最后发出完成通知。

const numbers$ = of(1, 2, 3, 4, 5, 6, 7, 8, 9, 10);

这里：

numbers$是一个Observable对象，订阅时将按顺序发出 1 到 10。
变量末尾的美元符号 ($)用于标识该变量是 Observable。

range - 按顺序发出一定范围的数字。

const numbers$ = range(1,10)

from - 发出数组、Promise 或可迭代对象。

const numbers$ = from([1,2,3,4,5,6,7,8,9,10]);

ajax - 通过 AJAX 获取 url，然后发出响应。

const api$ = ajax({ url: 'https://httpbin.org/delay/1', method: 'POST', headers: { 'Content-Type': 'application/text' }, body: "Hello" });

这里：

https://httpbin.org 是一款免费的 REST API 服务，它将以 JSON 格式返回提供的正文内容，如下所示：

{ 
   "args": {}, 
   "data": "Hello", 
   "files": {}, 
   "form": {}, 
   "headers": { 
      "Accept": "text/html,application/xhtml+xml,application/xml;q=0.9,image/webp,image/apng,*/*;q=0.8,application/signed-exchange;v=b3;q=0.9", 
      "Accept-Encoding": "gzip, deflate, br", 
      "Accept-Language": "en-US,en;q=0.9", 
      "Host": "httpbin.org", "Sec-Fetch-Dest": "document", 
      "Sec-Fetch-Mode": "navigate", 
      "Sec-Fetch-Site": "none", 
      "Upgrade-Insecure-Requests": "1", 
      "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/83.0.4103.106 Safari/537.36", 
      "X-Amzn-Trace-Id": "Root=1-5eeef468-015d8f0c228367109234953c" 
   }, 
   "origin": "ip address", 
   "url": "https://httpbin.org/delay/1" 
}

fromEvent - 监听 HTML 元素的事件，然后在监听的事件触发时发出事件及其属性。

const clickEvent$ = fromEvent(document.getElementById('counter'), 'click');

Angular 在内部广泛使用此概念来提供组件之间的数据传输和响应式表单。

订阅过程

订阅 Observable 非常容易。每个 Observable 对象都有一个用于订阅过程的方法，subscribe。Observer 需要实现三个回调函数来订阅 Observable 对象。它们如下所示：

next - 接收和处理从 Observable 发出的值
error - 错误处理回调
complete - 当 Observable 中的所有数据都发出时调用的回调函数。

定义完这三个回调函数后，必须调用 Observable 的 subscribe 方法，如下所示：

const numbers$ = from([1,2,3,4,5,6,7,8,9,10]); 
// observer 
const observer = { 
   next: (num: number) => {      this.numbers.push(num); this.val1 += num }, 
      error: (err: any) => console.log(err), 
      complete: () => console.log("Observation completed") 
}; 
numbers$.subscribe(observer);

这里：

next - 方法获取发出的数字，然后将其推入局部变量this.numbers中。
next - 方法还将数字添加到局部变量this.val1中。
error - 方法只是将错误消息写入控制台。
complete - 方法也向控制台写入完成消息。

我们可以跳过error和complete方法，只编写next方法，如下所示：

number$.subscribe((num: number) => { this.numbers.push(num); this.val1 += num; });

操作符

Rxjs库提供了一些操作符来处理数据流。一些重要的操作符如下：

filter - 使用回调函数过滤数据流。

const filterFn = filter( (num : number) => num > 5 ); 
const filteredNumbers$ = filterFn(numbers$); 
filteredNumbers$.subscribe( (num : number) => { 
this.filteredNumbers.push(num); this.val2 += num } );

map - 使用回调函数映射数据流并更改数据流本身。

const mapFn = map( (num : number) => num + num ); const mappedNumbers$ = mappedFn(numbers$);

pipe - 使能够组合两个或多个操作符。

const filterFn = filter( (num : number) => num > 5 ); 
const mapFn = map( (num : number) => num + num ); const processedNumbers$ = numbers$.pipe(filterFn, mapFn); 
processedNumbers$.subscribe( (num : number) => { this.processedNumbers.push(num); this.val3 += num } );

让我们创建一个示例应用程序来试用本章中学习的响应式编程概念。

使用以下命令创建一个新的应用程序 reactive：

ng new reactive

将目录更改为我们新创建的应用程序。

cd reactive

运行应用程序。

ng serve

更改 AppComponent 组件代码 (src/app/app.component.ts)，如下所示：

import { Component, OnInit } from '@angular/core'; import { Observable, of, range, from, fromEvent } from 'rxjs'; 
import { ajax } from 'rxjs/ajax'; 
import { filter, map, catchError } from 'rxjs/operators'; 
@Component({ 
   selector: 'app-root', 
   templateUrl: './app.component.html', 
   styleUrls: ['./app.component.css'] 
}) 
export class AppComponent implements OnInit { 
   title = 'Reactive programming concept'; 
   numbers : number[] = []; 
   val1 : number = 0; 
   filteredNumbers : number[] = []; 
   val2 : number = 0; 
   processedNumbers : number[] = []; 
   val3 : number = 0; 
   apiMessage : string; 
   counter : number = 0; 
   ngOnInit() { 
      // Observable stream of data Observable<number>
      // const numbers$ = of(1, 2, 3, 4, 5, 6, 7, 8, 9, 10); 
      // const numbers$ = range(1,10); 
      const numbers$ = from([1,2,3,4,5,6,7,8,9,10]); 
      // observer 
      const observer = { 
         next: (num: number) => {this.numbers.push(num); this.val1 += num }, 
         error: (err: any) => console.log(err), 
         complete: () => console.log("Observation completed") 
      }; 
      numbers$.subscribe(observer); 
      const filterFn = filter( (num : number) => num > 5 ); 
      const filteredNumbers = filterFn(numbers$); 
      filteredNumbers.subscribe( (num : number) => {this.filteredNumbers.push(num); this.val2 += num } ); 
      const mapFn = map( (num : number) => num + num ); 
      const processedNumbers$ = numbers$.pipe(filterFn, mapFn); 
      processedNumbers$.subscribe( (num : number) => {this.processedNumbers.push(num); this.val3 += num } ); 
      const api$ = ajax({ 
         url: 'https://httpbin.org/delay/1', 
         method: 'POST', 
         headers: {'Content-Type': 'application/text' }, 
         body: "Hello" 
      }); 
      api$.subscribe(res => this.apiMessage = res.response.data ); 
      const clickEvent$ = fromEvent(document.getElementById('counter'), 'click'); 
      clickEvent$.subscribe( () => this.counter++ ); 
   } 
}

这里：

使用 of、range、from、ajax 和 fromEvent 方法创建 Observable。
使用 filter、map 和 pipe 操作符方法处理数据流。
回调函数捕获发出的数据，处理它，然后将其存储在组件的局部变量中。

更改AppComponent模板(src/app/app.component.html)，如下所示：

<h1>{{ title }}</h1> 
<div> 
   The summation of numbers ( <span *ngFor="let num of numbers"> {{ num }} </span> ) is {{ val1 }} 
</div> 
<div> 
   The summation of filtered numbers ( <span *ngFor="let num of filteredNumbers"> {{ num }} </span> ) is {{ val2 }} 
</div> 
<div> 
   The summation of processed numbers ( <span *ngFor="let num of processedNumbers"> {{ num }} </span> ) is {{ val3 }} 
</div> 
<div> 
   The response from the API is <em>{{ apiMessage }}</em> </div> 
<div> 
   <a id="counter" href="#">Click here</a> to increment the counter value. The current counter value is {{ counter }} 
<div>

这里：

显示Observer回调函数处理的所有局部变量。

打开浏览器，https://127.0.0.1:4200。

单击单击此处链接五次。对于每个事件，都会发出事件并转发到Observer。将调用 Observer 回调函数。回调函数会为每次点击递增计数器，最终结果如下所示：

打印页面