RxJS入门 (4)

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

3456

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

3

4

5

6

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

346

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毫秒后重试,以每次失败选择2n ×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