使用 RxJS 掌控异步
3093279013
8年前
<p><img src="https://simg.open-open.com/show/33a1cfc8de65c9ba2e416a9e1f23655c.png"></p> <p>教你使用 RxJS 在 200 行代码内优雅的实现文件分片断点续传。</p> <p>本文是一系列介绍 RxJS 文章的第三篇，这一系列的文章将从一个小的例子开始，逐渐深入的讲解 RxJS 在各种场景下的应用。对应的，也会有对 RxJS 各种操作符的讲解。这篇文章将接着第二篇用 RxJS 连接世界 中的例子，实现一个文件分片断点续传的例子。在例子中，会使用更多操作符(RxJS Operator) 来处理我们的业务，后续的文章中将会详细的讲解这些操作符的作用和使用场景。</p> <h2>Intro</h2> <p><a href="/misc/goto?guid=4959735880308530312" rel="nofollow,noindex">ben lesh </a> 经常在他的各种 talking 中将 RxJS 比作 <em>Lodash for Async</em> 用来彰显 RxJS 的强大异步控制能力，而 RxJS 对于异步而言确实有着媲美 <a href="/misc/goto?guid=4959730793293430913" rel="nofollow,noindex"> lodash </a> 之于 Array 的强大功能。与 lodash 的优秀性能类似，RxJS 在操作异步任务的时的性能也是非常优秀的，并不会因为高等级的抽象而牺牲过多的性能。本篇文章将会以一个相对复杂的异步任务为例子，逐步介绍 RxJS 如何简洁优雅的进行复杂的异步控制。</p> <h2>准备工作</h2> <p>在 <a href="/misc/goto?guid=4959735880437693303" rel="nofollow,noindex"> learning-rxjs </a> clone 项目所需的 seed，并基于 article3-seed checkout 一个你的 article3 分支。本文中所有涉及到 RxJS 的代码将全部使用 TypeScript 编写。</p> <p>这篇文章中，我们将使用 RxJS 实现以下功能:</p> <p>article3-seed 分支附带了一个简单的文件上传的 server，它的作用是实现一个简单的文件分片上传 API。</p> <p>一共有三个 API 提供调用:</p> <ul> <li> <p>post /api/upload/chunk</p> <p>用来获取文件的分片信息，上传文件大小，文件 md5，文件修改时间和文件名</p> <p>服务端返回文件分片数量，每个分片大小和文件唯一的 fileKey</p> <ul> <li> <p>Request body:</p> </li> </ul> <pre> <code class="language-javascript">{ fileSize: string // 文件大小 fileMD5: string // 文件 md5 lastUpdated: ISOString // 文件上次修改时间 fileName: string // 文件名 }</code></pre> <ul> <li> <p>Response:</p> </li> </ul> <pre> <code class="language-javascript">{ chunkSize: number // 每个分片大小 chunks: number // 分片数量 fileKey: string // 唯一文件 id }</code></pre> </li> <li> <p>post /api/upload/chunk/:fileKey?chunk=:chunk&chunks=:chunks</p> <p>用来上传文件分片</p> <ul> <li> <p>Request header: 'Content-Type': 'application/octet-stream'</p> </li> <li> <p>Request body: blob</p> </li> <li> <p>Response: 'ok' or error message</p> </li> </ul> </li> <li> <p>post /api/upload/chunk/:fileKey</p> <p>结算文件分片，后端会将各分片拼接成一个完整的文件并返回</p> <p>Response: 'ok' or error message</p> </li> </ul> <p>在这几个 API 的基础上，我们将在这篇文章中实现以下的功能</p> <ol> <li> <p>在 add 按钮左边增加一个按钮，用来选择一个文件 & (暂停 & 恢复) 文件的上传</p> </li> <li> <p>在增加文件后：</p> <ul> <li> <p>计算文件 md5 ，文件名，文件上次修改时间，文件大小等信息</p> </li> <li> <p>调用 <em>post /api/upload/chunk</em> 接口，获取文件分片信息</p> </li> <li> <p>根据获取的分片信息对文件分片，并且调用 <em>post /api/upload/chunk/:fileKey?chunk=:chunk&chunks=:chunks</em> 上传文件分片，上传过程保证每次只有三个分片同时上传</p> </li> <li> <p>上传完所有的分片后，调用 <em>post /api/upload/chunk/:fileKey</em> 结算文件</p> </li> </ul> </li> <li> <p>上传的过程中，input 框下有一个进度条显示上传进度</p> </li> <li> <p>在上传开始后，选择文件的按钮变成暂停按钮，点击则暂停上传</p> </li> <li> <p>点击暂停上传按钮后，按钮变成继续上传按钮，点击则在暂停的地方继续上传</p> </li> </ol> <p>为了实现上面的功能，并且将这些功能与之前的 todo app 区分开来，我们在 src 文件夹下新建一个 FileUploader.ts 文件并在这个文件中实现这些需求:</p> <pre> <code class="language-javascript">// FileUploader.ts import { Observable } from 'rxjs' // @warn memory leak const $attachment = document.querySelector('.attachment') export class FileUploader { private file$ = Observable.fromEvent($attachment, 'change') .map((r: Event) => (r.target as HTMLInputElement).files[0]) .filter(f => !!f) uploadStream$ = this.file$ }</code></pre> <p>在 html 中加入 attachment 节点：</p> <pre> <code class="language-javascript">// index.html ... <div class="input-group-btn"> <label class="btn btn-default btn-file glyphicon glyphicon-paperclip attachment"> <input type="file" style="display: none;"> </label> <div class="btn btn-default button-add">Add</div> </div> ...</code></pre> <p>调整一下样式:</p> <pre> <code class="language-javascript">// style.css ... .attachment { top: 0; }</code></pre> <p>然后在 app.ts 中将这个我们将要实现功能的流 merge 到 app$ 中:</p> <pre> <code class="language-javascript">... import { FileUploader } from './FileUploader' ... const uploader = new FileUploader() const app$ = toggle$.merge(remove$, search$, uploader.uploadStream$) .do(r => { console.log(r) }) app$.subscribe()</code></pre> <p>这个时候通过 attachment 按钮选择一个文件，就已经可以在控制台中看到从 app$ 中流出的 file了：</p> <p><img src="https://simg.open-open.com/show/3a9be73d9b449c589fcf94b681e8baec.png"></p> <h2>获取文件分片信息</h2> <p>我们使用 FileReader + spark-md5 计算文件的 md5 信息，其它信息直接可以从 File 对象上拿到。而这里的 FileReader 读取文件是一个异步的过程，我们将它封装成 Observable 以便和 uploadStream$ 组合:</p> <pre> <code class="language-javascript">import { Observable, Observer } from 'rxjs' // spark-md5 没有第三方 .d.ts 文件，这里用 commonjs 风格的 require 它 // 如果未再 tsconfig.json 中设置 noImplicitAny: true 且 TypeScript 版本大于 2.1 则也可以用 // import * as SparkMD5 from 'spark-md5' 的方式引用 const SparkMD5 = require('spark-md5') const $attachment = document.querySelector('.attachment') interface FileInfo { fileSize: number fileMD5: string lastUpdated: string fileName: string } export class FileUploader { private file$ = Observable.fromEvent($attachment, 'change') .map((r: Event) => (r.target as HTMLInputElement).files[0]) .filter(f => !!f) uploadStream$ = this.file$ .switchMap(this.readFileInfo) private readFileInfo(file: File): Observable<{ file: File, fileinfo: FileInfo }> { const reader = new FileReader() const spark = new SparkMD5.ArrayBuffer() reader.readAsArrayBuffer(file) return Observable.create((observer: Observer<{ file: File, fileinfo: FileInfo }>) => { reader.onload = (e: Event) => { spark.append((e.target as FileReader).result) const fileMD5 = spark.end() observer.next({ file, fileinfo: { fileMD5, fileSize: file.size, lastUpdated: file.lastModifiedDate.toISOString(), fileName: file.name } }) observer.complete() } return () => { if (!reader.result) { console.warn('read file aborted') reader.abort() } } }) } }</code></pre> <p>此时已经可以看到文件的 FileInfo可以从 app$ 中流出:</p> <p><img src="https://simg.open-open.com/show/22abab58b5ca31d4d1912e363c37aaaf.png"></p> <p>再使用文件信息通过 post /api/upload/chunk 接口获取文件的分片信息:</p> <pre> <code class="language-javascript">... const apiHost = 'http://127.0.0.1:5000/api' ... interface ChunkMeta { fileSize: number chunkSize: number chunks: number fileKey: string } ... export class FileUploader { ... uploadStream$ = this.file$ .switchMap(this.readFileInfo) .switchMap(i => Observable.ajax .post(`${apiHost}/upload/chunk`, i.fileinfo) ) }</code></pre> <p><img src="https://simg.open-open.com/show/c0cd0cb4d4b50f5b7f1568b52c073f5f.png"></p> <h2>分片上传</h2> <p>获取分片信息之后，我们首先要做的事情是将文件按照分片信息分片，做一个 slice 方法来将文件分片:</p> <pre> <code class="language-javascript">... export class FileUploader { ... uploadStream$ = this.file$ .switchMap(this.readFileInfo) .switchMap(i => Observable.ajax .post(`${apiHost}/upload/chunk`, i.fileinfo) .map((r) => { const blobs = this.slice(i.file, r.response.chunks, r.response.chunkSize) return { blobs, chunkMeta: r.response } }) ) ... private slice(file: File, n: number, chunkSize: number): Blob[] { const result: Blob[] = [] for (let i = 0; i < n; i ++) { const startSize = i * chunkSize const slice = file.slice(startSize, i === n - 1 ? startSize + (file.size - startSize) : (i + 1) * chunkSize) result.push(slice) } return result } }</code></pre> <p>这时，我们就能看到分片后的 blobs 和 meta 信息:</p> <p><img src="https://simg.open-open.com/show/0a2ab9c6cb14e04a42e9df783bd59f59.png"></p> <p>将文件切片完成之后，我们需要实现一个上传分片的方法:</p> <pre> <code class="language-javascript">... export class FileUploader { ... uploadStream$ = this.file$ .switchMap(this.readFileInfo) .switchMap(i => Observable.ajax .post(`${apiHost}/upload/chunk`, i.fileinfo) .map((r) => { const blobs = this.slice(i.file, r.response.chunks, r.response.chunkSize) return { blobs, chunkMeta: r.response } }) ) .switchMap(({ blobs, chunkMeta }) => { const dists = blobs.map((blob, index) => this.uploadChunk(chunkMeta, index, blob)) const uploadStream = Observable.from(dists) .mergeAll(this.concurrency) return Observable.forkJoin(uploadStream) .mapTo(chunkMeta) }) constructor( private concurrency = 3 ) { } ... private uploadChunk(meta: ChunkMeta, index: number, blob: Blob) { const host = `${apiHost}/upload/chunk/${meta.fileKey}?chunk=${index + 1}&chunks=${meta.chunks}` return Observable.ajax({ url: host, body: blob, method: 'post', crossDomain: true, headers: { 'Content-Type': 'application/octet-stream' } }) } }</code></pre> <p>这里的 uploadChunk 是上传单个文件分片的方法，uploadStream$ 中最后面一个 <strong>switchMap</strong> 中的逻辑是使用 <strong>mergeAll</strong> 操作符将所有上传的流 <strong>merge</strong> 成一个 Observable，行为就是并发的上传所有的分片。而下面的 forkJoin 操作符则是等 <strong>merge</strong> 之后的 uploadStream <strong>complete</strong> 之后再 emit 一个结果。这里的 <strong>mergeAll</strong> + <strong>forkJoin</strong> 的用法其实与 <strong>Promise.all</strong> 的行为非常类似，这里也可以写成:</p> <pre> <code class="language-javascript">... const dists = blobs.map((blob, index) => this.uploadChunk(chunkMeta, index, blob)) return Observable.forkJoin(... dists) .mapTo(chunkMeta) ...</code></pre> <p>但我们有一个需求是 <em>上传过程保证每次只有三个分片同时上传</em> , 所以需要使用 mergeAll 方法并传入 concurrency = 3 来控制并发数量，现在可以选择一个文件在 Devtool 上观察上传的行为。如果程序没有出问题行为应该是：并发上传文件分片，并且永远只有 3 个分片同时上传，在上传完所有分片后 app$ 中流出 chunkMeta 数据。</p> <p>最后，我们只需要结算这些分片，这个文件就算上传完成了:</p> <pre> <code class="language-javascript">... export class FileUploader { ... uploadStream$ = this.file$ .switchMap(this.readFileInfo) .switchMap(i => Observable.ajax .post(`${apiHost}/upload/chunk`, i.fileinfo) .map((r) => { const blobs = this.slice(i.file, r.response.chunks, r.response.chunkSize) return { blobs, chunkMeta: r.response } }) ) .switchMap(({ blobs, chunkMeta }) => { const dists = blobs.map((blob, index) => this.uploadChunk(chunkMeta, index, blob)) const uploadStream = Observable.from(dists) .mergeAll(this.concurrency) return Observable.forkJoin(uploadStream) .mapTo(chunkMeta) }) .switchMap((r: ChunkMeta) => Observable.ajax.post(`${apiHost}/upload/chunk/${r.fileKey}`) .mapTo({ action: 'UPLOAD_SUCCESS', payload: r }) ) }</code></pre> <p>这时，选择一个文件后，可以看到它被分片上传，并且在结算后在 $app 中发送了一条数据：</p> <pre> <code class="language-javascript">{ "action": "UPLOAD_SUCCESS", "payload": { "chunkSize": 1048576, "chunks": 26, "fileKey": "00a12bdc10449d8ec93883a7d45292a30c", "fileSize": 26621938 } }</code></pre> <p>并且在项目的 chunks 文件夹下面可以找到这个被结算的文件。</p> <h2>进度条</h2> <p>为了实现在界面中实时显示进度条，我们先要在 index.html 中加入进度条标签:</p> <pre> <code class="language-javascript">// index.html ... <div class="progress"> <div class="progress-bar progress-bar-success" role="progressbar" aria-valuenow="0" aria-valuemin="0" aria-valuemax="100" style="width: 0%"> <span>0%</span> </div> </div> ...</code></pre> <p>调整一下样式中文字的颜色:</p> <pre> <code class="language-javascript">// style.css ... .progress-bar > span { color: black; }</code></pre> <p>这个时候界面看起来应该是这样的：</p> <p><img src="https://simg.open-open.com/show/ca206db0435c5e93efed46d12fcbc1ef.png"></p> <p>要获取总体的上传进度，必须先获取单个分片的上传进度，Observable.ajax 有一个方法可以获取 progress:..</p> <pre> <code class="language-javascript">... import { Observable, Observer, Subscriber } from 'rxjs' ... export class FileUploader { ... private uploadChunk(meta: ChunkMeta, index: number, blob: Blob): Observable<ProgressEvent> { const host = `${apiHost}/upload/chunk/${meta.fileKey}?chunk=${index + 1}&chunks=${meta.chunks}` return Observable.create((subscriber: Subscriber<ProgressEvent>) => { const ajax$ = Observable.ajax({ url: host, body: blob, method: 'post', crossDomain: true, headers: { 'Content-Type': 'application/octet-stream' }, progressSubscriber: subscriber }) const subscription = ajax$.subscribe() return () => subscription.unsubscribe() }) } }</code></pre> <p>这样一来我们就可以在 uploadSteram$ 中计算总体的上传进度了:</p> <pre> <code class="language-javascript">... export class FileUploader { uploadStream$ = this.file$ .switchMap(this.readFileInfo) .switchMap(i => Observable.ajax .post(`${apiHost}/upload/chunk`, i.fileinfo) .map((r) => { const blobs = this.slice(i.file, r.response.chunks, r.response.chunkSize) return { blobs, chunkMeta: r.response } }) ) .switchMap(({ blobs, chunkMeta }) => { const uploaded: number[] = [] const dists = blobs.map((blob, index) => { let currentLoaded = 0 return this.uploadChunk(chunkMeta, index, blob) .do(r => { currentLoaded = r.loaded / chunkMeta.fileSize uploaded[index] = currentLoaded const percent = uploaded.reduce((acc, val) => acc + (val ? val : 0)) const p = Math.round(percent * 100) $progressBar.style.width = `${p}%` $progressBar.firstElementChild.textContent = `${p > 1 ? p - 1 : p} %` }) }) const uploadStream = Observable.from(dists) .mergeAll(this.concurrency) return Observable.forkJoin(uploadStream) .mapTo(chunkMeta) }) .switchMap((r: ChunkMeta) => Observable.ajax.post(`${apiHost}/upload/chunk/${r.fileKey}`) .mapTo({ action: 'UPLOAD_SUCCESS', payload: r }) ) .do(() => { $progressBar.firstElementChild.textContent = '100 %' }) }</code></pre> <p>这个时候我们可以在界面中看到文件分片上传的进度了。</p> <p>而一般为了方便使用与调试，我们一般将所有的类似:</p> <pre> <code class="language-javascript">{ action: 'UPLOAD_SUCCESS', payload: { chunkSize: 1048576, chunks: 26, fileKey: "00a12bdc10449d8ec93883a7d45292a30c", fileSize: 26621938 } }</code></pre> <p>的 local state 放在一个流里面:</p> <pre> <code class="language-javascript">import { Observable, Subscriber, Subject } from 'rxjs' ... type Action = 'pause' | 'resume' | 'progress' | 'complete' ... export class FileUploader { ... private action$ = new Subject<{ name: Action payload?: any }>() private progress$ = this.action$ .filter(action => action.name === 'progress') .map(action => action.payload) .do(r => { const percent = Math.round(r * 100) $progressBar.style.width = `${percent}%` $progressBar.firstElementChild.textContent = `${percent > 1 ? percent - 1 : percent} %` }) .map(r => ({ action: 'PROGRESS', payload: r })) uploadStream$ = this.file$ ... return this.uploadChunk(chunkMeta, index, blob) .do(r => { currentLoaded = r.loaded / chunkMeta.fileSize uploaded[index] = currentLoaded const percent = uploaded.reduce((acc, val) => acc + (val ? val : 0)) this.action$.next({ name: 'progress', payload: percent }) }) ... .merge(this.progerss$) }</code></pre> <p>这时控制台会出现更直观的调试信息:</p> <p><img src="https://simg.open-open.com/show/4f5f1eed093a1ab859fd52d6b5221b26.png"></p> <h2>暂停，续传</h2> <p>根据需求，我们在选择文件后，选择文件的按钮将会变成一个暂停按钮，我们可以用 Observable.fromEvent来实现这个需求:</p> <pre> <code class="language-javascript">... export class FileUploader { ... private click$ = Observable.fromEvent($attachment, 'click') .map((e: Event) => e.target) .filter((e: HTMLElement) => e === $attachment) .scan((acc: number, val: HTMLElement) => { if (val.classList.contains('glyphicon-paperclip')) { return 1 } if (acc === 2) { return 3 } return 2 }, 3) .filter(v => v !== 1) .do((v) => { if (v === 2) { this.action$.next({ name: 'pause' }) $attachment.classList.remove('glyphicon-pause') $attachment.classList.add('glyphicon-play') } else { this.action$.next({ name: 'resume' }) this.buildPauseIcon() } }) uploadStream$ = this.file$ .switchMap... .switchMap... .do(() => this.buildPauseIcon()) ... .do(() => { $progressBar.firstElementChild.textContent = '100 %' // restore icon $attachment.classList.remove('glyphicon-pause') $attachment.classList.add('glyphicon-paperclip'); ($attachment.firstElementChild as HTMLInputElement).disabled = false }) .merge(this.progress$, this.click$) // side effect private buildPauseIcon() { $attachment.classList.remove('glyphicon-paperclip') $attachment.classList.add('glyphicon-pause'); ($attachment.firstElementChild as HTMLInputElement).disabled = true } }</code></pre> <p>这段代码用到涉及到的概念比较多，我们一点点来理解：</p> <p>在 uploadStream$ 的两个 switchMap 下插入了一个 do 操作符，这段代码的作用是将文件上传的图标变成暂停的图标。</p> <p>然后我们新建了一个 click$ 流，为了防止事件冒泡导致的重复推送值，我们用 map + filter 过滤掉了子节点冒泡上来的事件。而为了区分点击的是 上传文件按钮还是 暂停按钮还是 继续按钮，我们用 1，2，3 三个值代表三个不同的点击事件，并使用 scan操作符不停的生成这三个状态。scan 的行为与 Array#reduce 非常相似，它接受一个 accumulator 不停的根据当前的值和状态累加出新的状态(没错，和 Redux 中的 reducer 行为一致)。而在下面的 do 操作符中我们根据不同的状态改变按钮的icon 。</p> <p>这个时候我们观察上传的过程中，点击暂停/继续，图标的状态可以正确切换了。并且在上传完成后图标也被恢复成上传文件的初始状态了。</p> <p><img src="https://simg.open-open.com/show/649d577e4c54730264bd6c5417a5da9d.jpg"></p> <p>为了让整个文件上传可以暂停与继续，我们在 uploadChunk 下使用 takeUntil 与 repeatWhen & retryWhen 操作符:</p> <pre> <code class="language-javascript">... export class FileUploader { ... private action$ = ... private pause$ = this.action$.filter(ac => ac.name === 'pause') private resume$ = this.action$.filter(ac => ac.name === 'resume') private progress$ = this.action$ ... .distinctUntilChanged((x: number, y: number) => x - y >= 0) ... ... private uploadChunk(meta: ChunkMeta, index: number, blob: Blob): Observable<ProgressEvent> { ... return Observable.create( ... const ajax$ = Observable.ajax({ ... }) .takeUntil(this.pause$) .repeatWhen(() => this.resume$) const subscription = ajax$.subscribe() return () => subscription.unsubscribe() ) .retryWhen(() => this.resume$) } }</code></pre> <p><strong>takeUntil</strong> 操作符接受一个 <strong>Observable</strong> ，它在这个 <strong>Observable</strong> 发射值的时候终止上面的 <strong>Observable</strong></p> <p><strong>repeatWhen</strong> 与 <strong>retryWhen</strong> 操作符都是接受一个 <em>projectFunc</em> ，它返回一个 <strong>Observable</strong> 并在这个 <strong>Observable</strong> 发射值的时候 重复/重试。</p> <p>而在暂停恢复的过程中，进度条的数字可能显示错误：上传了一部分的请求被 abort，它的 progress 已经计算过一次了，重试的时候是重新上传，则可能会导致进度条后退，这时我们在 progress$ 后面用 <strong>distinctUntilChanged</strong> 方法即可实现 <strong> <em>只有在进度增长的时候发射值</em> </strong> 这一效果。</p> <h2>结语</h2> <p>这是一篇超级抽象的文章，并且受限于未使用框架，在程序中使用了大量的副作用操作do，总体看起来并没有特别优雅。真正优雅的 FRP 应该是将 RxJS 与 Redux + React 这样的框架结合起来，那时这个文件上传的组件就可以有更优雅的写法。当然它的功能并不完备，很多 edge case 例如各个步骤中的异常处理都没有做，但没有关系，这里只起到一个示范作用来展示 RxJS 在处理异步上的强大功能，并且让初学者有机会亲手把玩 RxJS 的各种操作符并实现一个复杂的异步场景。在后面的文章中，将会深入前面这三篇文章中涉及到或未涉及到的各种操作符，逐渐拨开 RxJS 的迷雾。</p> <p> </p> <p>来自：https://zhuanlan.zhihu.com/p/25059824</p> <p> </p>