RxJS：如何在Observable开始执行时绑定处理函数？

当然有办法实现你想要的效果！你提到的这种“在Observable被调用subscribe时立即触发处理逻辑”的需求，RxJS本身没有内置的executing操作符，但我们可以通过两种方式轻松搞定，完美适配你Angular中HTTP请求的场景：

方法一：自定义executing操作符 #

我们可以自己实现一个符合你预期的executing操作符，让它在Observable被订阅时立刻执行传入的回调函数。这样就能完全按照你示例里的写法来使用：

首先定义这个自定义操作符：

import { Observable } from 'rxjs';

/**
 * 当Observable被订阅时执行回调函数的操作符
 */
export function executing<T>(callback: () => void): (source: Observable<T>) => Observable<T> {
  return (source) => {
    return new Observable<T>(subscriber => {
      // 订阅发生的瞬间执行传入的回调
      callback();
      // 订阅源Observable，传递所有通知给下游
      const subscription = source.subscribe(subscriber);
      // 返回清理函数，确保订阅能正常取消
      return () => subscription.unsubscribe();
    });
  };
}

然后你就可以像示例里那样直接使用了：

this.http.post('someUrl', resource).pipe(
  catchError(error => {
    // 这里处理你的错误逻辑
    return throwError(() => new Error('请求失败'));
  }),
  executing(() => this.showLoader()), // 订阅时立刻显示加载器
  finalize(() => this.hideLoader())   // 请求完成/失败后隐藏加载器
).subscribe({
  next: (response) => {
    // 处理响应数据
  },
  error: (err) => {
    // 处理错误提示
  }
});

方法二：利用defer操作符组合实现 #

如果你不想自定义操作符，也可以用RxJS内置的defer操作符来达到同样的效果。defer会延迟创建Observable，直到它被订阅的那一刻，刚好可以在这里插入你的初始化逻辑：

import { defer } from 'rxjs';

// 用defer包裹请求逻辑，订阅时先执行showLoader，再发起请求
defer(() => {
  this.showLoader();
  return this.http.post('someUrl', resource);
}).pipe(
  catchError(error => {
    return throwError(() => new Error('请求失败'));
  }),
  finalize(() => this.hideLoader())
).subscribe(...);

注意事项 #

• 这两种方式都针对Cold Observable（比如Angular的HTTP请求就是典型的Cold Observable），每次订阅都会重新触发执行逻辑，符合“开始执行时触发”的需求。
• 如果是Hot Observable（比如通过subject.asObservable()创建的），订阅不会触发源Observable的重新执行，这时候你可能需要结合其他逻辑处理，但你的HTTP场景完全不需要担心这个问题。

内容的提问来源于stack exchange，提问作者Lerner