如何在自定义RxJS操作符中使用Angular服务？

当然可以实现这个需求。由于Angular服务依赖注入的特性，我们需要通过合理的方式将服务传递给自定义操作符，以下是两种常用的实现方案：

方案一：将服务作为参数传入操作符 #

直接修改自定义操作符，让它接受一个服务实例作为参数，就能在操作符内部调用服务的方法：

import { Observable } from 'rxjs';

// 自定义操作符，接收Angular服务作为参数
function myOperator(myService: MyAngularService) {
  return function <T>(source: Observable<T>): Observable<T> {
    return new Observable(subscriber => {
      const subscription = source.subscribe({
        next(value) {
          // 在这里直接使用传入的服务
          myService.handleValue(value);
          subscriber.next(value);
        },
        error(error) {
          subscriber.error(error);
        },
        complete() {
          subscriber.complete();
        }
      });
      // 清理订阅，避免内存泄漏
      return () => subscription.unsubscribe();
    });
  };
}

在组件中使用时，直接将注入的服务实例传递给操作符：

import { Component } from '@angular/core';
import { MyAngularService } from './my-angular.service';

@Component({
  selector: 'app-my-component',
  templateUrl: './my-component.component.html'
})
export class MyComponent {
  constructor(private myService: MyAngularService) {}

  processObservable() {
    someObservable$
      .pipe(
        myOperator(this.myService) // 传入服务实例
      )
      .subscribe(result => {
        // 处理订阅结果
      });
  }
}

这种方式简单直接，适合操作符仅在少数场景使用的情况。

方案二：将操作符封装在Angular服务中 #

如果操作符需要在多个组件复用，或依赖多个服务，更优雅的方式是将操作符封装在Angular服务里，利用服务的依赖注入能力直接获取所需服务：

import { Injectable } from '@angular/core';
import { Observable, MonoTypeOperatorFunction } from 'rxjs';
import { MyAngularService } from './my-angular.service';

@Injectable({ providedIn: 'root' })
export class MyOperatorService {
  // 直接注入需要使用的服务
  constructor(private myService: MyAngularService) {}

  // 定义操作符方法
  myOperator<T>(): MonoTypeOperatorFunction<T> {
    return (source: Observable<T>) => {
      return new Observable(subscriber => {
        const subscription = source.subscribe({
          next(value) {
            // 使用注入的服务
            this.myService.handleValue(value);
            subscriber.next(value);
          },
          error(error) {
            subscriber.error(error);
          },
          complete() {
            subscriber.complete();
          }
        });
        return () => subscription.unsubscribe();
      });
    };
  }
}

在组件中使用时，注入封装了操作符的服务并调用其方法即可：

import { Component } from '@angular/core';
import { MyOperatorService } from './my-operator.service';

@Component({
  selector: 'app-my-component',
  templateUrl: './my-component.component.html'
})
export class MyComponent {
  constructor(private operatorService: MyOperatorService) {}

  processObservable() {
    someObservable$
      .pipe(
        this.operatorService.myOperator() // 调用服务中的操作符方法
      )
      .subscribe(result => {
        // 处理订阅结果
      });
  }
}

注意事项 #

• 如果操作符需要调用服务的异步方法（比如返回Observable的HTTP请求），建议使用RxJS内置操作符（如switchMap、concatMap）组合逻辑，避免手动创建Observable时的订阅管理问题：

function myOperator(myService: MyAngularService) {
  return function <T>(source: Observable<T>): Observable<any> {
    return source.pipe(
      switchMap(value => {
        // 调用服务的异步方法，处理后返回结果
        return myService.processValueAsync(value);
      })
    );
  };
}

• 始终确保在操作符的teardown逻辑中清理订阅，避免内存泄漏。

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