Angular 2 AsyncPipe

更新时间 - 2017-03-21 15：02

今天我们来介绍一下 Angular 2 中 AsyncPipe (异步管道) ，使用 AsyncPipe 我们可以直接在模板中使用 Promise 和 Observable 对象，而不用通过定义一个类的成员属性来存储返回的结果。

AsyncPipe 订阅一个 Observable 或 Promise 对象，并返回它发出的最新值。 当发出新值时，异步管道会主动调用变化检测器的 markForCheck() 方法，标识组件需执行变化检测。 当组件被销毁时，异步管道自动取消订阅，以避免潜在的内存泄漏。

AsyncPipe with Promise

Promise 未使用 AsyncPipe

promise.component.ts

import { Component } from '@angular/core';

@Component({
    selector: 'exe-promise',
    template: `
     <h4>Promise Component</h4>
     <p>{{promiseData}}</p>
    `
})
export class PromiseComponent {
    promiseData: string;

    constructor() {
        this.getPromise().then(v => this.promiseData = v);
    }

    getPromise(): Promise<string> {
        return new Promise((resolve, reject) => {
            setTimeout(() => {
                resolve('Promise complete!');
            }, 2000);
        });
    }
}

app.component.ts

import { Component } from '@angular/core';

@Component({
  selector: 'exe-app',
  template: `
   <exe-promise></exe-promise>
  `
})
export class AppComponent { }

以上代码运行后浏览器显示的结果：

Promise 使用 AsyncPipe

promise-async-pipe.component.ts

import { Component } from '@angular/core';

@Component({
    selector: 'exe-promise-pipe',
    template: `
     <h4>Promise with AsyncPipeComponent</h4>
     <p>{{ promise | async }}</p>
    `
})
export class PromiseAsyncPipeComponent {
    promise: Promise<string>;
    constructor() {
       this.promise = this.getPromise();
    }

    getPromise(): Promise<string> {
        return new Promise((resolve, reject) => {
            setTimeout(() => {
                resolve('Promise with AsyncPipe complete!');
            }, 2000);
        });
    }
}

app.component.ts

import { Component } from '@angular/core';

@Component({
  selector: 'exe-app',
  template: `
   <exe-promise-pipe></exe-promise-pipe>
  `
})
export class AppComponent { }

以上代码运行后浏览器显示的结果：

AsyncPipe with Observables

Observable 未使用 AsyncPipe

observable.component.ts

import { Observable, Subscription } from 'rxjs/Rx';
import { Component, OnDestroy } from '@angular/core';

@Component({
    selector: 'exe-observable',
    template: `
     <h4>Observable Component</h4>
     <p>{{ observableData }}</p>
    `
})
export class ObservableComponent implements OnDestroy {
    observableData: number;
    subscription: Subscription = null;

    constructor() {
        this.subscribeObservable();
    }

    getObservable(): Observable<number> {
        return Observable
            .interval(1000) // 每隔1秒，生成一个值
            .take(10) // 获取前面10个数据
            .map(v => v * v); // 对每个数据进行乘方处理
    }

    subscribeObservable() {
        this.subscription = this.getObservable()
            .subscribe(v => this.observableData = v);
    }

    ngOnDestroy() { // 组件销毁时取消订阅，以避免潜在的内存泄漏
        if (this.subscription) {
            this.subscription.unsubscribe();
        }
    }
}

app.component.ts

import { Component } from '@angular/core';

@Component({
  selector: 'exe-app',
  template: `
   <exe-observable></exe-observable>
  `
})
export class AppComponent { }

以上代码运行后浏览器显示的结果：

Observable 使用 AsyncPipe

observable-async-pipe.component.ts

import { Component } from '@angular/core';
import { Observable } from 'rxjs/Rx';

@Component({
    selector: 'exe-observable-pipe',
    template: `
     <h4>Observable with AsyncPipe Component</h4>
     <p>{{ observable | async }}</p>
    `
})
export class ObservableAsyncPipeComponent {
    observable: Observable<number>

    constructor() {
        this.observable = this.getObservable();
    }

    getObservable(): Observable<number> {
        return Observable
            .interval(1000)
            .take(10)
            .map(v => v * v);
    }
}

app.component.ts

import { Component } from '@angular/core';

@Component({
  selector: 'exe-app',
  template: `
   <exe-observable-pipe></exe-observable-pipe>
  `
})
export class AppComponent { }

以上代码运行后浏览器显示的结果：

为了更让读者更好地掌握 AsyncPipe， 我们再来一个示例：

import { Component, OnInit } from '@angular/core';
import { Observable } from 'rxjs/Rx';

interface Hero {
    id: number;
    name: string;
}

@Component({
    selector: 'async-pipe-example',
    template: `
       <h4>Async Pipe Example</h4>
       <div>
          <h5>Heroes: </h5>
          <ul>
            <li *ngFor="let hero of heroes$ | async">
                {{ hero.name }}
            </li>
          </ul> 
          <h5>Hero: </h5>
          <p> 
            <span *ngIf="hero$">{{ (hero$ | async).id }}</span>
            <span>{{ (hero$ | async)?.name }}</span>
          </p>
       </div>
    `
})
export class AsyncPipeComponent implements OnInit {
    heroes$: Observable<Hero[]>;
    hero$: Observable<Hero>;

    getHeroes(): Observable<Hero[]> {
        return Observable.of([
            { id: 1, name: 'Windstorm' },
            { id: 13, name: 'Bombasto' },
            { id: 15, name: 'Magneta' },
            { id: 20, name: 'Tornado' }
        ]);
    }

    getHero(): Observable<Hero> {
        return Observable.of({
            id: 31, name: 'Semlinker'
        });
    }

    ngOnInit() {
        setTimeout(() => {
            this.heroes$ = this.getHeroes();
            this.hero$ = this.getHero();
        }, 2000);
    }
}

以上代码运行后浏览器显示的结果：

上面例子中有两个注意点：

1.使用 ngIf 控制 span 元素的显示：

<span *ngIf="hero$">{{ (hero$ | async).id }}</span>

2.使用 ?. 安全导航操作符，控制 name 属性的显示：

<span>{{ (hero$ | async)?.name }}</span>

若去掉 ngIf 或 ?. ，应用程序将会抛出异常，建议读者亲身体验一下。

我有话说

1.Promise vs Observable

• Promise

  • 返回单个值

  • 不可取消的

• Observable

  • 随着时间的推移发出多个值

  • 可以取消的

  • 支持 map、filter、reduce 等操作符

  • 延迟执行，当订阅的时候才会开始执行

详细内容可以参考 - RxJS 核心概念之Observable

2.使用 AsyncPipe 会发送多次请求

我们直接看一下具体示例：

app.component.ts

import { Component, OnInit } from '@angular/core';
import { Http } from '@angular/http';
import { Observable } from 'rxjs/Rx';

@Component({
  selector: 'exe-app',
  template: `
   <div>
      <p>{{ (person$ | async)?.id }}</p>
      <p>{{ (person$ | async)?.title }}</p>
      <p>{{ (person$ | async)?.body }}</p>
    </div>
  `
})
export class AppComponent implements OnInit {
  person$: Observable<{ id: number; title: string; body: string }>;

  constructor(private http: Http) { }

  ngOnInit() {
    this.person$ = this.http.get('https://jsonplaceholder.typicode.com/posts/1')
      .map(res => res.json())
  }
}

以上代码运行后能正常显示结果，但如果你切换到开发者工具的网络面板，你会发现发送了三个重复的请求。这是因为我们的 Observable 是 cold 的，每处使用 async 管道的地方都会执行一次。针对上述问题，大部分人推荐的解决方案如下：

this.http.get('https://jsonplaceholder.typicode.com/posts/1')
     .map(res => res.json()).share()

我们只需使用 RxJS 中的共享操作符，它在内部调用 publish().refCount()。是不是很开心，但先等等，还有一种情况又会触发 HTTP 请求，具体我们再来看一下示例：

import { Component, OnInit } from '@angular/core';
import { Http } from '@angular/http';
import { Observable } from 'rxjs/Rx';

@Component({
  selector: 'exe-app',
  template: `
   <div>
      <p>{{ (person$ | async)?.id }}</p>
      <p>{{ (person$ | async)?.title }}</p>
      <p>{{ (person$ | async)?.body }}</p>
      <button (click)="showPersonInfo()">显示用户ID</button>
      <p *ngIf="isShow">{{ (person$ | async)?.id }}</p>
    </div>
  `
})
export class AppComponent implements OnInit {
  person$: Observable<{ id: number; title: string; body: string }>;
  isShow: boolean = false;

  constructor(private http: Http) { }

  ngOnInit() {
    this.person$ = this.http.get('https://jsonplaceholder.typicode.com/posts/1')
      .map(res => res.json())
      .share();
  }

  showPersonInfo() {
    this.isShow = true;
  }
}

以上代码运行后浏览器显示的结果：

我们发现当点击 '显示用户ID' 按钮时，又触发了新的请求。看来我们的救世主 - share() 不给力了，幸运的是我们还有新的救世主 - publishReplay() ，具体代码如下：

import { Component, OnInit } from '@angular/core';
import { Http } from '@angular/http';
import { Observable, ConnectableObservable } from 'rxjs/Rx';

@Component({
  selector: 'exe-app',
  template: `
   <div>
      <p>{{ (person$ | async)?.id }}</p>
      <p>{{ (person$ | async)?.title }}</p>
      <p>{{ (person$ | async)?.body }}</p>
      <button (click)="showPersonInfo()">显示用户ID</button>
      <p *ngIf="isShow">{{ (person$ | async)?.id }}</p>
    </div>
  `
})
export class AppComponent implements OnInit {
  person$: ConnectableObservable<{ id: number; title: string; body: string }>;
  isShow: boolean = false;

  constructor(private http: Http) {
    this.preparePersonInfo();
  }

  ngOnInit() {
    this.person$.connect();
  }

  preparePersonInfo() {
    this.person$ = this.http.get('https://jsonplaceholder.typicode.com/posts/1')
      .map(res => res.json())
      .publishReplay()
  }

  showPersonInfo() {
    this.isShow = true;
  }
}

我们使用 publishReplay 替换了 share 操作符。调用 publishReplay() 方法后将返回一个ConnectableObservable 对象，当你调用 connect() 方法的时候，将主动执行订阅操作。

是不是感觉快奔溃了，就想简单的获取用户的信息，然后使用 AsyncPipe，怎么就那么多坑。。。

在大多数情况下，我们只需要从服务器获取数据并显示数据。如果只是这样的话，我们可以使用 Promise 来修复 AsyncPipe 发送多次 HTTP 请求的问题：

this.person = this.http.get("https://jsonplaceholder.typicode.com/posts/1")
 .map(res => res.json()).toPromise()

3.AsyncPipe 执行流程

接下来我们看一下 PromiseStrategy 与 ObservableStrategy 策略：

SubscriptionStrategy 接口

interface SubscriptionStrategy {
  createSubscription(async: any, updateLatestValue: any): any;
  dispose(subscription: any): void;
  onDestroy(subscription: any): void;
}

PromiseStrategy

class PromiseStrategy implements SubscriptionStrategy {
  createSubscription(async: Promise<any>, 
    updateLatestValue: (v: any) => any): any {
      return async.then(updateLatestValue, e => { throw e; });
  }
  dispose(subscription: any): void {}
  onDestroy(subscription: any): void {}
}

ObservableStrategy

class ObservableStrategy implements SubscriptionStrategy {
  createSubscription(async: any, updateLatestValue: any): any {
    return async.subscribe({next: updateLatestValue, 
       error: (e: any) => { throw e; }});
  }
  dispose(subscription: any): void { subscription.unsubscribe(); }  // 取消订阅
  onDestroy(subscription: any): void { subscription.unsubscribe(); } 
}

总结

这篇文章我们介绍了 AsyncPipe 如何搭配 Promise 和 Observable 使用，此外介绍了 AsyncPipe 会发送多次请求的问题，最后我们得出的结论是如果只是需要从服务器获取数据并显示数据，可以使用 Promise 来修 AsyncPipe 发送多次 HTTP 请求的问题。

参考文章

• Stop using observable when you should use a promise

• Code VS Hot Observables