精选

多年以来，RxJS 一直是 Angular 中响应式编程的基石。然而，在数据和视图同步方面，它的主要缺点在于其 无状态性。从概念上讲，Observables 更侧重于处理事件，而不太适合管理数据。对此，Angular 团队引入了一个新概念——信号。那么，我们现在应该如何看待 RxJS 呢？我们又应该如何使用它呢？

如前所述，可观察对象的核心概念在于通知。流及其操作符在描述复杂的事件驱动逻辑方面表现出色，特别是在处理浏览器环境中的命令式API时尤为突出。随着Angular现在推出工具来实现信号与可观察对象之间的无缝交互，我们因此获得了极其强大的协同效应。

本文将展示一些代码示例来说明这两个概念如何相辅相成。

模板中不再有可观察对象。再见 | 异步 👋

在引入信号之前，Angular 的响应性之前依赖于在模板中使用 AsyncPipe 来订阅可观察对象。此管道主要解决两个关键问题：标记视图脏 以进行变更检测（当使用 OnPush 策略时是必须的），并在视图销毁时从可观察对象取消订阅，以确保与组件生命周期的绑定。

然而，这并没有解决我之前提到的根本问题：可观察对象在订阅时并不能保证状态，因此，Angular 将管道返回值类型定义为可能包含 null 的联合类型。这迫使开发者添加额外的检查步骤，结果证明这相当让人头疼。

现在，是时候完全放弃在模板中使用可观察对象了，转而使用信号了，信号本质上是有状态的，并带来更多我们将在示例中讨论的好处。

⚠️ 警告: 代码示例使用的是RxJS 互操作 API，该API仍处于开发者预览版。

让我们完成一个简单的任务：当点击复制按钮时，我们希望在几秒钟内更改图标和文本来给用户反馈，然后恢复到原始状态。

readonly copied = toSignal(
  fromEvent(inject(ElementRef).nativeElement, 'click').pipe(
    exhaustMap(() => 
      timer(2000).pipe(
        map(() => false),
        startWith(true)
      )
    )
  ),
  { initialValue: false }
);
/* 当元素被点击时，触发一个延迟2秒的信号，初始值为true，之后变为false。 */

结果如下 (stackblitz):

借助 RxJS 操作符，我们描述了 copied 状态是如何通过声明式的方式发生变化的。

• 如果用户多次点击该按钮，exhaustMap 确保当前的 timer 先完成后再置为 false
• 通过 toSignal 处理同步发出的值和信号，并从源可观察对象取消订阅的逻辑，无需手动进行订阅

如前一节所述，不要使用AsyncPipe，我们可以用新的替代方案toSignal来处理流。

让我们开始一个小挑战。我们的页面包含锚链接，当用户导航到具有相应片段标识符的 URL 时，这些链接应该被突出显示。由于包含链接的部分可能在导航时超出视口，高亮显示应该在自动滚动完成后应用。视口是指当前显示区域。

    private readonly id = 注入(new HostAttributeToken('id')); // 注入ID标识符
    private readonly route = 注入(ActivatedRoute); // 注入激活的路由

    readonly highlighted = toSignal(
      this.route.fragment.pipe(
        startWith(this.route.snapshot.fragment), // 从快照中的fragment开始
        filter(fragment => this.id === fragment), // 过滤fragment，当其等于id时
        switchMap(() => 
          concat(
            fromEvent(window, 'scroll').pipe(
              startWith(true), // 从true开始
              debounceTime(100), // 延迟100毫秒
              take(1) // 取第一个事件
            ),
            timer(2000).pipe(map(() => false)) // 2秒后返回false
          )
        )
      ),
      { initialValue: false } // 初始值为false
    );

除此之外，管理状态时，另一个使用可观察对象的缺点是无法直接将数据绑定到宿主节点。信号解决了这个问题，让我们能够做到这一点：

    host: {  
      '[class.highlighted]': 'highlighted()'  // 高亮显示的函数
    }

这段代码定义了host对象中[class.highlighted]的值为"highlighted()"，用于高亮显示相关元素。请根据需要正确使用此代码片段。

结果如下（stackblitz）：

RxJS 再次帮助我们避免了大量命令式代码：

• debounceTime 结合 take 可以让我们在自动滚动完成后将状态更新为 true，这里的“在……后”可以更自然地表达为“在……之后”。
• timer 将状态重置为初始值
• concat 管理订阅顺序，确保在第一个可观察对象完成后才订阅重置定时器

让我们来实现一个具有自动关闭功能的通知组件。它接受输入显示的时间长度，并在应该关闭时发出一个关闭事件。此外，还有一个额外的要求：如果用户将鼠标悬停在通知上，计时器应暂停，直到鼠标离开通知，计时器才会重新开始，从而防止在此期间通知的自动关闭。

    private readonly el = inject(ElementRef).nativeElement;  

    readonly duration = input(Infinity);  

    readonly close = outputFromObservable(  
      toObservable(this.duration).pipe(  
        switchMap(value => Number.isFinite(value) ? timer(value) : EMPTY),  
        takeUntil(fromEvent(this.el, 'mouseenter')),  
        repeat({ delay: () => fromEvent(this.el, 'mouseleave') })  
      )  
    );

结果如下（stackblitz，这是一个在线代码编辑器链接）：

在这项任务中，我们用了两个辅助函数来处理 Angular 更新的 input 和 output 接口。

toObservable：将我们的基于信号的输入转换成一个流。Angular 在增强其反应式系统方面做得非常出色，使指令和组件的 输入属性 更具反应性。这些属性现在可以用于状态管理场景（例如，与 计算属性 一起使用），并且还可以用于根据与可观察对象的交互来创建逻辑。

⚠️ 需要注意的是，信号本质上是无干扰的 并且永远不会同步传递变化。如果一个信号的值被同步更新多次，订阅者只会收到一次信号在变化检测过程中最后一次更新的值的通知。如果你计划使用toObservable 构建反应式链，这点很重要。

outputFromObservable：从一个可观察对象（observable）创建一个新的输出流。随着新输入选项的引入，Angular 从之前的基于 RxJS 的 EventEmitter 扩展中移开。虽然这里没有什么突破性的变化，但输出现在在后台使用了一个新的类 OutputEmitterRef 的实例。这个类本质上与 Subject 模型相似，但形式更轻盈。

谈到浏览器环境中的APIs，用声明式的方式用它们可能不太方便。大多数这些API都使用回调，比如其中一些还包括清理逻辑，类似于RxJS中所说的拆卸逻辑。

但是，虽然 RxJS 提供了一种构建反应式链的统一方式，但在使用命令式 API 时常常需要编写更冗长且一致的指令。

由于信号量可以无缝地与可观察对象集成，因此，可以轻松地将命令式API视为流。例如，让我们定义一个自定义运算符，将ResizeObserver 转换成可观察对象：

export function fromResizeObserver(
  target: Element,
  options?: ResizeObserverOptions,
): Observable<ResizeObserverEntry[]> {
  return new Observable(subscriber => {
    const ro = new ResizeObserver((entries) => {
      subscriber.next(entries);
    });
    ro.observe(target, options);
    return () => {
      ro.disconnect();
    };
  });
}

💡 此运算符可以在整个代码库中重复使用。例如，我们可以用它为宿主元素的 width 创建一个响应式状态的。

    private readonly el = inject(ElementRef).nativeElement;  

    readonly width = toSignal(  
      fromResizeObserver(this.el).pipe(map(() => this.el.offsetWidth)),  
      { initialValue: 0 }  
    );

准确跟踪一个元素的宽度通常不仅仅需要ResizeObserver。在某些特殊情况下，可能还需要用到像MutationObserver这样的观察器。这就是RxJS大放异彩的地方：通过利用合并操作符，我们可以声明式地组合事件，从而可靠地检测元素宽度的变化：

    private readonly el = inject(ElementRef).nativeElement;  

    readonly width = toSignal(  
      merge(  
        fromResizeObserver(this.el),  
        fromMutationObserver(this.el, {  
          childList: true,  
          subtree: true,  
          characterData: true,  
        })  
        // ...以及其他可能的特殊情况  
      ).pipe(  
        map(() => this.el.offsetWidth),  
        distinctUntilChanged()  
      ),  
      { initialValue: 0 }  
    );

在之前的例子中，我们创建了一个自定义操作来封装特定于浏览器的API。像这样的方法也可以用于基于Angular的某些API创建可观察对象，从而实现反应式交互。

在之前的某些代码片段中，我使用了仅在浏览器环境中可用的全局对象（例如，window）。然而，为了确保代码在服务器端（SSG/SSR）正常运行，我们需要确保某些逻辑仅在客户端渲染时才初始化。在这种情况下，我们可以使用基于回调的 afterNextRender API（⚠️ 还在开发者预览阶段），并将其转换为可观察对象。

export function fromAfterNextRender(options?: AfterRenderOptions): Observable<void> {  
  !options?.injector && assertInInjectionContext(fromAfterNextRender);  

  return new Observable(subscriber => {  
    const ref = afterNextRender(() => {  
      subscriber.next();  
      subscriber.complete();  
    }, { ...options, manualCleanup: true });  

    return () => ref.destroy();  
  });  
}

然后，我们可以在客户端仅计算状态值：

    readonly state = toSignal(  
      fromAfterNextRender().pipe(  
        switchMap(() => {  
          // 这里是在浏览器环境中  
        }),  
      ),  
    );

💡 另一个例子是，你可以基于频繁的操作序列创建自定义操作，以简化代码并提高可读性。

在我的一个项目里，有很多应用逻辑与Router事件相关，所以我没有这样写代码，而是这样写：

    private readonly router = inject(Router);  

    constructor() {  
      this.router.events.subscribe(e => {  
        if (e instanceof NavigationEnd) {  
          // 注意: 类型细化仅在此代码块内有效  
        }  
      })  
    }

这样写方便多了：

    /*
      现在，我们可以任意使用这个可观察对象来处理对应事件的类型化值 ✨

      返回 Observable<NavigationEnd>
    */
    fromRouterEvent(NavigationEnd);

要做到这一点，只需要使用现有的过滤操作和一个类型谓词即可，这能确保类型细化。（类型谓词，如这里所示）。

    import { Event, Router } from '@angular/router';  

    /**

* 从路由器事件生成 Observable 对象

* @param event 事件类型

* @param options 可选参数，包含注入器

* @returns 返回一个 Observable 对象
     */
    export function fromRouterEvent<T extends Event>(  
      event: { new (...args: any[]): T },  
      options?: { injector: Injector },  
    ): Observable<T> {  
      let router: Router;  

      if (!options?.injector) {  
        // 确保当前环境支持注入
        assertInInjectionContext(fromRouterEvent);  
        router = inject(Router);  
      } else {  
        router = options.injector.get(Router);  
      }  

      // 过滤出指定类型的事件
      return router.events.pipe(filter((e: unknown): e is T => e instanceof event));  
    }

换句话说，这使我们能够使框架所需的API和常用操作能够与流一起工作，从而大大减少样板代码，使其更加符合声明式风格。

下面提到的例子是来自实际项目的实用任务。如这些解决方案所展示的，RxJS 完全适合应对事件驱动的任务。此外，反应式库依然无缝集成在 Angular 生态系统中，通过公开 API（例如在 @angular/{router,forms,сommon/http,cdk} 中）。

重要的是：信号和可观察对象之间的协同作用将有助于Angular框架响应性的重大进展，提供了一种更加结构化的处理方式，其中信号主要负责状态管理，而可观察对象则主要用于处理事件。

敬请期待下一篇文章，我们将探讨更多更高级的主题，例如如何利用Signals优化性能以及如何使用RxJS处理复杂用例！