从命令式到响应式（九）

使用rxjs时何时取消订阅是咱们必需要关心的，这个系列的前面几篇也提到过，原则是尽可能不去手动订阅流，但手动订阅终究是没法避免的，今天主要总结下如何适时的取消订阅。

让angular帮咱们取消

这个不用多说了，主要是采用 async pipe，在HTML模版本中让angular自已去取数据，再自动的取消。

在component中管理subscription

思路是，在component中订阅的流，在合适的生命周期中取消，最多见就是在OnDestroy的时候。两种方式，第一种组件维护一个subscription，其它的subscription经过add方法添加至这个subscription上，取消的时候调用这个subscription的unsubscribe方法，第二种，组件维护一个subscription数据，取消的时候遍历数组，调用每个subscription的unsubscribe方法。

假设咱们有一个服务，能够发送websocket的请求，处理请求的错误，同时还能够提供一些公共逻辑的处理，大概像下面这样：

// service.ts
@Injectable()
export class MyService {
    constructor(public websocket: WebsocketService) {}

    // 发送websocket请求
    request(paramObs: Observable<MyInterface>): Subscription {
        return paramObs.subscribe(params => this.websocket.send(params));
    }

    // 处理响应的错误
    handleError(): Subscribe {
        return this.websocket.message.pipe(
            filter(res => res.flag === 'request flag') // 取这个上面请求的结果
        ).subscribe(res => ...)
    }

    // 其它须要在服务中订阅后处理的逻辑
    otherLogic(params: Observable<any>): Subscription {
        return params.subscribe(...) 
    }
}

第一种思路：

@Component({...})
export class MyComponent implement OnInit, OnDestroy {
    subscription: Subscription;

    constructor(private ser: MyService) { }

    ngOnInit() {
        this.subscription = this.ser.request(paramsObs1) // 参数是包含请求数据的流
            .add(this.otherLogic(paramsObs2)) // 参数是须要处理的数据流
            .add(this.ser.handleError())
    }

    ngOnDestroy() {
        this.subscription.unsubscribe();
    }
}

第二种思路：

@Component({...})
export class MyComponent implement OnInit, OnDestroy {
    subscriptions: Subscription[] = [];

    constructor(private ser: MyService) { }

    ngOnInit() {
        this.subscriptions = [
            this.ser.request(paramObs1), // 参数是包含请求数据的流
            this.ser.handleError(),
            this.otherLogic(paramsObs2) // 参数是须要处理的数据流
        ];
    }

    ngOnDestroy() {
        this.subscriptions.forEach(sub => sub.unsubscribe());
    }
}

除了写法上的不一样外，最大的不一样在于采用第一种写法时，你可能须要注意添加的顺序，假如例子中的paramsObs2参数流会出完成通知则会致使handleError也被取消掉，这种场景你们能够本身写个demo试下，第二种写法则不会，但可能重复去取消一些已经被取消过的流，好在这并不会致使错误的发生。

使用rxjs的操做符管理订阅

思路是使用那些可让流发出结束通知的操做符，将其添加到须要管理的流上，让rxjs自动取消订阅。经常使用的有下面这些：

take操做符

此操做符会上当前流上取指定数量的值，而后发出完成通知，使用只想让otherLogic处理前3个数据，

ngOnInit() {
    this.ser.otherLogic(paramsObs2.take(3)); // 再也不须要理会订阅产生的subscription了，处理3个值后自动取消订阅；
}

相似的操做符还有first，from，of等都会发出完成通知。

takeWhile操做符

此操做符添加到流上后，每一次发出值时都要检查操做符中传入的断定函数返回的结果是否为true，一旦返回false，输出流将会被取消掉，再也不发出值，假设咱们的paramsObs2的数据来自于一个formControl：

@Component({
    ...
    template: `<input type="text" [formControl]="control">`
})
export class MyComponent implement OnInit, OnDestroy {
    control = new FormControl('xxx');

    isAlive = true; // 添加一个变量控制流的结束

    ...

    ngOnInit() {
        ...
        this.ser.otherLogic(this.control.valueChanges.pipe(
            takeWhile(() => this.isAlive) // 传入了一个断定函数
        ))
    }

    ngOnDestroy() {
        this.isAlive = false; // 这里变为false；
    }
}

takeUntil操做符

此操做符和takeWhile不一样，第一，接受的参数不是断定函数，而是Observable， 第二，传入的observable一旦发出值，输入流将会被取消订阅，无论发出的值是true仍是fasle，或者是其它值。

@Component({
    ...
    template: `<input type="text" [formControl]="control">`
})
export class MyComponent implement OnInit, OnDestroy {
    control = new FormControl('xxx');

    constructor(
        ...
        private router: Router
    ){}

    ngOnInit() {
        ...
        this.ser.otherLogic(this.control.valueChanges.pipe(
            takeWhile(this.router.events) // 把router的事件流传了进去，只要router上有事件发出输出流就被取消
        ))
    }

    ...
}

这几种方法各有各有的特色，有的须要额外的变量但用起来简单粗爆，有的简洁明了但你须要花心思在其它条件上，在项目中能够根据实际状况选择最适合的使用。