Angular 中的 asyncPipe 源码探究

它是最优雅的订阅observable 的方式，不仅语法短小精悍，还会自动取消订阅；

async 管道用于解包异步原始数据。说到异步数据，就自然而然的会想起 observable 和 promise， async 就是用来订阅他们，然后返回他们发布的最近一个值，然后将组件标记为变更做准备。当组件被销毁时，async 会自动取消订阅以防内存泄漏。

来看看它的源码中的具体实现：

    @Pipe({name: 'async', pure: false})
    export class AsyncPipe implements OnDestroy, PipeTransform {
      //  是在视图引擎上渲染的数据，通常称为展示数据
      private _latestValue: any = null;
      private _latestReturnedValue: any = null;
    
      private _subscription: SubscriptionLike|Promise<any>|null = null;
    	
    	// this.obj 是我们保存在管道中的observable 的状态
      private _obj: Observable<any>|Promise<any>;
      private _strategy: SubscriptionStrategy = null !;
    
      constructor(private _ref: ChangeDetectorRef) {}
    
      ngOnDestroy(): void {
        if (this._subscription) {
          this._dispose();
        }
      }
    	
    	// obj 是我们传入的异步数据源
      transform(obj: Observable<any>|Promise<any>): any {
        if (!this._obj) {
          if (obj) {
    				console.log('Async: start subscription');
            this._subscribe(obj);
          }
    			console.log('Async: initial value = ' + this.lastestValue);
          this._latestReturnedValue = this._latestValue;
          return this._latestValue;
        }
    
    		// 保证是同一个异步数据源，当不是同一个observable，销毁上一次的订阅，订阅这次的异步数据
        if (obj !== this._obj) {
          this._dispose();
          return this.transform(obj as any);
        }
      }
    
      private _subscribe(obj: Observable<any>|Promise<any>|EventEmitter<any>): void {
        this._obj = obj;
        this._subscription = this._strategy.createSubscription(
            obj, (value: Object) => this._updateLatestValue(obj, value));
      }
    
      private _dispose(): void {
        this._strategy.dispose(this._subscription !);
        this._latestValue = null;
        this._latestReturnedValue = null;
        this._subscription = null;
        this._obj = null;
      }
    
      private _updateLatestValue(async: any, value: Object): void {
        if (async === this._obj) {
          this._latestValue = value;
          this._ref.markForCheck();
        }
      }
    }

从源码中可以看出，当一个 async pipe 被创建时，它会在每一次的变更检测循环中调用 transform()；

在transform()中，会先判断管道中是否已有数据源，如果没有，则会创建一个对于该异步数据源的订阅，这其中包括将数据源赋给管道中存储的数据源，同时改值会在 dom 上被渲染，并且视图标记为变更待检测。

如果已有数据，就会判断这次的数据源和上次的数据源是不是同一个，如果不是同一个，则销毁上一次的订阅，订阅此次的异步数据；

这样我们可以重复使用 async 管道里的最新值并维护单个订阅，这有益于应用的性能和可维护性。

async pipe 的重置和完成

重置：如果异步数据源发生变更，async 管道会自动取消订阅，做好清理工作。然后订阅新的 observable。源码如下：

 if (obj !== this._obj) {
      this._dispose();
      return this.transform(obj as any);
    }

完成：如果 observable 完成，停止发布，最新的值会被在 dom上渲染；但是 async 管道在每次变更检测周期还是会触发 transform 直到它被销毁(组件销毁，dom 销毁)；这里会产生多余的触发，引起性能问题；

优化：使用 onPush 进行优化

异步管道在每个变更检测周期调用 transform()，因为它是一个不纯的管道。对于每次调用转换，async 管道都会执行一次检查，以确定最新的值是否发生了变更，如果没有，则提前返回；看源码：

if (ɵlooseIdentical(this._latestValue, this._latestReturnedValue)) {
      return this._latestReturnedValue;
}

我们可以避免这些冗余的检测。所以 async 管道用 ChangeDectionStrategey.onPush 的父组件。当组件使用这个策略时，仅仅会在被标为变更时才会调用 transfrom；

当一个值被 async 管道订阅的 observable 发布时，它会通过 markForCheck() 将视图标记为变更待检测。这意味着只有当obervable 发布值时，他才会告诉组件需要重新渲染了。源码如下：

private _subscribe(obj: Observable<any>|Promise<any>|EventEmitter<any>): void {
    this._obj = obj;
    this._strategy = this._selectStrategy(obj);
    this._subscription = this._strategy.createSubscription(
        obj, (value: Object) => this._updateLatestValue(obj, value));
}

private _updateLatestValue(async: any, value: Object): void {
    if (async === this._obj) {
      this._latestValue = value;
      this._ref.markForCheck();
    }
}

注意：尽管 on-push 可以提高组件的性能，但必须谨慎的使用它，并且只在适用时使用它，不正确的使用 on-push 可能会导致视图未按预期更新的bug;

总结

使用一个异步管道避免冗余订阅；自动取消订阅，可以避免产生内存泄漏；这个也可以使用 rxjs 中的 takeUntil，它也会取消订阅；
使用 rxjs 中的 combineLatest 将数据组合进一个 async 管道中；
只有在应用有需要的时候使用 async 管道和 on-push 策略，并且要谨慎使用；

链接: https://fly63.com/article/detial/8254