前端框架 Rxjs 实践指北

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

完美的合作关系

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

UI = f(data)

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

data = g(source)

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

UI = f(g(source))

是不是和MV定义很像：

先从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，它的自我介绍非常简单：

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个：

1. UI渲染的数据在哪里定义？
2. Rxjs流在哪里构建？
3. 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不够直观，依赖的是约定和认知，不够清晰明确。

Vue Composition API

正如React hooks，Vue Composition API也旨在解决逻辑碎片化的问题。

基于Vue Composition API，如何集成Rxjs有了新的讨论，优点在于对于使用方，逻辑更加聚合。

具体讨论看看这里：Vue Composition API and vue-rx。

总结

首先，明确了Rxjs和React/Vue等前端框架的关系，这两个者在应用上可以是个合作关系。

其次，通过 rxjs-hooks、vue-rx 了解如何在前端框架中集成 Rxjs。这是一个在给定框架内，找寻最合适的机制的问题，React当仁不让的hooks、Vue相对繁琐的插件。但本质上，集成Rxjs要解决的问题是一致的：

1. 在哪里做最后消费数据的定义，准备好一个坑位；
2. 流的逻辑：流的构建，流是什么 => 流执行 => 数据订阅，数据赋值；
3. 更好的场景覆盖：如何实现依赖驱动、行为驱动；

最后，希望Rxjs能在你的框架你的日常开发中发挥它的魔力！