RxJS useObservable副作用失效引发重复ACK问题求助

问题场景 #

我使用useObservable来避免接收重复或过期的ACK，代码如下：

import { useObservable } from 'rxjs-hooks';
import { distinctUntilChanged, filter, map, tap } from 'rxjs/operators';

const ack = useObservable<InstantAckValueDecodedData | null>(() => {
  // 用于防止接收来自之前命令的ACK
  let lastAckTimestamp = Date.now();
  console.log('initial lastAckTimestamp', lastAckTimestamp);

  return device.instant$.pipe(
    map(toDecodedAckFrame),
    filter(guardOnlyInstantAckValueDecodedData),
    filter(onlyMatchingTag),
    filter(({ timestamp }) => {
      console.log(
        'filter',
        timestamp,
        lastAckTimestamp,
        timestamp > lastAckTimestamp,
      );
      return timestamp > lastAckTimestamp;
    }),
    tap(({ timestamp }) => {
      console.log('tap', timestamp, lastAckTimestamp);
      lastAckTimestamp = timestamp;
    }),
    distinctUntilChanged(isEqual),
  );
}, null);

运行后得到如下日志：

LOG  initial lastAckTimestamp 1677164910821

LOG  filter 1677164914666 1677164909553 true
LOG  tap 1677164914666 1677164909553
LOG  filter 1677164914666 1677164910821 true
LOG  tap 1677164914666 1677164910821
LOG  filter 1677164914666 1677164914666 false
LOG  filter 1677164914666 1677164914666 false

我无法理解第二个filter中为何会出现初始值1677164910821，这导致了我收到重复的ACK。

问题原因 #

核心问题是组件重渲染时，useObservable的工厂函数会被重新执行，每次执行都会创建一个新的lastAckTimestamp变量，同时之前的流并未被销毁，仍然在监听device.instant$事件。同一个ACK事件会被多个流实例处理：

• 旧流实例使用的是已经更新过的lastAckTimestamp（1677164909553）
• 新流实例使用的是重新初始化的初始值（1677164910821）
  这就导致同一个ACK被多次通过filter校验，最终出现重复接收的情况。

解决方法 #

方法1：用useRef保存时间戳状态 #

useRef的内容在组件整个生命周期内保持不变，不会随重渲染重新初始化，确保所有流实例共享同一个时间戳状态：

import { useRef } from 'react';
import { useObservable } from 'rxjs-hooks';
import { distinctUntilChanged, filter, map, tap } from 'rxjs/operators';

// ...

const lastAckTimestampRef = useRef(Date.now());
console.log('initial lastAckTimestamp', lastAckTimestampRef.current);

const ack = useObservable<InstantAckValueDecodedData | null>(() => {
  return device.instant$.pipe(
    map(toDecodedAckFrame),
    filter(guardOnlyInstantAckValueDecodedData),
    filter(onlyMatchingTag),
    filter(({ timestamp }) => {
      console.log(
        'filter',
        timestamp,
        lastAckTimestampRef.current,
        timestamp > lastAckTimestampRef.current,
      );
      return timestamp > lastAckTimestampRef.current;
    }),
    tap(({ timestamp }) => {
      console.log('tap', timestamp, lastAckTimestampRef.current);
      lastAckTimestampRef.current = timestamp;
    }),
    distinctUntilChanged(isEqual),
  );
}, null);

方法2：让流成为单例 #

使用share()操作符确保流被共享，避免每次工厂函数执行都创建新的流实例，从而减少重复的时间戳变量：

const ack = useObservable<InstantAckValueDecodedData | null>(() => {
  let lastAckTimestamp = Date.now();
  console.log('initial lastAckTimestamp', lastAckTimestamp);

  return device.instant$.pipe(
    map(toDecodedAckFrame),
    filter(guardOnlyInstantAckValueDecodedData),
    filter(onlyMatchingTag),
    filter(({ timestamp }) => {
      console.log(
        'filter',
        timestamp,
        lastAckTimestamp,
        timestamp > lastAckTimestamp,
      );
      return timestamp > lastAckTimestamp;
    }),
    tap(({ timestamp }) => {
      console.log('tap', timestamp, lastAckTimestamp);
      lastAckTimestamp = timestamp;
    }),
    distinctUntilChanged(isEqual),
    share() // 共享流，避免多次订阅创建新实例
  );
}, null);

推荐优先使用方法1，因为它能从根源上保证时间戳状态的唯一性，不受流订阅次数的影响。

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