问题

我有一个可观察的流水线，当订阅时调用一个API。我正在使用distinctUntilChanged操作符，只有在当前HTTP响应与前一个HTTP响应不同时才发出下一个通知。尽管API响应相同，但distinctUntilChanged操作符会发出下一个通知。
我正在使用lodash中的isEqual函数来比较对象，我看到它返回true。

这里是代码片段：

@Injectable({
  providedIn: 'root',
})
export class CustomerFacadeService {
  private eventBus$ = new BehaviorSubject<void>(undefined);
  private cachedCartData$ = this.http.get(
    'https://62c41dc97d83a75e39f0d512.mockapi.io/api/v1/cart'
  );

  cart$ = this.eventBus$.pipe(
    mergeMap(() => this.cachedCartData$),
    distinctUntilChanged((prev, current) => {
      const result = isEqual(prev, current);
      console.log('result', result);
      return result;
    }),
  );

  constructor(private http: HttpClient) {}

  cacheBust() {
    this.eventBus$.next();
  }
}

屏幕截图:

仓库:
https://github.com/phalgunv/caching-data-example

cart$ 可观察对象通过 async 管道在2个不同的组件中订阅。当我点击"Update"按钮时，会发出下一个通知，导致调用HTTP GET API。我期望distinctUntilChanged 不会发出下一个通知，因为数据没有改变。尽管对象引用可能已更改，但由于我使用isEqual函数。

@Injectable({
  providedIn: 'root',
})
export class CustomerFacadeService {
  private eventBus$ = new BehaviorSubject<void>(undefined);
  private cachedCartData$ = this.http.get(
    'https://62c41dc97d83a75e39f0d512.mockapi.io/api/v1/cart'
  );

  cart$ = this.eventBus$.pipe(
    mergeMap(() => this.cachedCartData$),
    distinctUntilChanged((prev, current) => {
      const result = isEqual(prev, current);
      console.log('result', result);
      return result;
    }),
  );

  constructor(private http: HttpClient) {}

  cacheBust() {
    this.eventBus$.next();
  }

答案1

得分: 1

cart$ observable subscribed in 2 different components via async pipe

这导致了两个不同组件中通过异步管道订阅的 cart$ 可观察对象。这会导致对冷可观察对象的两次订阅。这意味着你将为两个订阅者获得两个生产者。因此，每个订阅都会获得一个单独的 distinctUntilChanged()。如果你想在多个消费者之间共享单个生产者，这就是你可以使用 share() 将冷可观察对象转换为热可观察对象的地方。

请记住，本质上是在单个生产者和多个订阅者之间放置了一个 Subject 作为中间人。关于 Subject 何时实际订阅生产者以及如何管理其生命周期，有一些详细信息。在这里，你的两个组件都在构建时订阅，并且不应错过通过 share() 传递的任何事件。如果需要延迟订阅者的缓存，你可以查看 shareReplay()。

@Injectable({
  providedIn: 'root',
})
export class CustomerFacadeService {
  private eventBus$ = new BehaviorSubject<void>(undefined);
  private cachedCartData$ = this.http.get(
    'https://62c41dc97d83a75e39f0d512.mockapi.io/api/v1/cart'
  );

  cart$ = this.eventBus$.pipe(
    mergeMap(() => this.cachedCartData$),
    distinctUntilChanged((prev, current) => {
      const result = isEqual(prev, current);
      console.log('result', result);
      return result;
    }),
    share(),
  );

  constructor(private http: HttpClient) {}

  cacheBust() {
    this.eventBus$.next();
  }
}

<details>
<summary>英文:</summary>

&gt; cart$ observable subscribed in 2 different components via async pipe

This causes 2 separate subscriptions of a cold observable. This means you will get 2 producers for your 2 subscribers. So each subscription is getting a separate `distinctUntilChanged()`. If you want to share a single producer across multiple consumers, this is where you can use `share()` to turn a cold observable into a hot one.

Keep in mind this is essentially putting a `Subject` as a middle-man between your single producer and multiple subscribers. [There are some details][1] regarding when the `Subject` actually subscribes to the producer, and how it manages its lifecycle. Here, both your components are subscribing upon construction and shouldn&#39;t miss any events passed through `share()`. You can look into `shareReplay()` if you need caching for late subscribers.

    @Injectable({
      providedIn: &#39;root&#39;,
    })
    export class CustomerFacadeService {
      private eventBus$ = new BehaviorSubject&lt;void&gt;(undefined);
      private cachedCartData$ = this.http.get(
        &#39;https://62c41dc97d83a75e39f0d512.mockapi.io/api/v1/cart&#39;
      );
    
      cart$ = this.eventBus$.pipe(
        mergeMap(() =&gt; this.cachedCartData$),
        distinctUntilChanged((prev, current) =&gt; {
          const result = isEqual(prev, current);
          console.log(&#39;result&#39;, result);
          return result;
        }),
        share(),
      );
    
      constructor(private http: HttpClient) {}
    
      cacheBust() {
        this.eventBus$.next();
      }

  [1]: https://rxjs.dev/api/index/interface/ShareConfig

</details>