👍 實用rxjs學習案例

前言：

我搜集了網絡和本身實踐中的一些案例，讓你們感覺一下rxjs處理異步時的優點。javascript

本文主要目的：php

一、讓一些同窗對rxjs對稍微複雜異步的簡潔處理感興趣（不須要懂api，僅僅感覺rxjs的優點）前端

二、讓一些缺少實戰案例的同窗練手（網上多講api的，跟業務相關的案例較少，這跟rxjs自己流行程度並不高有關）java

三、本身初步學習後的總結web

從一個字節面試題開始：控制最大併發數

題目以下：面試

實現一個批量請求函數 multiRequest(urls, maxNum)，要求以下：
• 要求最大併發數 maxNum
• 每當有一個請求返回，就留下一個空位，能夠增長新的請求
• 全部請求完成後，結果按照 urls 裏面的順序依次打出
複製代碼

你能夠先想一想若是不用rxjs，你會怎麼作，咱們先看用rxjs有多簡單ajax

// 假設這是你的http請求函數
function httpGet(url) {
  return new Promise(resolve => setTimeout(() => resolve(`Result: ${url}`), 2000));
}
複製代碼

使用rxjs只須要1行代碼就能夠解決這個面試題，後面會寫一版不用rxjs，能夠看看promise實現有多麼麻煩。編程

const array = [
  'https://httpbin.org/ip', 
  'https://httpbin.org/user-agent',
  'https://httpbin.org/delay/3',
];

// mergeMap是專門用來處理併發處理的rxjs操做符
// mergeMap第二個參數2的意思是，from(array)每次併發量是2，只有promise執行結束才接着取array裏面的數據
// mergeMap第一個參數httpGet的意思是每次併發，從from(array)中取的數據如何包裝，這裏是做爲httpGet的參數
const source = from(array).pipe(mergeMap(httpGet, 2)).subscribe(val => console.log(val));
複製代碼

在線代碼預覽：stackblitz.com/edit/rxjs-q… （注意控制檯打印順序）json

如下是promise的版本，代碼多並且是面向過程的麪條代碼（若是不用rxjs的化，通常場景建議使用ramda庫，用「流」或者函數組合的方式來編寫函數，讓你的功能模塊遠離麪條代碼（麪條代碼 = 難以維護的面向過程的代碼）），文章最後閒聊會講業務不復雜的場景，怎麼使用ramda。redux

如下是用promise解決上述面試題的思路，能夠看到大量的臨時變量，while函數，if語句，讓代碼變得難以維護（並非拒絕這種代碼，畢竟優雅的接口後面極可能是「齷齪的實現」),但若是有工具幫助你直接使用優雅的接口，下降了複雜度，何樂而不用呢

function multiRequest(urls = [], maxNum) {
  // 請求總數量
  const len = urls.length;
  // 根據請求數量建立一個數組來保存請求的結果
  const result = new Array(len).fill(false);
  // 當前完成的數量
  let count = 0;
 
  return new Promise((resolve, reject) => {
    // 請求maxNum個
    while (count < maxNum) {
      next();
    }
    function next() {
      let current = count++;
      // 處理邊界條件
      if (current >= len) {
        // 請求所有完成就將promise置爲成功狀態, 而後將result做爲promise值返回
        !result.includes(false) && resolve(result);
        return;
      }
      const url = urls[current];
      console.log(`開始 ${current}`, new Date().toLocaleString());
      fetch(url)
        .then((res) => {
          // 保存請求結果
          result[current] = res;
          console.log(`完成 ${current}`, new Date().toLocaleString());
          // 請求沒有所有完成, 就遞歸
          if (current < len) {
            next();
          }
        })
        .catch((err) => {
          console.log(`結束 ${current}`, new Date().toLocaleString());
          result[current] = err;
          // 請求沒有所有完成, 就遞歸
          if (current < len) {
            next();
          }
        });
    }
  });
}
複製代碼

咱們再來一個面試題，這是我本身面騰訊的時候本身遇到的關於ajax請求併發的問題，當時剛轉前端回答的不是很好。題目以下：

再進一步說明問題

按鈕A按了以後，ajax請求的數據顯示在input type=text框裏，B按鈕也是。

問題就是若是先按A，此時ajax發出去了，可是數據還沒返回來, 咱們等不及了，立刻按B按鈕，結果此時A按鈕請求的數據先回來，這就尷尬了，按的B按鈕，結果先顯示A按鈕返回的數據，怎麼解決？

這個問題能夠在在A按鈕按了以後，再按B按鈕的時候，取消a按鈕發出的請求，這個ajax和fetch都是有方法實現的，ajax原生自帶cancel方法，fetch的話要本身寫一下，大概思路以下（如何取消fetch）

function abortableFetch(request, opts) {
  const controller = new AbortController();
  const signal = controller.signal;

  return {
    abort: () => controller.abort(),
    ready: fetch(request, { ...opts, signal })
  };
}
複製代碼

別看上面封裝的挺不錯的，可是用起來仍是有點麻煩，並且耦合性有點高，由於我要在B按鈕的onClick事件裏面去調用A按鈕的abort方法。

好了，咱們基於rxjs來寫一個通用的處理方案（要說函數間的解耦，發佈訂閱模式有點萬能的感受，rxjs的new Subject也是同樣的思想）

import { Subject } from 'rxjs';
import { switchMap } from 'rxjs/operators';

// 假設這是你的http請求函數
function httpGet(url: any): any {
  return new Promise(resolve =>
    setTimeout(() => resolve(`Result: ${url}`), 2000)
  );
}

class abortableFetch {
  search: Subject<any>;
  constructor() {
    this.search = new Subject();
    this.init();
  }
  init() {
    this.search
      .pipe((switchMap as any)((value: any): any => httpGet(value)))
      .subscribe(val => console.log(val));
  }

  trigger(value) {
    this.search.next(value);
  }
}

// 使用方式，很是簡單，就一個trigger方法就能夠了
const switchFetch = new abortableFetch();

switchFetch.trigger(123);
setTimeout(() => {
  switchFetch.trigger(456);
}, 1000);

複製代碼

請注意此案例控制檯輸出的是456而不是123，由於456後輸出把以前的123覆蓋了，至關於取消了以前的請求

在線預覽此案例： stackblitz.com/edit/rxjs-z…

好了，上面兩個例子能夠看出rxjs最大的優勢：

一、函數式編程在寫一些小功能的時候，解耦很是簡單，自然知足高內聚、低耦合

二、rxjs在處理異步（好比網絡IO和UI交互）時，寫一些小功能時，代碼量較少，語義性很強

可是問題也很突出，就是掌握好rxjs，真的不容易，都不說rxjs了，用ramda庫或函數式變成的庫的前端在我經歷裏都不多。

接下來，下面就是基於rxjs的一些案例。

一、buffer相關操做符案例

bufferTime：好比你寫一個基於 websocket 的在線聊天室，不可能每次 ws 收到新消息，都馬上渲染出來，這樣在不少人同時說話的時候，通常會有渲染性能問題。。

因此你須要收集一段時間的消息，而後把它們一塊兒渲染出來，例如每一秒批量渲染一次。用原生 JS 寫的話，你須要維護一個隊列池，和一個定時器，收到消息，先放進隊列池，而後定時器負責把消息渲染出來，相似：

let messagePool = []
ws.on('message', (message) => {
    messagePool.push(message)
})

setInterval(() => {
    render(messagePool)
    messagePool = []
}, 1000)
複製代碼

這裏已是最簡化的代碼了，但邏輯依然很破碎，而且還要考慮清理定時器的問題。若是用 RxJS，代碼就好看了不少

import { fromEvent } from 'rxjs';
import { switchMap } from 'rxjs/operators';
 fromEvent(ws, 'message')
     .pipe(bufferTime(1000))
    .subscribe(messages => render(messages))
複製代碼

記錄鼠標兩秒能點擊多少次

fromEvent(document,'click').pipe(
    bufferTime(2000),
    map(array=>array.length)
).subscribe(count => {
    console.log("兩秒內點擊次數", count);
  });
複製代碼

bufferCount: 另一個例子，好比咱們在寫一個遊戲，當用戶連續輸入"上上下下左右左右BABA"的時候，就彈出隱藏的彩蛋，用原生 JS 的話也是須要維護一個隊列，隊列中放入最近12次用戶的輸入。而後每次按鍵的時候，都識別是否觸發了彩蛋。RxJS 的話就簡化了不少，主要是少了維護隊列的邏輯：

const code = [
   "ArrowUp",
   "ArrowUp",
   "ArrowDown",
   "ArrowDown",
   "ArrowLeft",
   "ArrowRight",
   "ArrowLeft",
   "ArrowRight",
   "KeyB",
   "KeyA",
   "KeyB",
   "KeyA"
]

fromEvent(document, 'keyup').pipe(
   map(e => e.code),
   bufferCount(12, 1)
).subscribe(last12key => {
        if (_.isEqual(last12key, code)) {
            console.log('隱藏的彩蛋 \(^o^)/~')
        }
    })
複製代碼

固然 RxJS 還能夠複雜得多的邏輯，好比要求只有在兩秒內連續輸入祕籍，才能觸發彩蛋，這裏該怎麼寫

import { fromEvent } from 'rxjs';
import { bufferCount, map, auditTime } from 'rxjs/operators';

const code = ['KeyA', 'KeyB', 'KeyA'];

fromEvent(document, 'keyup')
  .pipe(
    map(e => (e as any).code),
    bufferCount(3, 1),
    auditTime(2000)
  )
  .subscribe(last3key => {
    if (_.isEqual(last3key, code)) {
        console.log('隱藏的彩蛋 \(^o^)/~')
    }
  });

複製代碼

二、簡易拖拉

實現內容以下：

一、首先頁面上有一個元素(#drag)

二、當鼠標在元素(#drag)上按下左鍵(mousedown)時，開始監聽鼠標移動(mousemove)的位置

三、當鼠標左鍵釋放（mouseup）時，結束監聽鼠標的移動

四、當鼠標移動被監聽時，跟着修改原件的樣式屬性

import { of, fromEvent} from 'rxjs'; 
import { map, concatMap, takeUntil, withLatestFrom } from 'rxjs/operators';

 // 樣式省略是絕對定位
const dragEle = document.getElementById('drag')
const mouseDown = fromEvent(dragEle, 'mousedown')
const mouseUp = fromEvent(document, 'mouseup')
const mouseMove = fromEvent(document, 'mousemove')

mouseDown.pipe(
  concatMap(e => mouseMove.pipe(takeUntil(mouseUp))),
  withLatestFrom(mouseDown, (move: MouseEvent, down: MouseEvent) => {
        return {
            x: move.clientX - down.offsetX,
            y: move.clientY - down.offsetY
        }
  })
).subscribe(pos => {
        dragEle.style.top = pos.y + 'px';
        dragEle.style.left = pos.x + 'px';
})
複製代碼

在線預覽：stackblitz.com/edit/rxjs-s…

三、簡易autoComponent功能

實現內容以下：

一、準備 input#search 以及 ul#suggest-list 的 HTML 與 CSS

二、在 input#search 輸入文字時，等待 100 毫秒後若無輸入，就發送 HTTP Request

三、當 Response 還沒回來時，使用者又輸入了下一哥文字就捨棄前一次的，並再發送一次新的 Request

四、接受到 Response 以後顯示下拉選項

五、鼠標左鍵選中對應的下拉響，取代 input#search 的文字

import { fromEvent } from "rxjs";
import { map, debounceTime, switchMap } from "rxjs/operators";

const url = 'https://zh.wikipedia.org/w/api.php?action=opensearch&format=json&limit=5&origin=*';

const getSuggestList = (keyword) => fetch(url + '&search=' + keyword, { method: 'GET', mode: 'cors' })
                                    .then(res => res.json())

const searchInput = document.getElementById('search');
const suggestList = document.getElementById('suggest-list');

const keyword = fromEvent(searchInput, 'input');
const selectItem = fromEvent(suggestList, 'click');

const render = (suggestArr = []) => suggestList.innerHTML = suggestArr.map(item => '<li>'+ item +'</li>').join('')

keyword.pipe(
  debounceTime(100),
  switchMap(
    (e: any) => getSuggestList(e.target.value),
    (e, res) => res[1]
  )
).subscribe(list => render(list))
  
 

selectItem.pipe(
  map(e => e.target.innerText)
).subscribe(text => { 
      searchInput.value = text;
      render();
  })
複製代碼

在線預覽：stackblitz.com/edit/rxjs-x…

好了，介紹到這裏，我我的的感受是rxjs在處理網絡層和ui層的邏輯時，在某些特定場景會很是簡單。我在此推薦兩個很是好的教程，我看到網上竟然沒人推薦這兩個rxjs學習的教程（上面有個別案例就是今後來的）。

打通rxjs任督二脈： ithelp.ithome.com.tw/users/20020…

30天精通rxjs： ithelp.ithome.com.tw/articles/10…

文章最後，安利另外一個函數式編程的庫ramdajs，rxjs屬於最近才學的，還沒用到項目中

ramdajs是本身用了大概3個月了，有一些心得，確實寫了以後代碼的可維護性變的高不少，緣由就是你必須遵照設計模式裏的單一職責原則，全部功能能複用的函數都會提取出來（用函數式編程會強行讓你養成這個習慣）

附一段我本身項目裏ramdajs的代碼，完畢。

// 這個函數的意思是，在函數鏈裏面，若是其中一個函數返回是null或者undefined就終止函數鏈
const pipeWhileNotNil = R.pipeWith((f, res) =>
  R.isNil(res) ? res : f(res),
);

pipeWhileNotNil([
  // checkData用表單校驗用的，若是不經過返回null，這個函數鏈條就終止，不會往下走
  // R__是佔位符，被R.curry函數柯里化以後，就能夠用R.__充當你函數參數的佔位符
  // 數組裏的參數不用管，是業務上須要自定義的
  checkData(R.__, ['sdExchangeRate', 'sdEffectDate']),
  // 此函數用來篩選,至關於數組的find方法，format2YYYYMMDD是dayjs用來格式化日期的
  R.find(
      (v) =>
        v?.effectDate === format2YYYYMMDD(sdEffectDate),
  ),
  // pipeP是promise函數流的方法，第一個參數必須是promise函數，後面的函數至關於promise裏的then裏面的函數
  R.pipeP(
    // 上一個函數執行結果會傳給existEqualEffectDate
    // promiseModal是一個彈框組件，詢問是否要繼續某個操做
    async (existEqualEffectDate) => {
      if (existEqualEffectDate) {
        return await promiseModal({
          title: `生效日期已存在，是否直接修改匯率？`,
        });
      } else {
        return await promiseModal({
          title: `保存後生效日期不可修改，肯定保存？`,
        });
      }
    },
    // 最後根據上一個返回的結果是ture仍是false進行最後的操做，R.pipe是同步函數鏈方法，dispatch是redux裏的dispatch
    (isGo) => isGo ? R.pipe(R.tail, saveData(dispatch))(data) : dispatch({
      type: 'currencyAndExchange/getExchangePairList',
    });
  ),
])(record);

複製代碼