如何为Observable编程式添加takeUntil？AutoUnsubscribe装饰器失效排查

问题根源 #

你的装饰器未生效核心原因有三点：

• unsubscribe$ 作用域错误：装饰器内的unsubscribe$是装饰器函数的局部变量，所有组件实例会共用同一个Subject；同时组件实例上不存在this.unsubscribe$属性，takeUntil(this.unsubscribe$)相当于传入undefined，完全不起作用。
• 未捕获直接订阅的服务流：AppComponent中直接订阅了服务的test$流，但该订阅未保存到组件的任何属性中，装饰器遍历组件实例属性时无法找到它，自然无法添加takeUntil逻辑。
• Observable管道未替换原变量：即便找到组件的Observable属性，property.pipe(takeUntil(...))仅创建了新的Observable，未替换原属性，后续订阅仍使用原始流。

修复后的装饰器实现 #

以下是修正后的装饰器代码，解决了上述所有问题：

import { Observable, Subject, takeUntil, isObservable } from 'rxjs';

export function AutoUnsubscribe(skipList: string[] = []) {
  return function (constructor: any) {
    const originalNgOnDestroy = constructor.prototype.ngOnDestroy;
    const originalNgOnInit = constructor.prototype.ngOnInit;

    // 给每个组件实例添加专属的取消订阅Subject，并重写ngOnDestroy
    constructor.prototype.ngOnDestroy = function (...args: any[]) {
      if (this.__unsubscribe$) {
        this.__unsubscribe$.next();
        this.__unsubscribe$.complete();
      }
      if (originalNgOnDestroy && typeof originalNgOnDestroy === 'function') {
        originalNgOnDestroy.apply(this, args);
      }
    };

    // 初始化时处理组件内的Observable属性
    constructor.prototype.ngOnInit = function (...args: any[]) {
      this.__unsubscribe$ = new Subject<void>();

      // 遍历组件实例属性，替换Observable为带takeUntil的版本
      for (const prop in this) {
        if (skipList.includes(prop)) continue;
        
        const property = this[prop];
        if (property && isObservable(property)) {
          this[prop] = property.pipe(takeUntil(this.__unsubscribe$));
        }
      }

      if (originalNgOnInit && typeof originalNgOnInit === 'function') {
        originalNgOnInit.apply(this, args);
      }
    };
  };
}

正确使用方式 #

方式1：将服务流赋值给组件属性后订阅 #

修改AppComponent，把服务的Observable赋值给组件自身属性，让装饰器能捕捉并处理：

@AutoUnsubscribe()
@Component({
  selector: 'app-root',
  templateUrl: './app.component.html',
  styleUrls: ['./app.component.scss']
})
export class AppComponent {
  // 将服务流绑定到组件属性
  test$!: Observable<any>;

  constructor(private testClass: TestClass, private router: Router) {
    this.test$ = this.testClass.test$;
    // 订阅组件自身的test$，此时已被装饰器添加takeUntil逻辑
    this.test$.subscribe(v => {
      console.log('App component', v);
    });

    this.testClass.test$.next(2);

    setTimeout(() => {
      this.router.navigate(['/teste']);
    }, 4000);
  }
}

方式2：添加组件专属订阅方法（可选） #

如果不想修改现有订阅写法，可在装饰器中给组件添加统一的订阅方法：

// 在装饰器的constructor.prototype.ngOnInit之后添加：
constructor.prototype.autoSubscribe = function (observable: Observable<any>, ...args: any[]) {
  return observable.pipe(takeUntil(this.__unsubscribe$)).subscribe(...args);
};

组件内使用该方法订阅：

this.autoSubscribe(this.testClass.test$, v => {
  console.log('App component', v);
});

验证效果 #

修复后，当AppComponent销毁（跳转到TestpageComponent）时，其订阅会被自动取消。TestpageComponent触发test$.next(3)时，只会打印Testpage component 3，不会再出现App component 3。

内容的提问来源于stack exchange，提问作者Lucas Buchalla Sesti