Vue中的非标准Observable集成：实现Custom Ref与外部数据源的同步与调度

Vue中的非标准Observable集成：实现Custom Ref与外部数据源的同步与调度

大家好，今天我们来深入探讨一个在Vue开发中相对高级但非常实用的主题：如何将非标准的Observable数据源集成到Vue的响应式系统中，并通过自定义Ref实现数据的同步与调度。

通常情况下，我们使用Vue内置的响应式系统，比如ref、reactive，来管理组件的状态。然而，在实际项目中，我们经常会遇到需要与外部数据源交互的情况。这些数据源可能来自第三方库（例如RxJS的Observable），或者是由WebSocket、EventSource等技术驱动的实时数据流。直接将这些数据源集成到Vue的响应式系统中会带来一些挑战，因为它们的数据更新机制与Vue的依赖追踪机制并不完全兼容。

为了解决这个问题，我们可以利用Vue 3提供的强大的customRef API，创建一个桥梁，将外部Observable数据源的变化同步到Vue的响应式系统中，并根据需要进行调度。

1. 为什么需要自定义Ref？

首先，让我们明确为什么不能直接使用内置的ref或reactive：

• 依赖追踪不匹配： Vue的ref和reactive是通过Proxy实现的，它们能够自动追踪组件对数据的访问，并在数据变化时触发更新。而外部Observable数据源通常有自己的通知机制，Vue无法自动追踪这些变化。
• 数据类型不兼容： 有些Observable数据源返回的数据类型可能不是Vue所期望的，例如，它们可能返回一个Stream，而不是一个具体的值。
• 更新策略不一致： 外部数据源的更新频率可能很高，而Vue的更新机制有一定的开销。我们需要根据实际情况控制更新的频率，避免不必要的渲染。

2. Custom Ref API 介绍

customRef 是 Vue 3 提供的一个高级 API，允许我们创建自定义的 ref。它接受一个工厂函数作为参数，该工厂函数接收 track 和 trigger 两个函数作为参数，并返回一个包含 get 和 set 属性的对象。

• track(): 在 get 函数中调用，用于告诉 Vue 追踪当前 ref 的依赖。当这个 ref 的值被访问时，Vue 会将当前组件或计算属性添加到这个 ref 的依赖列表中。
• trigger(): 在 set 函数中或在外部数据源更新时调用，用于触发依赖于当前 ref 的组件或计算属性的更新。

3. 实现Custom Ref集成Observable数据源

接下来，我们通过一个具体的例子来演示如何使用 customRef 将 RxJS 的 Observable 集成到 Vue 的响应式系统中。

假设我们有一个 Observable 数据源，它每隔一段时间发出一个新的数字：

import { interval, BehaviorSubject } from 'rxjs';
import { startWith } from 'rxjs/operators';

// 创建一个 BehaviorSubject，初始化值为 0
const numberSubject = new BehaviorSubject<number>(0);

// 每隔 1 秒发出一个新数字，并更新 BehaviorSubject 的值
interval(1000).pipe(startWith(0)).subscribe(value => {
  numberSubject.next(value);
});

// 导出 Observable
export const numberObservable = numberSubject.asObservable();

现在，我们创建一个自定义的 ref，它将监听这个 Observable 的变化，并将最新的值同步到 Vue 的响应式系统中：

import { customRef, onUnmounted } from 'vue';
import { numberObservable } from './observable'; // 引入上面的 observable

export function useObservable(observable: any, initialValue: any) {
  let value = initialValue;
  let subscription: any;

  const customRefImplementation = (track: any, trigger: any) => {
    return {
      get() {
        track(); // 告诉 Vue 追踪依赖
        return value;
      },
      set(newValue: any) {
        value = newValue;
        trigger(); // 触发更新
      }
    };
  };

  const observableRef = customRef(customRefImplementation);

  // 订阅 Observable
  subscription = observable.subscribe((newValue: any) => {
    value = newValue;
    // 手动触发更新，这里不能直接用 observableRef.value = newValue, 会造成无限循环
    // 因为set方法里面又会触发observable.subscribe...
    trigger();

  });

  // 在组件卸载时取消订阅
  onUnmounted(() => {
    if (subscription) {
      subscription.unsubscribe();
    }
  });

  return observableRef;
}

代码解释:

1. useObservable(observable, initialValue) 函数：
   • 接收一个 Observable 和一个初始值作为参数。
   • 使用 customRef 创建一个自定义的 ref。
   • 订阅 Observable，并在收到新值时更新 ref 的值。
   • 在组件卸载时取消订阅，防止内存泄漏。
2. customRefImplementation(track, trigger) 函数：
   • get() 函数：调用 track() 告诉 Vue 追踪依赖，并返回当前值。
   • set() 函数：设置新值，并调用 trigger() 触发更新。
3. 订阅 Observable：
   • observable.subscribe(newValue => { ... }) 订阅 Observable，并在收到新值时更新 value 变量。
   • 关键点：在这里，我们不能直接使用 observableRef.value = newValue 来更新 ref 的值。 因为 observableRef.value = newValue 会触发 set() 函数，而 set() 函数又会调用 trigger() 触发更新，导致无限循环。所以，我们直接更新 value 变量，然后手动调用 trigger() 触发更新。
4. onUnmounted():
   • 在组件卸载时取消订阅，释放资源。

4. 在Vue组件中使用Custom Ref

现在，我们可以在 Vue 组件中使用这个自定义的 ref：

<template>
  <div>
    <p>Observable Value: {{ number }}</p>
  </div>
</template>

<script lang="ts">
import { defineComponent } from 'vue';
import { numberObservable } from './observable';
import { useObservable } from './useObservable';

export default defineComponent({
  setup() {
    const number = useObservable(numberObservable, 0);

    return {
      number
    };
  }
});
</script>

在这个组件中，我们使用 useObservable 函数将 numberObservable 集成到 Vue 的响应式系统中，并将返回的 ref 赋值给 number 变量。现在，当 numberObservable 发出新值时，组件会自动更新，显示最新的数字。

5. 调度更新策略

在某些情况下，外部数据源的更新频率可能很高，而我们并不需要每次更新都立即触发 Vue 的更新。例如，我们可能只需要每隔一段时间更新一次界面，或者只在特定条件下才更新界面。

为了实现这种调度策略，我们可以在 useObservable 函数中添加一个节流或防抖机制：

import { customRef, onUnmounted, ref } from 'vue';
import { numberObservable } from './observable'; // 引入上面的 observable
import { throttle } from 'lodash-es'; // 引入 lodash 的 throttle 函数

export function useObservableWithThrottle(observable: any, initialValue: any, wait: number) {
  let value = initialValue;
  let subscription: any;

  const throttledTrigger = throttle(() => {
      trigger();
  }, wait);

  const customRefImplementation = (track: any, trigger: any) => {
    return {
      get() {
        track(); // 告诉 Vue 追踪依赖
        return value;
      },
      set(newValue: any) {
        value = newValue;
        throttledTrigger(); // 触发更新
      }
    };
  };

  const observableRef = customRef(customRefImplementation);

  // 订阅 Observable
  subscription = observable.subscribe((newValue: any) => {
    value = newValue;
    // 手动触发更新，这里不能直接用 observableRef.value = newValue, 会造成无限循环
    // 因为set方法里面又会触发observable.subscribe...
    throttledTrigger();

  });

  // 在组件卸载时取消订阅
  onUnmounted(() => {
    if (subscription) {
      subscription.unsubscribe();
    }
  });

  return observableRef;
}

在这个例子中，我们使用了 lodash 的 throttle 函数来限制 trigger() 的调用频率。现在，只有在距离上次调用 trigger() 超过 wait 毫秒时，才会真正触发更新。

6. 错误处理

在与外部数据源交互时，错误处理非常重要。我们需要捕获 Observable 发出的错误，并将其显示给用户，或者进行其他处理。

import { customRef, onUnmounted } from 'vue';
import { numberObservable } from './observable'; // 引入上面的 observable

export function useObservableWithErrorHandling(observable: any, initialValue: any) {
  let value = initialValue;
  let subscription: any;
  const error = ref<Error | null>(null);

  const customRefImplementation = (track: any, trigger: any) => {
    return {
      get() {
        track(); // 告诉 Vue 追踪依赖
        return value;
      },
      set(newValue: any) {
        value = newValue;
        trigger(); // 触发更新
      }
    };
  };

  const observableRef = customRef(customRefImplementation);

  // 订阅 Observable
  subscription = observable.subscribe({
      next: (newValue: any) => {
          value = newValue;
          trigger();
          error.value = null; // 清空错误
      },
      error: (err: any) => {
          console.error("Observable error:", err);
          error.value = err; // 设置错误
      }
  });

  // 在组件卸载时取消订阅
  onUnmounted(() => {
    if (subscription) {
      subscription.unsubscribe();
    }
  });

  return {
      data: observableRef,
      error
  };
}

在这个例子中，我们使用 subscribe 的对象形式，提供了 next 和 error 两个回调函数。在 error 回调函数中，我们将错误信息存储到一个 ref 中，并在组件中显示该错误信息。

7. 示例：集成WebSocket数据源

以下是一个使用 customRef 集成 WebSocket 数据源的示例：

import { customRef, onUnmounted } from 'vue';

export function useWebSocket(url: string) {
  let value: any = null;
  let socket: WebSocket | null = null;

  const customRefImplementation = (track: any, trigger: any) => {
    return {
      get() {
        track();
        return value;
      },
      set(newValue: any) {
        value = newValue;
        trigger();
      },
    };
  };

  const webSocketRef = customRef(customRefImplementation);

  const connect = () => {
    socket = new WebSocket(url);

    socket.onopen = () => {
      console.log('WebSocket connected');
    };

    socket.onmessage = (event) => {
      try {
          const parsedData = JSON.parse(event.data);
          value = parsedData;
          trigger(); // 触发更新
      } catch (e) {
          console.error("Failed to parse WebSocket data:", e);
      }
    };

    socket.onclose = () => {
      console.log('WebSocket disconnected');
      socket = null;
    };

    socket.onerror = (error) => {
      console.error('WebSocket error:', error);
      socket = null;
    };
  };

  const close = () => {
    if (socket) {
      socket.close();
      socket = null;
    }
  };

  onUnmounted(() => {
    close();
  });

  connect(); // 立即连接

  return {
    data: webSocketRef,
    close,
  };
}

表格：不同数据源集成方案对比

特性	内置 Ref/Reactive	Custom Ref + Observable	Custom Ref + WebSocket
适用场景	简单状态管理	复杂外部数据源集成	实时数据流集成
依赖追踪	自动	手动 (track, trigger)	手动 (track, trigger)
更新策略	立即	可自定义 (节流, 防抖)	可自定义
错误处理	需手动实现	需手动实现	需手动实现
代码复杂度	低	中	中

8. 注意事项

• 避免循环依赖： 在 set() 函数中更新 ref 的值时，要避免触发循环依赖。
• 内存泄漏： 确保在组件卸载时取消订阅 Observable 或关闭 WebSocket 连接，防止内存泄漏。
• 错误处理： 妥善处理外部数据源可能发生的错误。
• 性能优化： 根据实际情况调整更新策略，避免不必要的渲染。
• 类型安全： 使用 TypeScript 等静态类型检查工具，确保类型安全。

总结

使用customRef 可以优雅地将非标准的Observable数据源集成到Vue的响应式系统中，实现数据的同步与调度。 通过合理地使用track 和 trigger 函数，以及根据实际需要添加调度策略和错误处理机制，可以构建出更加健壮和高效的Vue应用。 这个技术在处理来自外部数据源的复杂状态时非常有用，特别是当数据源具有自己的更新机制时。

更多IT精英技术系列讲座，到智猿学院