探索 RxJS - 核心概念

SGAFleta 7年前
   <p>Table of Contents generated withDocToc</p>    <ul>     <li>探索 RxJS - Core Concept</li>    </ul>    <h2><strong>探索 RxJS - Core Concept</strong></h2>    <h3><strong>Steam in ReactiveX</strong></h3>    <p>ReactiveX,又称 Reactive Extensions(响应式扩展),其中的 X 代表各种语言。因为它实质上只是一个事件流处理库,不需要什么其他依赖。</p>    <p>讲一讲 ReactiveX 中的 “流”:</p>    <p>我的理解是,Rx 通过 “流” 的概念,将事件串联成一个个事件流,各个事件流之间还可进行 "并联" 的作用。当某个流上的事件被调用时,就可以触发我们设定好的监听回调。</p>    <p>那么什么样的事件可以成为流呢?答案是任何事件。无论是异步非阻塞事件(setTimeOut、网络请求等),还是同步可阻塞事件(点击事件、对迭代器的遍历等),一切都是流。此时不得不祭出一张神棍图:</p>    <p style="text-align:center"><img src="https://simg.open-open.com/show/3e948faedd76cb8c56186f166c52b28f.jpg"></p>    <p>事件的串联是流。比方说,用户对一个按钮进行了猛烈的点击,所有的点击事件就是一个流;再或者,并发多个网络请求,它们也是一个流。而 Rx 的主要作用,就是为流的处理提供了一整套的解决方案:将不同的流进行组合,或者监听事件的触发及时给予响应等等。</p>    <h3><strong>Quick Intro</strong></h3>    <p>RxJS 的作者曾在 The introduction to Reactive Programming you've been missing 一文中详细讲解了 RxJS 的初步使用和流的概念,我们先仅看文中的一幅图来理解 Rx 的概念:</p>    <p style="text-align:center"><img src="https://simg.open-open.com/show/cc507c7a91ab2865fbce4764f291c237.png"></p>    <p>这是一个多次点击事件形成的事件流,在从左往右的时间线上有很多个点击事件,而每个点击事件的时间间隔各不相同。通过 Rx 我们可以对流上的各个事件进行筛选,并获取到在某一段时间内的连续点击次数。</p>    <p>我们在页面上的点击事件就可以组成流。比如一个记录每次点击时坐标的流。随机点击页面多次之后,可能会产生这样一个流:</p>    <p style="text-align:center"><img src="https://simg.open-open.com/show/eb84d138388266928461145d91c0cd6e.png"></p>    <p>而我们可以通过 RxJS 中的方法对这个流内的各个事件进行筛选,比如选出横坐标 x < 250 的点击:</p>    <pre>  <code class="language-javascript">filter( (event) -> event.x < 250 )</code></pre>    <p style="text-align:center"><img src="https://simg.open-open.com/show/3d455084964228c5d9b339c283edcb74.png"></p>    <p>也就是说,我们可以像对待 JavaScript 中可遍历对象一样,对流上的各个事件进行遍历,选出符合条件的事件。这就是 Rx 的魅力所在。</p>    <h3><strong>主要运用场景</strong></h3>    <p>既然 Rx 是为了流而生的,那么最佳运用场景当然是面对一系列较复杂的事件流时了。</p>    <p>含有异步请求和事件触发的混合流</p>    <p>比如,用户在一个 input 内输入文字。每次 keyup 的时候就会根据输入内容,请求 Wikipedia 的 API 进行搜索:</p>    <pre>  <code class="language-javascript">var input = document.getElementById('input');  // 通过 fromEvent 以及 input keyup 事件创建一个流  var dictionarySuggest = Rx.Observable.fromEvent(input, 'keyup')    // 获取到每次 keyup 时的input value,并通过 filter 保证其合法性    .map(function () { return input.value; })    .filter(function (text) { return !!text; })    .distinctUntilChanged()    .debounce(250)    // searchWikipedia 为异步请求方法    .flatMapLatest(searchWikipedia)    .subscribe(      // onNext      function (results) {        list = [];        list.concat(results.map(createItem));      },      // onError      function (err) {        logError(err);      }    );</code></pre>    <p>我们创建了一个流来处理从用户 keyup ,到 searchWikipedia ,再到处理网络请求结果这一系列事件,并且在其中对事件进行了筛选判断:</p>    <ul>     <li>filter 剔除掉不合法的值</li>     <li>distinctUntilChanged 当用户按下例如 左、右 这种按钮时,不会改变 input 的值,但也会触发 keyup 事件。这种时候就完全没有必要重复发送异步请求了。 distinctUntilChanged 会剔除流中有着相同的值的元素</li>     <li>debounce 在过了一段指定的时间还没触发事件时才触发下一个事件。也就是说,在打字过程中,如果用户在指定事件间隔(250ms)内没有再打字,则触发下一个事件(searchWikipedia);否则我们认为用户在连续打字,所以不会频繁的发送网络请求</li>     <li>flatMapLatest      <ul>       <li>首先,它是一个 flatMap 方法。它用一个指定的函数(searchWikipedia)对原始流中的每一项数据执行变换操作,并返回一个 Observable , flatMap 将所有的返回值组成一个新的流。</li>       <li>其次, flatMapLatest 拥有 flatMap 的全部特性。但不同的是,在 flatMapLatest 的遍历调用过程中,如果一个事件 A 还没有触发完毕获取到返回值,就触发了下一个事件 B,则将忽略 A 返回的值。这样,我们就可以 避免 A 异步的返回值因为返回较慢,反而覆盖了之后 B 异步的返回值 。用图解释如下:</li>      </ul> </li>    </ul>    <p style="text-align:center"><img src="https://simg.open-open.com/show/8657a6340ccda34099a8cb3708cc02f1.png"></p>    <ul>     <li>subscribe 创建对流的监听,并提供了成功和失败的回调</li>    </ul>    <p>而在传统的编写方法里,我们可能会创建 input 的 keyup 监听事件,并缓存上一次的值;每次 keyup 时,要判断当前值是否合法,并且与上一次的值不一样。除此以外,还要创建一个定时器,每隔一段时间就用合法的值去请求 searchWikipedia 方法 --- 即便这样,也无法保证不在用户连续打字时发送请求。</p>    <p>可以看到,在我们把事件串成流并进行处理之后,要比传统的编写方式方便很多。</p>    <p>处理一系列的异步请求队列</p>    <p>假设我们要读取一个 4GB 的大文件,将其加密后写入到一个新文件里。直接将整个文件读到内存里再加密、写入肯定是不行的,反之,我们依赖 RxJS 的流,创建多个读取、加密、写入事件,形成三个流出来:</p>    <ul>     <li>文件读取流:每次调用方法时异步读取 64k 的文件</li>     <li>加密流:对读取的文件进行加密</li>     <li>写入流:将加密好的内容异步写入新文件</li>     <li>最后对整个 observable 进行监听</li>    </ul>    <pre>  <code class="language-javascript">var fs = require('fs');  var Rx = require('rx');    // Read/write from stream implementation  function readAsync(fd, chunkSize) { /* impl */ }  function appendAsync(fd, buffer) { /* impl */ }  function encrypt(buffer) { /* impl */}    // 打开一个 4GB 的文件,每次只读取 64k  var inFile = fs.openSync('4GBfile.txt', 'r+');  var outFile = fs.openSync('Encrypted.txt', 'w+');    readAsync(inFile, 2 << 15)    .map(encrypt)    .flatMap(function (data) {      return appendAsync(outFile, data);    })    .subscribe(      // onNext      function () { },      // onError      function (err) {        console.log('An error occurred while encrypting the file: %s', err.message);        fs.closeSync(inFile);        fs.closeSync(outFile);      },      // onCompleted      function () {        console.log('Successfully encrypted the file.');        fs.closeSync(inFile);        fs.closeSync(outFile);      }    );</code></pre>    <p>由此可以看出,在应对较复杂的事件流或者处理多个异步事件的时候,使用 RxJS 会有一定优势;但如果复杂度没有这么高的时候则没有太大的使用必要。</p>    <p>目前为止,本文基本介绍了 RxJS 的核心概念 --- 针对事件流的管理与掌控。</p>    <p> </p>    <p> </p>    <p>来自:https://github.com/ecmadao/Coding-Guide/blob/master/Notes/RxJS/探索RxJS-CoreConcept.md</p>    <p> </p>