前端框架 Rxjs 实践指北

用户1250838

发布于 2021-05-31 14:47:58

5.4K0

发布于 2021-05-31 14:47:58

文章被收录于专栏：洛竹早茶馆洛竹早茶馆

本文由涂鸦大前端成员无限撰写，已授权涂鸦大前端独家使用，包括但不限于编辑、标注原创等权益。「洛竹早茶馆」已获作者授权转载。

本文主要介绍如何在前端框架 React、Vue 使用 Rxjs，开源的 rxjs-hooks、vue-rx背后做了哪些事情。在开始之前，希望你对响应式编程、Rxjs 有一个基本的认识。让我们开始吧！

完美的合作关系

前端框架的职责（比如React、Vue）：数据和UI的同步，当数据发生变化的时候，UI 自动刷新；

UI = f(data)

响应式编程干了什么（比如Rxjs）：关注的点在数据，从数据流的源头、到数据的处理、数据的订阅（数据的消费）；

data = g(source)

两者的关系呢并不冲突，甚至在某些场景是完美的合作关系，前端框架可以作为响应式编程数据的一个消费者：

UI = f(g(source))

是不是和MV定义很像：

MVVM happens to be a good fit for Rx*. Quoting Wikipedia: The view model of MVVM is a value converter meaning that the view model is responsible for exposing the data objects from the model in such a way that those objects are easily managed and consumed. In this respect, the view model is more model than view, and handles most if not all of the view’s display logic.

先从React开始：rxjs-hooks

在React中（仅考虑函数式组件）有两种形式可直接表达“非一次性赋值”：

useMemo

const greeting = React.useMemo(() => `${greet}, ${name}!`, [greet, name]);

useState+useEffect

const [greeting, setGreeting] = useState(() => `${greet}, ${name}!`);

useEffect(() => {
  setGreeting(() => `${greet}, ${name}!`);
}, [greet, name]);

注意：useMemo计算数据在render之前，而useState+useEffect的数据计算逻辑在useEffect，在render之后。

想要接入Rxjs，要做整个“管道”的搭建，包括Observable的准备、数据处理、数据订阅，甚至是产生一些副作用（tap），而这些超出了useMemo的承载力。反观useEffect却很适合（为副作用而生），来尝试以useState+useEffect去扩展。

首先来一个基础版本：

import * as React from 'react';
import { combineLatest, from, of } from 'rxjs';
import { catchError, map, startWith } from 'rxjs/operators';

const GreetSomeone = ({ greet = 'Hello' }) => {
    const [greeting, setGreeting] = React.useState('');

    React.useEffect(() => {
        const greet$ = of(greet);
       // fetchSomeName: 远程搜索数据
        const name$ = from(fetchSomeName()).pipe(
            startWith('World'),
            catchError(() => of('Mololongo')),
        );

        const greeting$ = combineLatest(greet$, name$).pipe(
            map(([greet, name]) => `${greet}, ${name}!`)
        );

        const subscription = greeting$.subscribe(value => {
            setGreeting(value);
        });

        return () => {
            subscription.unsubscribe();
        }
    }, []);

    return <p>{greeting}</p>;
};

有点模样了，在useEffect中搭建了Rxjs流，数据订阅后，把数据记录在组件内用作数据渲染，同时当组件销毁时，取消订阅。

但这里有一个问题，组件接受的 prop greet是会发生变化的，而greet$的数据不会发生更新。怎么解决呢？如果这样处理：

React.useEffect(() => {
    const greet$ = of(greet);    
    /**
     * 同上，流构建逻辑
    **/
}, [greet]);

这样的问题是，每次Rxjs流会因为 greet更新而重新生成，继而接口调用fetchSomeName会再次调用。成本有点大。

怎么解决呢？

再引入一个useEffect，用Rxjs的Subject.next主动去推数据，而保证构建Rxjs流仅执行一次，贴上完整代码：

import * as React from 'react';
import { BehaviorSubject, combineLatest, from, of } from 'rxjs';
import { catchError, map, startWith } from 'rxjs/operators';

const GreetSomeone = ({ greet = 'Hello' }) => {
    // 使用React.useRef在组件生命周期保持不变
    const greet$ = React.useRef(new BehaviorSubject(greet));

    // Subject.next 推数据，使得Rxjs数据更新
    React.useEffect(() => {
        greet$.current.next(greet);
    }, [greet]);

    const [greeting, setGreeting] = React.useState('');
  
    // 逻辑不变，仅执行一次
    React.useEffect(() => {
        const name$ = from(fetchSomeName()).pipe(
            startWith('World'),
            catchError(() => of('Mololongo')),
        );

        const greeting$ = combineLatest(greet$.current, name$).pipe(
            map(([greet, name]) => `${greet}, ${name}!`)
        );

        const subscription = greeting$.subscribe(value => {
            setGreeting(value);
        });

        return () => {
            subscription.unsubscribe();
        }
    }, [greet$]);

    return <p>{greeting}</p>;
};

有了基于hooks实现的基本认识，我们来看一下开源实现的 Rxjs-hooks，它的自我介绍非常简单：

React hooks for RxJS.

Rxjs-hooks设计了两个hook，一个是useObservable，一个是useEventCallback。

看一下useObservable：摘除TS类型后，是不是和上面提到的结构是一致的

export function useObservable(
  inputFactory,
  initialState,
  inputs,
){
  const [state, setState] = useState(typeof initialState !== 'undefined' ? initialState : null)
 
  const state$ = useConstant(() => new BehaviorSubject(initialState))
  const inputs$ = useConstant(() => new BehaviorSubject(inputs))

  useEffect(() => {
    inputs$.next(inputs)
  }, inputs || [])

  useEffect(() => {
    let output$
    if (inputs) {
      output$ = inputFactory(state$, inputs$)
    } else {
      output$ = inputFactory(state$) 
    }
    const subscription = output$.subscribe((value) => {
      state$.next(value)
      setState(value)
    })
    return () => {
      subscription.unsubscribe()
      inputs$.complete()
      state$.complete()
    }
  }, []) // immutable forever

  return state
}

使用举例：

import React from 'react'
import ReactDOM from 'react-dom'
import { useObservable } from 'rxjs-hooks'
import { of } from 'rxjs'
import { map } from 'rxjs/operators'

function App(props: { foo: number }) {
  const value = useObservable((_, inputs$) => inputs$.pipe(
    map(([val]) => val + 1),
  ), 200, [props.foo])
  return (
    // render three times
    // 200 and 1001 and 2001
    <h1>{value}</h1>
  )
}

可见useObservable针对props、state去构建Observable，最后返回被订阅的数据。所以入参就是：inputFactory（即Rxjs流的构建逻辑）、initialState、inputs。

useEventCallback类似，除了hook返回了被订阅的数据外，还返回了callback，它处理事件响应的情况：

const event$ = useConstant(() => new Subject<EventValue>())

function eventCallback(e: EventValue) {
  return event$.next(e)
}

return [returnedCallback as VoidableEventCallback<EventValue>, state]

思考：rxjs落地环境需要的条件

回顾一下Rxjs在React中的落地，要解决的问题有3个：

UI渲染的数据在哪里定义？
Rxjs流在哪里构建？
Rxjs流如何使得Observable持续冒（emit）出值而流动？

动动手：Vue + Rxjs

基于同样的想法，尝试在Vue中实现一下Rxjs的使用：

<template>
  <div>{{ greeting }}</div>
</template>

<script>
import { from, combineLatest, BehaviorSubject } from "rxjs";
import { map } from "rxjs/operators";

let subscription = null,
  greet$ = null;

export default {
  name: "TryRxInVue",
  props: {
    greet: {
      type: String,
      default: "hello",
    },
  },
  data() {
    return {
      greeting: "",
    };
  },
  // 监听依赖，使得流动
  watch: {
    greet(value) {
      this.greet$.next(value);
    },
  },
  // 不同生命周期钩子
  mounted() {
    this.initStream();
  },
  beforeDestroy() {
    subscription = null;
    greet$ = null;
  },
  methods: {
    // 初始化流，在组件mounted时调用
    initStream() {
      greet$ = new BehaviorSubject(this.greet);
      const name$ = from(Promise.resolve("world"));

      const greeting$ = combineLatest(greet$, name$).pipe(
        map(([greet, name]) => `${greet},${name}!`)
      );

      subscription = greeting$.subscribe((value) => {
        this.greeting = value;
      });
    },
  },
};
</script>

会发现缺点在于逻辑非常分散，那么有没有什么好的封装形式呢？

Vue提供的插件机制！

概括来说：将流的构建写在约定的配置位置，通过插件翻译配置，塞入相应的生命周期、监听等执行。

对比开源库的实现

找到了Vue官方实现的基于Rxjs V6的Vue.js集成：vue-rx。正如 vue-router、vuex等一样，它也是一个Vue插件。

看了源码后，思路基本和自己考虑的是一致的。有以下几个重要的点做下记录。

最最核心的 subscriptions 配置，它这样使用：

<template>
  <div>
    <p>{{ num }}</p>
  </div>
</template>

<script>
import { interval } from "rxjs";

export default {
  name: "Demo",
  subscriptions() {
    return {
      num: interval(1000).pipe(take(10))
    };
  },
};
</script>

它背后做了哪些事呢？即怎么翻译的呢？

通过Mixin，在生命周期 created时候：
- 同名key，定义为响应式数据，挂在vm实例上，即这里的num会挂在vm.num;
- 对每个ob，挂在vm.observables上，即vm.observables.num可以获取到这个ob，但貌似没啥用...；
- 执行ob，数据订阅，赋值同名vm[key]，即vm.num和这个ob绑定了（注：这里对于一个vm，用了一个Subscription对象，目的是可以做统一订阅、取消订阅ob）；
通过Mixin，在生命周期beforeDestroy时候：取消订阅；

简单看下源码：

import { defineReactive } from './util'
import { Subject, Subscription } from 'rxjs'

export default {
  created () {
    const vm = this

    // subscriptions来来
    let obs = vm.$options.subscriptions
    
    if (obs) {
      vm.$observables = {}
      vm._subscription = new Subscription()
      Object.keys(obs).forEach(key => {

        // 定义了响应式数据，key挂在vm实例上
        defineReactive(vm, key, undefined)
        // obs也挂在了vm.$observables上
        const ob = vm.$observables[key] = obs[key]

        // 执行ob，数据订阅，最后赋值给准备好的obs[key]坑位
        vm._subscription.add(obs[key].subscribe(value => {
          vm[key] = value
        }, (error) => { throw error }))
      })
    }
  },

  beforeDestroy () {
    // 取消订阅
    if (this._subscription) {
      this._subscription.unsubscribe()
    }
  }
}

subscriptions搭起来后，核心问题就解决了，剩下的是如何实现依赖驱动和行为驱动；

如何实现依赖驱动呢？

vue-rx暴露了一个$watchAsObservable方法，它可以这样用：

import { pluck, map } from 'rxjs/operators'

const vm = new Vue({
  data: {
    a: 1
  },
  subscriptions () {
    // declaratively map to another property with Rx operators
    return {
      aPlusOne: this.$watchAsObservable('a').pipe(
        pluck('newValue'),
        map(a => a + 1)
      )
    }
  }
})

$watchAsObservable参数为一个表达式，返回一个ob，当表达式值发生变化时，ob冒出值。它的源码实现侵入了New Observable({...})：

import { Observable, Subscription } from 'rxjs'

export default function watchAsObservable (expOrFn, options) {
  const vm = this
  const obs$ = new Observable(observer => {
    let _unwatch
    const watch = () => {
      _unwatch = vm.$watch(expOrFn, (newValue, oldValue) => {
        observer.next({ oldValue: oldValue, newValue: newValue })
      }, options)
    }
    
  // 这里简单了一下
    watch()

    // 返回取消订阅
    return new Subscription(() => {
      _unwatch && _unwatch()
    })
  })

  return obs$
}

这样的方式在vue-rx中很常见。会发现，逻辑和自己写的简单Demo也是一致的，只不过ob的声明、观察值的变化冒出值的逻辑给封装进插件了。

如何实现行为驱动呢？

自己写的简单Demo没有包括，但无非是定义个Subject，这个Subject参与到流的构建，在事件响应的时候由它冒出值去推动流数据的变化。

嗨，别说，这确实是vue-rx提供行为驱动方法之一背后做的事情，通过自定义指令v-stream+配置domStreams，这里不做展开了。

另外一种方式是vue-rx暴露的实例observableMethods，它的实现还挺精妙，简单讲讲。比如使用是这样的：

new Vue({
  observableMethods: {
    submitHandler: 'submitHandler$'
    // or with Array shothand: ['submitHandler']
  }
})

它会在Mixin created 生命周期时，挂载两个属性，vm.submitHandler$是一个参与流构建的ob，vm.submitHandler是这个ob的数据生产者，暴露的接口，参数就是ob冒出的值。这样的机制，即包含了ob的声明，又包含了推动ob.next方法的暴露。缺点就是，哪个是驱动的方法，哪个是ob不够直观，依赖的是约定和认知，不够清晰明确。