漫谈RxJS之基础原理篇

[EasyTuan]

我们要谈及RxJS，不得不先说下Reactive Extensions（Rx），这是一种编程模式，他的目标是对异步的集合进行操作（ps: 这种编辑模式有利有弊，当然这不是本文的重点，这里不再赘述，有兴趣的同学可以自行查阅相关资料。）。Rx家族成员很多，如RxCpp、RxJava、Rx.NET、Rx.rb、RxPy等，我们从命名上不难看出，后缀就是语言的名称，那么就不难理解RxJS是Javascript语言对Reactive Extensions（Rx）的实现。

RxJS是什么？

引自官方文档的概念：

RxJS 是一个库，它通过使用 observable 序列来编写异步和基于事件的程序。它提供了一个核心类型 Observable，附属类型 (Observer、 Schedulers、 Subjects) 和受 [Array#extras] 启发的操作符 (map、filter、reduce、every, 等等)，这些数组操作符可以把异步事件作为集合来处理。

可以把 RxJS 当做是用来处理事件的 Lodash 。

官方文档引入了很多概念，如Observer、 Schedulers、 Subjects，这些对初学者并不友好，因为这些都是RxJS的概念。RxJS的学习路线对于大多数人来说都太陡峭了，很容易从入门到放弃，甚至还没入门就放弃了。

我们把概念做个精简，RxJS其实就是发布者和订阅者之间玩的游戏。

在设计模式中，有一个模式就是发布订阅模式，这个模式中，

• 发布者只负责事件的通知
• 订阅者只负责接收到通知后完成自身的业务逻辑，而并不关心消息的来源

大家都责任都很明确，但是必然有一些负责的情况，比如广播是私有的还是公开的，如果订阅者迟到了，那么前面的消息是否需要告知订阅者，对着这些情况的处理，RxJS定义了一套完整的操作符，注意是完整的，这个世界上所有的情况都被涵盖在操作符中了，你不必再造轮子了。

解决了什么问题？

上文说到RxJS其实就是发布者和订阅者之间玩的游戏，可能你还是不理解，那我们再说的具体点。

1. RxJS是一个事件的处理库，用于处理各种事件，比如DOM的点击、定时器的触发、API的消息获取
2. 事件需要订阅，比如DOM的点击通过addEventListener，API获取通过注册回调函数

通过两个步骤，我们成功划分出了发布者和订阅者，但是在实际业务场景中，这些事件的订阅和发布会更加复杂，涉及一对多、多对一、多对多等场景。

而异步事件的处理则更加复杂，相信很多人都听说过回掉地狱这个词，指的是在没有promise前，API的订阅需要用回调函数进行注册，一旦API有串行请求，就很容易写出冲击波代码，像这样：

    function getList() {
        request('url', {
            succcess: function(res) {
                request('url', {
                    data: res,
                    succcess: function(res) {
                        request('url', {
                            data: res,
                            succcess: function(res) {
                                // TODO
                            }
                        })
                    }
                })
            }
        })
    }

在promise成为规范后，后面async await这种用同步写异步代码的方式流行后，这种情况得到了改善，但是在RxJS中，你会看到一个不一样的编码方式，毕竟promise也属于另一个编程范式。

import { ajax } from 'rxjs/ajax';
import { concatMap } from 'rxjs/operators';

/* 1. 判断登录
 * 2. 获取基础信息
 * 3. 根据基础信息换取商品信息
 */
const order$ = ajax('url') // 1. 判断登录
  .pipe(
    concatMap(isLogin => ajax('url')), // 2. 获取基础信息
    concatMap(info => ajax('url')), // 3. 根据基础信息换取商品信息
  )
  .subscribe(goods => {
      // TODO
  });

带来了什么新问题？

不得不说，Reactive Extensions（Rx）是一种非常棒的编码模式，不过我们得结合实际的业务场景做具体分析。

1. 团队内的成员是否具备高度抽象思维，因为这是一种函数式、响应式编程范式，团队成员是否可以快速学习进入开发？
2. RxJS最擅长处理的是异步事件，那么我们的业务场景是否真的有这么复杂吗？
3. RxJS的代码是高度抽象的，抽象的代码是不如命令式代码易读，这点无可厚非，那么我们该如何组织我们的业务代码？

函数式编程范式

这里顺带提一下函数式编程范式，首先，这是一种有约束的编程范式：

1. 声明式(Declarative)

声明式，与之对应的是命令式，我们可以看下代码有何不同：

const arr = [1, 2, 3, 4, 5]
// 命令式
for (let i = 0; i < arr.length; i++) {
    console.log(arr[i])
}

// 声明式
arr.map((item) => {
    console.log(item)
})

我们不难发现，使用声明式，代码简洁很多，看不到for了，也没有了额外的变量声明。

2. 纯函数(Pure Function)

函数的调用不会有额外的副作用（IO操作、DOM操作、网络请求等），每次的输入都有唯一确认的输出。

3. 数据不可变性(Immutability)

不改变源数据，举个例子：

const arr = [1, 2, 3, 4, 5]
// 改变了源数据
arr.push(6)

// 添加了新元素，但并没有改变原始数据，符合数据不可变性原则
function push2(source, item) {
    return [...source, item]
}

结合实际业务场景

这也是我最近做的例子，个人觉得结合RxJS会比较合适，就是IM聊天会话页。

我们先来梳理一下逻辑：

事件：

1. 通过HTTP GET拉取历史会话列表；
2. 通过websocket协议实时接收消息；
3. 输入事件；
4. 发送内容事件；
5. 点击内容事件（如图片预览、视频播放、语音播放等）;

我们结合下UI交互，再组织下逻辑层，不难发现：

1. 对于【聊天内容显示区域】，很明显是订阅者，订阅者的特点是什么？上文已提过：订阅者只负责接收到通知后完成自身的业务逻辑，而并不关心消息的来源。

2. 发布者有哪些？

   • 通过websocket推送的消息（异步事件）
   • 通过HTTP GET拉取的消息（异步事件）
   • 用户点击发送的消息（同步事件）

一个订阅者支持订阅多个事件流吗？RxJS当然支持，我们可以通过合并事件流、转化事件流等手段，来达到我们的目的。

这里代码就不贴了，感兴趣的同学可以按上述思路实践一下。

小结

本文还有很多RxJS的概念没有介绍到，比如：

• Hot Observable和Cold Observable用于解决存在多个订阅者，有人会迟到的问题
• 经典的弹珠图表示法，让我们可以清晰的看到事件的流向
• 多播，用于处理一个事件流存在多个订阅者的问题
• 异常处理，上游的数据出错了如何处理
• ...

如果大家有感兴趣的点，也欢迎随时讨论。