xState 和 RxJS 简介：状态管理的两个有趣的 JS 库

xState 和 rxjs

xState 是有限状态机在 js 中的实现。rxjs 是响应式编程在 js 中的实现。都是非常有趣的库，这里放在一起讲并不是用来比较，而是这段时间在研究状态管理时发现他两都可以用来实现状态管理。虽然两者毫不相关，但也有跨界合作的地方，比如 xstate 里可以调用 rxjs 的 Observables，把 rxjs 视为数据源。除了 Js 之外，有限状态机和响应式编程在很多编程语言中都有对应的实现。

xState

概念

xstate 概念

有限状态机例子：红绿灯、自动售货机、地铁闸机等

状态机和状态图简介

状态、转换和事件、初始状态、复合状态、最终状态、并行状态、延迟转换、动作（entry/、exit/）

借用 xstate 官网的图：

掌握了状态机和状态图之后，可以开始计划状态图了：

不需要掌握 xState
不需要编写代码，可使用图形化界面
易于需求确认，查漏补缺

使用 xState 构建状态机

import { createMachine, assign } from 'xstate'

const lightMachine = createMachine(
  {
    // 状态机标识
    id: 'light',

    // 初始状态
    initial: 'green',

    // 整个状态机的本地 context
    context: {
      elapsed: 0,
      direction: 'east',
    },

    // 状态定义
    states: {
      green: {
        type: 'atomic',
        // 通过字符串引用 action
        entry: 'alertGreen',
        on: {
          // 事件和转换
          CHANGE: { target: 'yellow' },
        },
      },
      yellow: {
        /* ... */
        always: {
          target: 'red',
          cond: 'elapsedNotZero',
        },
      },
      red: {
        /* ... */
      },
    },
    on: {
      RESET: { target: 'green' },
    },
  },
  {
    actions: {
      // action 执行
      alertGreen: (context, event) => {
        assign({
          elapsed: ++context.elapsed,
        })
        console.log('Green!')
      },
    },
    guards: {
      // 守卫
      elapsedNotZero: (context) => context.elapsed !== 0,
    },
    services: {
      /* ... */
    },
  }
)

可视化状态图

执行副作用

动作 Action 通常也称为作用（effects）或副作用（side-effects）。 “副作用”是使用状态图的主要目的。动作事件，对后续的其余部分没有影响，事件只是被触发，流程还是原来设置的那样。

在状态图中，“副作用”可以分为两类：

“即发即弃”副作用，执行同步副作用
调用作用，它执行一个可以异步发送和接收事件的副作用：调用 Promises 、调用 Callbacks 、调用 Machines、调用 Observables

拓展状态机

API：.withConfig 、 .withContext

可以用来隔离相对稳定的业务逻辑与频繁修改的副作用代码；
或者在不同表现 UI 下复用业务逻辑；
甚至可以用于低代码平台，因为描述状态转换的是 JSON

延伸

在 React 使用

官方包@xstate/react 里提供了 useMachine、useSelector 等钩子

有趣的例子

https://codesandbox.io/s/7guis-timer-2gzst?from-embed=&file=/src/timerMachine.ts

演员

actor 和调用 services 的区别，能做什么

Rxjs

为什么用 ReactiveX？

在网页的世界存取任何资源都是非同步(Async)的，比如说我们希望拿到一个档案，要先发送一个请求，然后必须等到档案回来，再执行对这个档案的操作。这就是一个非同步的行为，而随着网页需求的复杂化，我们所写的 JavaScript 就有各种针对非同步行为的写法，例如使用 callback 或是 Promise 对象甚至是新的语法糖 async/await —— 但随着应用需求愈来愈复杂，撰写非同步的代码仍然非常困难。

非同步常见的问题

竞态

每当我们对同一个资源同时做多次的非同步存取时，就可能发生 Race Condition 的问题。比如说我们发了一个 Request 更新使用者资料，然后我们又立即发送另一个 Request 取得使用者资料，这时第一个 Request 和第二个 Request 先后顺序就会影响到最终接收到的结果不同，这就是 Race Condition。

内存泄漏

Memory Leak 是最常被大家忽略的一点。原因是在传统网站的行为，我们每次换页都是整页重刷，并重新执行 JavaScript，所以不太需要理会内存的问题！但是当我们希望将网站做得像应用程式时，这件事就变得很重要。例如做 SPA (Single Page Application) 网站时，我们是透过 JavaScript 来达到切换页面的内容，这时如果有对 DOM 注册监听事件，而没有在适当的时机点把监听的事件移除，就有可能造成 Memory Leak。比如说在 A 页面监听 body 的 scroll 事件，但页面切换时，没有把 scroll 的监听事件移除。

复杂的状态

当有非同步行为时，应用程式的状态就会变得非常复杂！比如说我们有一支付费用户才能播放的视频，首先可能要先抓取这部视频的资讯，接着我们要在播放时去验证使用者是否有权限播放，而使用者也有可能再按下播放后又立即按了取消，而这些都是非同步执行，这时就会各种复杂的状态需要处理。

错误处理

JavaScript 的 try/catch 可以捕捉同步的错误，但非同步的程式就没这么容易，尤其当我们的非同步行为很复杂时，这个问题就愈加明显。

种类繁多的异步 API

DOM Events
XMLHttpRequest
fetch
WebSockets
Server Send Events
Service Worker
Node Stream
Timer

如果我们使用 RxJS，上面所有的 API 都可以透过 RxJS 来处理，就能用同样的 API 操作 (RxJS 的 API)。

这里我们举一个例子，假如我们想要监听点击事件(click event)，但点击一次之后不再监听。

原生 JavaScript

var handler = (e) => {
  console.log(e)
  document.body.removeEventListener('click', handler) // 结束监听
}

// 注册监听
document.body.addEventListener('click', handler)

使用 Rx 大概的样子

import { fromEvent, take } from 'rxjs'

// 注册监听
fromEvent(document.body, 'click')
  // 只取一次
  .pipe(take(1))
  .subscribe(console.log)

不管是针对 DOM Event 还是上面列的各种 API 我们都可以透过 RxJS 的 API 来做数据操作，像是示例中用 take(n) 来设定只取一次，之后就释放内存。

一些概念

函数式编程

Functional Programming 是一种编程范式，属于声明式编程的一种，与声明式编程相对的是命令式编程。

简单说 Functional Programming 核心思想就是做运算处理，并用 function 来思考问题，例如像以下的算数运算式：

5 + 6 - 1 * 3

我们可以写成

const add = (a, b) => a + b
const mul = (a, b) => a * b
const sub = (a, b) => a - b

sub(add(5, 6), mul(1, 3))

我们把每个运算包成一个个不同的 function，并用这些 function 组合出我们要的结果，这就是最简单的 Functional Programming。

不同于面向过程，函数式编程不会涉及到中间态，不会涉及到状态变更。

特性

函数为一等公民，可以赋值给变量、可以作为参数传入、可以作为返回值
纯函数、无副作用
利用参数保存状态，例如 array.prototype.reduce 和 Redux

优势

可读性高，可写成流式操作；纯函数可维护性高，易于单元测试等

观察者模式

在许多 API 的设计上都用了 Observer Pattern 实例，最简单的例子就是 DOM 事件的事件监听，很常见不多讲了。

迭代器模式

也是一种设计模式，其中迭代器用于遍历容器并访问容器的元素。从 ES 6 开始，引入的一种新的遍历机制——迭代器，其就是迭代器模式在 JavaScript 中的一种实现；

特性

访问集合中的内容而不用了解底层的实现。
提供了一个统一的接口遍历不同的集合结构，从而支持同样的算法在不同的集合结构上进行操作。

pull 和 push 系统

pull 和 push，是在软件中消费数据的两种方式，它们描述了数据生产者（或持有者）与数据消费者之间是如何通讯的，在 pull 系统中，数据消费者决定自己何时请求并接收数据；数据持有者只能被动地响应请求，函数、Generator function、async/await 就是 pull 方式；在 push 系统中，数据生产者决定何时向消费者推送数据。数据消费者不知道何时会收到数据更新，Promise 是 push 方式；

响应式编程

Reactive Programming 简单来说就是 当变数或资源发生变动时，由变数或资源自动告诉我发生变动了

这句话看似简单，其实背后隐含两件事

当发生变动 => 非同步：不知道什么时候会发生变动，反正变动时要跟我说
由变数自动告知我 => 我不用写通知我的每一步代码

我们熟知的 vue 也属于响应式编程的一个范例；

作为一种响应式编程，rxjs 则是将观察者模式与迭代器模式和函数式编程与流式操作相结合，以满足对管理事件序列的理想方式的需求；

Rx 的一个核心两个重点

一个核心是 Observable 再加上相关的 Operators(map, filter...)，这是最重要的，其余都是围绕 Observable 来的

两个重点分别是

Observer (是一个回调集合，它知道如何监听 Observable 传递的值)
Subject (相当于一个 EventEmitter，也是将一个值或事件多播到多个 Observers 的唯一方式）

Observable

Observable 属于 push 系统，可以简单理解为高级点的 Promise，因为它可以多次 push，而 Promise 只能一次。

创建 Observable

import { Observable } from 'rxjs'

const observable = new Observable(function subscribe(subscriber) {
  subscriber.next(1)
  subscriber.next(2)
  subscriber.next(3)
  setTimeout(() => {
    // 可以异步输出值
    subscriber.next(4)
    // 标记为完成状态
    subscriber.complete()
    // 5将无法抛出，因为已经完成
    subscriber.next(5)
  }, 1000)
})

// 订阅 Observable，也可以理解为消费它
observable.subscribe((x) => {
  console.log(x)
})

特性

Observable 是惰性的，只有被订阅了才会执行，订阅几次就执行几次，这些特点很像是函数

const foo = new Observable((subscriber) => {
  console.log('Hello')
  subscriber.next(42)
})

foo.subscribe((x) => {
  console.log(x)
})
foo.subscribe((y) => {
  console.log(y)
})

输出

Hello
42
Hello
42

Observable 本质上执行是同步的，除非主动进行异步 push

import { Observable } from 'rxjs'

const foo = new Observable((subscriber) => {
  console.log('Hello')
  subscriber.next(42)
  subscriber.next(100)
  subscriber.next(200)
  setTimeout(() => {
    subscriber.next(300) // happens asynchronously
  }, 1000)
})

console.log('before')
foo.subscribe((x) => {
  console.log(x)
})
console.log('after')

输出

before
Hello
42
100
200
after
300

Observable 向外输出值，除了以上提到的 .next(...) 和 .complete()之外，还可以调用 .error(err)，next 可以调用多次，“错误”和“完成”通知可能只发生一次，并且只能有其中之一
Observable 的订阅和 DOM 的事件监听不同，使用 observable.subscribe，给定的 Observer 不会在 Observable 中注册为侦听器。Observable 甚至不维护附加的观察者列表。调用只是一种 subscribe 启动“可观察执行”并将值或事件传递给该执行的观察者的方法。同一个 Observable 的多个观察者之间不共享的；
Observable.subscribe 会返回一个函数，用来取消订阅

const observable = new Observable(function subscribe(subscriber) {
  const intervalId = setInterval(() => {
    subscriber.next('hi')
  }, 1000)

  return function unsubscribe() {
    clearInterval(intervalId)
  }
})

const subscription = observable.subscribe((x) => console.log(x))
// Later:
subscription.unsubscribe()

Observer

observable.subscribe 方法也可以传入对象，传入函数在内部会被转换为一个对象: { next: fn }，这个对象就是 Observer(观察者)

观察者只是具有三个回调的对象，一个用于 Observable 可能传递的每种类型的通知，Observer 里的回调允许只有部分，比如没有 complete

const observer = {
  next: (x) => console.log('Observer got a next value: ' + x),
  error: (err) => console.error('Observer got an error: ' + err),
  complete: () => console.log('Observer got a complete notification'),
}

observable.subscribe(observer)

Subject

什么是 Subject？

每个 Subject 都是 Observable

你可以对它订阅 subscribe，就像订阅 Observable 一样

每个 Subject 都是 Observer

它是一个具有方法 next(v)、error(e)和 complete()的对象。要为 Subject 提供一个新值，只需调用 next(theValue);

import { Subject } from 'rxjs'

const subject = new Subject<number>()

subject.subscribe({
  next: (v) => console.log(`observerA: ${v}`),
})
subject.subscribe({
  next: (v) => console.log(`observerB: ${v}`),
})

subject.next(1)
subject.next(2)

输出：

observerA: 1
observerB: 1
observerA: 2
observerB: 2

Subject 是一种特殊的 Observable

它允许将值多播到多个 Observer， Subjects 是多播的。像是 DOM 事件监听：它们维护着许多监听器的注册表。

import { Subject, Observable } from 'rxjs'

const subject = new Subject<number>()

subject.subscribe({
  next: (v) => console.log(`observerA: ${v}`),
})
subject.subscribe({
  next: (v) => console.log(`observerB: ${v}`),
})

const observable = new Observable((subscriber) => {
  subscriber.next(1)
  subscriber.next(2)
  subscriber.next(3)
  setTimeout(() => {
    subscriber.next(4)
  }, 1000)
})

// Subject也是Observer，所以可以这样用
observable.subscribe(subject)

subject.subscribe({
  next: (v) => console.log(`observerC: ${v}`),
})

输出：

observerA: 1
observerB: 1
observerA: 2
observerB: 2
observerA: 3
observerB: 3
# 1秒后
observerA: 4
observerB: 4
observerC: 4

运算符

运算符是 rxjs 中非常重要的一部分，它允许以声明方式轻松组合复杂的异步代码。

Before

const observable = new Observable((subscriber) => {
  subscriber.next(1)
  subscriber.next(2)
  subscriber.next(3)
})

After

import { from } from 'rxjs'

const observable = from([1, 2, 3])

再比如：

import { fromEvent } from 'rxjs'

const clicks = fromEvent(document, 'click')
clicks.subscribe((x) => console.log(x))

未提到的

多播操作符、refCount
变体 Subject
调度器
非常多但又非常有用的操作符

学会了我们能做什么

import fetch from 'node-fetch'
import { from, fromEvent, map, merge, mergeMap, switchMap } from 'rxjs'

/**
 * 限制请求并发数量
 */
function multiRequest(urls: string[], maxNum: number) {
  from(urls)
    .pipe(
      mergeMap((v) => fetch(v).then((res) => res.json()), maxNum)
    )
    .subscribe(console.log)
}

// multiRequest(['https://httpbin.org/ip', 'https://httpbin.org/user-agent', 'https://httpbin.org/delay/3'], 2)

/**
 * 竞态
 */
function competitionRequest() {
  const clickA$ = fromEvent(document.getElementById('a')!, 'click').pipe(map(() => 'a'))
  const clickB$ = fromEvent(document.getElementById('b')!, 'click').pipe(map(() => 'b'))

  const click$ = merge(clickA$, clickB$)

  click$.pipe(switchMap((v) => fetch(v).then((res) => res.json()))).subscribe(console.log)
}

与 React 结合

rxjs-hooks

延伸

自己创造运算符
弹珠图
热&冷的 Observables
circleJs - MVI 模式的前端框架

学习操作符的网站

难在哪

rxjs 的概念不复杂，操作符虽然多，但也都是解决具体问题的，容易掌握。我认为难点在于响应式编程的思维装换，比如我们 rxjs 开篇提到的那个场景：

比如说我们有一支付费用户才能播放的视频，首先可能要先抓取这部视频的资讯，接着我们要在播放时去验证使用者是否有权限播放，而使用者也有可能再按下播放后又立即按了取消，而这些都是非同步执行，这时就会各种复杂的状态需要处理。

哪些是流？
怎么组合式声明？
怎么处理竞态？
怎么处理异常？

推荐大家去阅读这篇优秀的文章（github 21k start），是 circleJs 和 stream 的作者写的，原来大佬一开始学也难

响应式编程（Reactive Programming）介绍

Demo

这段时间疫情抢菜，刚好也在学这两个库，随意写了个 demo，用来描述抢菜的流程

xstate 和 rxjs 抢菜 Demo