RxJS入门

一、RxJS是什么？

官方文档使用了一句话总结RxJS: Think of RxJS as Lodash for events。那么Lodash主要解决了什么问题？Lodash主要集成了一系列关于数组、对象、字符串等集合操作,极大的方便了对这些集合数据进行衍生。举个简单的例子:求数组偶数元素的平方和

const { pipe, filter, map, reduce } = require('lodash/fp')
const source = [0, 1, 2, 3, 4]
const result = pipe(
  filter(x => x % 2 === 0),
  map(x => x * x),
  reduce((acc, cur) => acc + cur, 0)
)(source)
console.log(result) // 20

那么如果source中的元素序列是异步产生的呢，如何处理？其中一种解决方案是：Observer Pattern（观察者模式） + Iterator Pattern（迭代器模式）

const event = new (require('events')).EventEmitter()
let count = 0,
  sum = 0
const source = []
const itr = source[Symbol.iterator]()
event.on('pushData', data => {
  source.push(data)
  const { value } = itr.next()
  if (value % 2 === 0) {
    sum += value * value
  }
})

event.on('pushDataComplete', () => {
  console.log(sum) // 20
})
const timer = setInterval(() => {
  if (count > 4) {
    clearInterval(timer)
    event.emit('pushDataComplete')
    return
  }
  event.emit('pushData', count++)
}, 2000)

上述代码有什么问题？——没问题，但是结构松散，不易阅读，不符合函数式编程规范。用RxJS实现则简单的多，代码如下：

const { interval } = require('rxjs')
const { reduce, take, filter, map } = require('rxjs/operators')

const source$ = interval(1000)
const result$ = source$.pipe(
  take(5)
  filter(x => x % 2 === 0),
  map(x => x * x),
  reduce((acc, cur) => acc + cur, 0)
)
result$.subscribe(x => console.log(x))

这段代码与上文中Lodash实现的代码基本一致，唯一不同的是RxJS处理的是异步数据序列，这个异步数据序列在RxJS中被称为流(stream)。RxJS提供了很多操作符，可以对单条数据流进行转化、过滤等操作，也可以对多条数据流进行合并等操作。

二、RxJS数据表示方法

RxJS中表示流的方法是Observable对象，也可以这么说，RxJS就是通过Observable组合各种异步行为的库。RxJS结合了观察者和迭代器模式的思想，可以简单的表示为：

Observable = Publisher + Iterator

下面是一个简单的例子

const { Observable } = require('rxjs')
const onSubscribe = observer => {
  observer.next(0)
  observer.next(1)
  setTimeout(() => {
    observer.next(3)
    observer.complete()
    observer.next(4)
  }, 1000)
  observer.next(2)
}
// 创建流
const source$ = new Observable(onSubscribe)
// 创建观察者
const observer = {
  next: item => console.log(item),
  complete: () => console.log('complete'),
  error: error => console.log(error)
}
// 订阅流
console.log('start')
source$.subscribe(observer)
console.log('end')
// start
// 0
// 1
// 2
// end
// 3
// complete

前面我们说过Observable是Publisher和Iterator的结合,也仅仅是思想上的，实际上还是有很多区别,一般发布-订阅模式会在内部维护一个listeners清单，在要发布通知时会逐一的调用订阅者。但是Observable不是这样的，在其内部并没有一份订阅者的清单。订阅Observable的行为像是执行一个回调方法（onSubscribe），并且这个回调方法是把观察者observer当做参数的，而这里的观察者observer是一个具有三个方法属性的普通对象，观察者的三个方法(method)：

• next：每当Observable吐出新的值，next方法就会被调用。
• complete：在Observable再也没有值吐出时调用，在complete被调用之后，next方法就不会再起作用。
• error：每当Observable内发生错误时，error方法就会被调用。

没有强制要求observer对象必须要具有这三种方法，但至少需要有next方法,除此之外,Observable的subscribe方法还可以直接依次传入next/error/complete方法，其内部会自动组成完整的observer对象。

从上面的例子可以看出RxJS可以同时处理同步和异步行为，Observable可以通过创建时传入的回调onSubscribe方法控制数据吐出的节奏，这种数据流的节奏可以用一个时间轴来表示，在RxJS中被称为弹珠图（Marble Diagrams），上面的例子可以使用下面的弹珠图表示，第一颗弹珠表示同步吐出的0,1,2,第二颗弹珠表示1秒后吐出的3,弹珠上的竖线表示数据流不再产生数据，也就是调用了observer的complete方法.

理解弹珠图的意义的话，可以很容易画出本文第一节中例子对应的弹珠图

const { interval } = require('rxjs')
const { reduce, take, filter, map } = require('rxjs/operators')

const source$ = interval(1000)
const result$ = source$.pipe(
  take(5)
  filter(x => x % 2 === 0),
  map(x => x * x),
  reduce((acc, cur) => acc + cur, 0)
)
result$.subscribe(x => console.log(x))

interval(1000)

take(5)

filter(x => x % 2 === 0)

map(x => x * x)

reduce((acc, cur) => acc + cur, 0)

三、RxJS操作符

如同lodash,RxJS完成复杂异步操作的关键是其实现了大量的操作符,RxJS实现了多达100+的操作符，包括创建类、转换类、过滤类、联合类、工具类等,如上面的例子中，interval属于创建类操作符，它创建了一个Observable对象，作为数据的源头，take和filter属于过滤类操作符，map属于转换类, reduce属于聚合类。实际应用中，我们会花很多时间在操作符的选择上，想熟悉掌握这些操作符不是短期内能完成的，但至少初学者要了解大部分操作符能完成什么样的操作，由于篇幅限制，本文不打算一一介绍所有的操作符，这些操作符可以具体可参考官方文档，后续例子中如果应用到的操作符会着重介绍一下,下面还是借着前面的例子说一下操作符的实现原理，RxJS中大多数操作符都是Pipeable Operators,例子中除了interval以外都是Pipeable Operators,Pipeable Operators本质上是一个纯函数，它将一个Observable作为输入，生成另一个Observable作为输出。订阅输出Observable也将订阅输入Observable。在RxJS中自定义一个操作符非常简单，只需要符合上述指导原则。下面的代码自行实现了例子中所有操作符，看起来一目了然。

const { Observable } = require('rxjs')
const interval = duration =>
  new Observable(observer => {
    let count = 0
    setInterval(() => {
      observer.next(count++)
    }, duration)
  })

const take = num => observable =>
  new Observable(observer => {
    let count = 0
    const subscription = observable.subscribe({
      next(value) {
        if (count <= num) {
          observer.next(value)
          ++count
          if (count === num) {
            observer.complete()
            subscription.unsubscribe()
          }
        }
      },
      error(err) {
        observer.error(err)
      },
      complete() {
        observer.complete()
      }
    })

    return () => {
      subscription.unsubscribe()
    }
  })

const filter = handler => observable =>
  new Observable(observer => {
    const subscription = observable.subscribe({
      next(value) {
        if (handler(value)) {
          observer.next(value)
        }
      },
      error(err) {
        observer.error(err)
      },
      complete() {
        observer.complete()
      }
    })
    return () => {
      subscription.unsubscribe()
    }
  })

const map = handler => observable =>
  new Observable(observer => {
    const subscription = observable.subscribe({
      next(value) {
        observer.next(handler(value))
      },
      error(err) {
        observer.error(err)
      },
      complete() {
        observer.complete()
      }
    })
    return () => {
      subscription.unsubscribe()
    }
  })

const reduce = (handler, seed) => observable =>
  new Observable(observer => {
    const arr = []
    const subscription = observable.subscribe({
      next(value) {
        arr.push(value)
      },
      error(err) {
        observer.error(err)
      },
      complete() {
        seed = arr.reduce(handler, seed)
        observer.next(seed)
        observer.complete()
      }
    })
    return () => {
      subscription.unsubscribe()
    }
  })

const source$ = interval(1000).pipe(
  take(5),
  filter(x => x % 2 === 0),
  map(x => x * x),
  reduce((acc, seed) => acc + seed, 0)
)

source$.subscribe(item => console.log(item), null, () =>
  console.log('complete')
)

四、RxJS与Promise

目前主流的异步解决方案是Promise，Await本质也是Promise,那么RxJS解决方案相比Promise有什么优势呢？
1.Observable可以处理异步事件流，但是Promise只能处理单次事件

const { Observable } = require('rxjs')
const source$ = new Observable(observer => {
  setTimeout(() => observer.next(1), 1000)
  setTimeout(() => observer.next(2), 2000)
  setTimeout(() => observer.next(3), 3000)
  setTimeout(() => observer.complete(), 4000)
})
source$.subscribe(result => console.log(result))

2.Observable是懒执行的(Lazyable),而new Promise(executor)的executor会立即执行

const { Observable } = require('rxjs')

const source$ = new Observable(observer => {
  setTimeout(() => observer.next(1), 1000)
  setTimeout(() => observer.next(2), 2000)
  setTimeout(() => observer.complete(), 3000)
})
setTimeout(() => {
  console.log(3)
  source$.subscribe(result => console.log(result))
}, 3000)
// 3
// 1
// 2

3.Observable 数据是可丢弃的（Cancellable/Abortable）
如前面例子中的take操作符，实际上只取了前5个数据，而丢弃了后面所有的数据，RxJS中还有很多操作符具有类似的性质，如takeUntil( observable ), takeWhile( predicate ), take( n ), first(), first( predicate )从它们的名称和参数就大概能猜到它们的作用。
再比如实际应用中可以会遇到需要丢弃网络请求的结果，如果单纯使用Promise是无法实现的，

const delay = wait => {
  return new Promise(resolve => {
    setTimeout(resolve, wait)
  })
}
delay(3000).then(() => console.log('xxxx'))

上面的Promise无论如何都会打印出xxxx,目前未知ES6规范的Promise仍未实现cancellation，但是使用Observable可以很方便的实现

const { defer } = require('rxjs')
const delay = wait => {
  return new Promise(resolve => {
    setTimeout(resolve, wait)
  })
}
const source$ = defer(() => delay(3000))
const subscription = source$.subscribe(() => console.log('xxxx'))
setTimeout(subscription.unsubscribe.bind(subscription), 2000)

只需要在结果返回前取消订阅就不会打印出结果

4.Observable 是可以重试的(Retryable)

const { Observable } = require('rxjs')
const { retry } = require('rxjs/operators')

const source$ = new Observable(observer => {
  observer.next(1)
  throw 'Error!'
  setTimeout(() => observer.complete(), 4000)
})
source$
  .pipe(retry(3))
  .subscribe(result => console.log(result), err => console.log('Error'))

// 1
// 1
// 1
// 1
// Error

从上面的比较可以看出，RxJS可以处理很多Promise难以处理的场景，而Promise也可以很方便的用过defer操作符转化成Observable.

const { defer } = require('rxjs')
const delay = wait => {
  return new Promise(resolve => {
    setTimeout(resolve, wait)
  })
}
defer(() => delay(3000)).subscribe(() => console.log('xxx'))

五、RxJS使用案列

1. 搜索类问题

搜索类问题是我们实际开发中常遇到的一类问题，如下面的两种场景，上图是查询翻译结果，下图是获取保险报价,这两种场景实际上是一类问题——根据特定的条件查询正确的结果，这类问题在实践的时候需要注意几点：

• 查询请求需要防抖
• 检查查询条件是否合理
• 检查结果是否和条件对应

参照上述三个注意点，我们首先看使用普通JS实现方式.

const input = document.querySelector('#input')
let lastShowedResult = 0
let timer = null
input.addEventListener('change', evt => {
  clearTimeout(timer)
  timer = setTimeout(
    async query => {
      if (query && query.length > 0) {
        const requestTime = +Date.now()
        const data = await fetch(url)
        if (requestTime > lastShowedResult) {
          lastShowedResult = requestTime
          showResult(data)
        }
      }
    },
    500,
    evt.target.value.trim()
  )
})

上面的功能涉及到三个非同步行为:输入框输入、防抖、网络请求，如果使用普通JS实现，可以看到这三种异步行为使用了不同的范式,另外上面的代码还有一个丑陋的地方是使用了很多外部标识，如timer、lastShowedResult。下面是RxJS实现的版本。

import { fromFetch } from "rxjs/fetch"
import {
  debounceTime,
  pluck,
  map,
  filter,
  switchMap
} from "rxjs/operators"

const search$ = fromEvent(document.querySelector('#input'), 'change').pipe(
  debounceTime(500),
  pluck('target', 'value'),
  map(query => query.trim()),
  filter(query => query.length !== 0),
  switchMap(query => fromFetch(`${url}?keyword=${query}`))
)
search$.subscribe(data => {
  showResult(data)
})

上面使用的一些操作符在前面没有提到过，fromEvent是一个创建类操作符，它可以基于给定事件目标的特定类型事件创建一个Observable对象，debounceTime很好理解，只有在特定时间间隔过去源Observable没有吐出下个值，才从源Observable获取一个值，pluck用来获取嵌套熟悉值,fromFetch创建一个网络请求的Observable对象,switchMap是一个关键的操作符，也是比较难以理解的，实际上switchMap 等价于 pipe(map, switchAll),为了便于演示，将上述模型进行一下简化。

const { defer, interval, range } = require('rxjs')
const {
  debounceTime,
  take,
  map,
  concatAll,
  scan,
  mapTo,
  switchMap
} = require('rxjs/operators')
const fetchData = query => {
  return new Promise((resolve, reject) => {
    setTimeout(resolve, (query + 1) * 100, query)
  })
}

const source$ = range(2, 6).pipe(
  map(s => interval(s * 100).pipe(take(1))),
  concatAll(),
  mapTo(1),
  scan((acc, one) => acc + one, 0)
)

const search$ = source$.pipe(
  debounceTime(500),
  map(query => defer(() => fetchData(query))),
  switchAll()
  // switchMap(query => defer(() => fetchData(query))
)

search$.subscribe(data => {
  console.log(data)
})

// 3
// 4
// 6

假设数据不是来源于用户输入而是来源于模拟的source$可观测对象，响应数据会原样返回请求的数据。使用弹珠图分析异步数据的变化。

source$
这里我们不关注构造source$的原理,需要它能够模拟用户输入，产生不同时间间隔的异步数据序列,如下弹珠图所示

debounceTime(500)
500ms内无数据吐出，则释放数据，注意数据6和数据5时间间隔比较近是因为source$吐出数据6后就complete了，后面再无数据了，因此会立即释放

map(query => defer(() => fetchData(query)))
这里map操作符产生的数据也是Observable,因此经过此操作符后产生的Observable是一个高阶Observable

switchAll()
switchAll操作符的作用是将高阶Observable转换成一阶Observable,这个一阶Observable吐出的数据为最新的内层Observable产生的数据

2. 网络请求问题

前面比较Observable和Promise的时候，提到过Observable是可以重试的(Retryable),而支持失败重试可以保证应用的高可用性。
RxJx中实现重试最简单的方法是使用retry操作符

source$.pipe(retry(3)).subscribe((data) => console.log(data))

上述操作会在source$出错时立即重试，最多重试3次,但是在真实的应用中往往由于系统问题，不能即刻恢复正常，解决方案是延时一段时间再重试,借助retryWhen操作符可以实现

source$
  .pipe(retryWhen(err$ => err$.pipe(delay(100))))
  .subscribe(data => console.log(data))

但是不能无限的重试下去，还是需要添加重试上限,借助scan操作符的数据累计功能可以实现

source$
  .pipe(
    retryWhen(
      err$.pipe(
        scan((errorCount, err) => {
          if (errorCount >= 3) {
            throw err
          }
          return errorCount + 1
        }, 0),
        delay(100)
      )
    )
  )
  .subscribe(data => console.log(data))

当访问某个服务器API,第一次失败,可以等100毫秒之后再尝试,结果又失败了,这时候一个比较经验性的做法不是再等100毫秒之后重试,过去的100毫秒服务器没有恢复,那估计再等100毫秒恢复的概率也不高,而且访问太频繁对服务器造成压力也不大好,所以,可以选择200毫秒之后重试,如果再失败,就进一步增加重试延迟,400毫秒之后重试,然后800毫秒后重试,以每次失败选择2ⁿ ×100毫秒的延时,n为失败次数。

source$
  .pipe(
    retryWhen(
      err$.pipe(
        scan((errorCount, err) => {
          if (errorCount >= 3) {
            throw err
          }
          return errorCount + 1
        }, 0),
        delayWhen(errorCount => {
          const delayTime = Math.pow(2, errorCount - 1) * 100
          return timer(delayTime)
        })
      )
    )
  )
  .subscribe(data => console.log(data))

综上我们可以自定义一个重试的操作符

const retryWithExpotentialDelay = (
  maxRetry,
  initialDelay,
  delayFunction
) => source$ => {
  return source$.pipe(
    retryWhen(err$ =>
      err$.pipe(
        scan((errorCount, err) => {
          if (errorCount >= maxRetry) {
            throw err
          }
          return errorCount + 1
        }, 0),
        delayWhen(errorCount => {
          const delayTime = delayFunction(initialDelay, errorCount)
          return timer(delayTime)
        })
      )
    )
  )
}

可以对现有应用进行一些小的改造，将网络请求替换成下面的request方法就能保障应用接口请求的成功率

const request = async (options = {}) => {
  if (!options.url) {
    throw new Error('invalid request options')
  }
  const {
    maxRetry = 2,
    initialDelay = 100,
    delayFunction = (initialDelay, errorCount) =>
      Math.pow(2, errorCount - 1) * initialDelay,
    testResSuccess = rawData => {
      return rawData && rawData.code === 0
    },
    ...restOptions
  } = options
  return new Promise((resovle, reject) => {
    defer(async () => {
      const { data } = await axios(restOptions)
      if (!testResSuccess(data)) {
        return Promise.reject()
      }
      return Promise.resolve(data)
    })
      .pipe(retryWithExpotentialDelay(maxRetry, initialDelay, delayFunction))
      .subscribe(
        data => resovle(data),
        err => {
          reject(err)
        }
      )
  }).catch(e => {
    console.log(e)
  })
}

3. 拖动排序

下面是cms系统中常用的拖拽排序功能，读者可以对照RxJS官方文档进行分析

六、总结

这篇文章仅仅介绍了RxJS的冰山一角，使用的操作符不过十几个，旨在学习RxJS的基本概念和使用场景，还有诸如多播(multicast)、时间调度(Schedule)以及与常用前端技术栈结合等问题都未涉及，文章结尾有一些学习资料/网站/工具可以参考

RxJS
深入浅出RxJS
Thirty-days-RxJS
Learn RxJS
reactive.how
Rx Visualizer