在最近的業(yè)務(wù)中，接到了一個需要處理約十萬條數(shù)據(jù)的需求。這些數(shù)據(jù)都以字符串的形式給到，并且處理它們的步驟是異步且耗時的（平均處理一條數(shù)據(jù)需要 25s 的時間）。如果以串行的方式實(shí)現(xiàn)，其耗時是相當(dāng)長的：

總耗時時間 = 數(shù)據(jù)量 × 單條數(shù)據(jù)處理時間

T = N * t (N = 100,000; t = 25s)

總耗時時間 = 2,500,000 秒 ≈ 695 小時 ≈ 29 天

顯然，我們不能簡單地把數(shù)據(jù)一條一條地處理。那么有沒有辦法能夠減少處理的時間呢？經(jīng)過調(diào)研后發(fā)現(xiàn)，使用異步任務(wù)隊列是個不錯的辦法。

一、異步任務(wù)隊列原理

我們可以把“處理單條數(shù)據(jù)”理解為一個異步任務(wù)，因此對這十萬條數(shù)據(jù)的處理，就可以轉(zhuǎn)化成有十萬個異步任務(wù)等待進(jìn)行。我們可以把這十萬條數(shù)據(jù)塞到一個隊列里面，讓任務(wù)處理器自發(fā)地從隊列里面去取得并完成。

任務(wù)處理器可以有多個，它們同時從隊列里面把任務(wù)取走并處理。當(dāng)任務(wù)隊列為空，表示所有任務(wù)已經(jīng)被認(rèn)領(lǐng)完；當(dāng)所有任務(wù)處理器完成任務(wù)，則表示所有任務(wù)已經(jīng)被處理完。

其基本原理如下圖所示：

首先來解決任務(wù)隊列的問題。在這個需求中，任務(wù)隊列里面的每一個任務(wù)，都包含了待處理的數(shù)據(jù)，數(shù)據(jù)以字符串的形式存在。為了方便起見，我們可以使用 Redis 的 List 數(shù)據(jù)格式來存放這些任務(wù)。

由于項目是基于 NodeJS 的，我們可以利用 PM2 的 Cluster 模式來啟動多個任務(wù)處理器，并行地處理任務(wù)。以一個 8 核的 CPU 為例，如果完全開啟了多進(jìn)程，其理論處理時間將提升 8 倍，從 29 天縮短到 3.6 天。

接下來，我們會從實(shí)際編碼的角度來講解上述內(nèi)容的實(shí)現(xiàn)過程。

二、使用 NodeJS 操作 Redis

異步任務(wù)隊列使用 Redis 來實(shí)現(xiàn)，因此我們需要部署一個單獨(dú)的 Redis 服務(wù)。在本地開發(fā)中為了快速完成 Redis 的安裝，我使用了 Docker 的辦法（默認(rèn)機(jī)器已經(jīng)安裝了 Docker）。

Docker 拉取 Redis 鏡像   

docker pull redis:latest 

Docker 啟動 Redis   

docker run -itd --name redis-local-p 6379:6379 redis 

此時我們已經(jīng)使用 Docker 啟動了一個 Redis 服務(wù)，其對外的 IP 及端口為 127.0.0.1:6379。此外，我們還可以在本地安裝一個名為 Another Redis DeskTop Manager的 Redis 可視化工具，來實(shí)時查看、修改 Redis 的內(nèi)容。

在 NodeJS 中，我們可以使用 node-redis 來操作 Redis。新建一個 mqclient.ts 文件并寫入如下內(nèi)容： 

import* asRedisfrom'redis'  
const client = Redis.createClient({  
  host: '127.0.0.1',  
  port: 6379  
})  
exportdefault client 

Redis 本質(zhì)上是一個數(shù)據(jù)庫，而我們對數(shù)據(jù)庫的操作無非就是增刪改查。node-redis 支持 Redis 的所有交互操作方式，但是操作結(jié)果默認(rèn)是以回調(diào)函數(shù)的形式返回。為了能夠使用 async/await，我們可以新建一個 utils.ts 文件，把 node-redis 操作 Redis 的各種操作都封裝成 Promise 的形式，方便我們后續(xù)使用。   

import client from'./mqClient'  
   // 獲取 Redis 中某個 key 的內(nèi)容  
   exportconst getRedisValue = (key: string): Promise<string| null> => newPromise(resolve => client.get(key, (err, reply) => resolve(reply)))  
   // 設(shè)置 Redis 中某個 key 的內(nèi)容  
   exportconst setRedisValue = (key: string, value: string) => newPromise(resolve => client.set(key, value, resolve))  
   // 刪除 Redis 中某個 key 及其內(nèi)容  
   exportconst delRedisKey = (key: string) => newPromise(resolve => client.del(key, resolve)) 

除此之外，還能在 utils.ts 中放置其他常用的工具方法，以實(shí)現(xiàn)代碼的復(fù)用、保證代碼的整潔。

為了在 Redis 中創(chuàng)建任務(wù)隊列，我們可以單獨(dú)寫一個 createTasks.ts 的腳本，用于往隊列中塞入自定義的任務(wù)。 

import{ TASK_NAME, TASK_AMOUNT, setRedisValue, delRedisKey } from'./utils'  
  import client from'./mqClient'  
  client.on('ready', async() => {  
  await delRedisKey(TASK_NAME)  
  for(let i = TASK_AMOUNT; i > 0; i--) {  
      client.lpush(TASK_NAME, `task-${i}`)  
  }  
    client.lrange(TASK_NAME, 0, TASK_AMOUNT, async(err, reply) => {  
  if(err) {  
        console.error(err)  
  return  
  }  
      console.log(reply)  
      process.exit()  
  })  
  }) 

在這段腳本中，我們從 utils.ts 中獲取了各個 Redis 操作的方法，以及任務(wù)的名稱 TASKNAME （此處為 localtasks）和任務(wù)的總數(shù) TASKAMOUNT（此處為 20 個）。通過 LPUSH 方法往 TASKNAME 的 List 當(dāng)中塞入內(nèi)容為 task-1 到 task-20 的任務(wù)，如圖所示：

三、異步任務(wù)處理

首先新建一個 index.ts 文件，作為整個異步任務(wù)隊列處理系統(tǒng)的入口文件。   

import taskHandler from'./tasksHandler'  
   import client from'./mqClient'  
   client.on('connect', () => {  
     console.log('Redis is connected!')  
   })  
   client.on('ready', async() => {  
     console.log('Redis is ready!')  
   await taskHandler()  
   })  
   client.on('error', (e) => {  
     console.log('Redis error! '+ e)  
   }) 

在運(yùn)行該文件時，會自動連接 Redis，并且在 ready 狀態(tài)時執(zhí)行任務(wù)處理器 taskHandler()。

在上一節(jié)的操作中，我們往任務(wù)隊列里面添加了 20 個任務(wù)，每個任務(wù)都是形如 task-n 的字符串。為了驗證異步任務(wù)的實(shí)現(xiàn)，我們可以在任務(wù)處理器 taskHandler.ts 中寫一段 demo 函數(shù)，來模擬真正的異步任務(wù)： 

function handleTask(task: string) {  
 returnnewPromise((resolve) => {  
       setTimeout(async() => {  
         console.log(`Handling task: ${task}...`)  
         resolve()  
 }, 2000)  
 })  
 } 

上面這個 handleTask() 函數(shù)，將會在執(zhí)行的 2 秒后打印出當(dāng)前任務(wù)的內(nèi)容，并返回一個 Promise，很好地模擬了異步函數(shù)的實(shí)現(xiàn)方式。接下來我們將會圍繞這個函數(shù)，來處理隊列中的任務(wù)。

其實(shí)到了這一步為止，整個異步任務(wù)隊列處理系統(tǒng)已經(jīng)基本完成了，只需要在 taskHandler.ts 中補(bǔ)充一點(diǎn)點(diǎn)代碼即可：   

import{ popTask } from'./utils'  
    import client from'./mqClient'  
    function handleTask(task: string) { /* ... */}  
    exportdefaultasyncfunction tasksHandler() {  
    // 從隊列中取出一個任務(wù)  
    const task = await popTask()  
    // 處理任務(wù)  
    await handleTask(task)  
    // 遞歸運(yùn)行  
    await tasksHandler()  
    } 

最后，我們使用 PM2 啟動 4 個進(jìn)程，來試著跑一下整個項目：   

pm2 start ./dist/index.js -i 4&& pm2 logs 

可以看到，4 個任務(wù)處理器分別處理完了隊列中的所有任務(wù)，相互之前互不影響。

事到如今已經(jīng)大功告成了嗎？未必。為了測試我們的這套系統(tǒng)到底提升了多少的效率，還需要統(tǒng)計完成隊列里面所有任務(wù)的總耗時。

四、統(tǒng)計任務(wù)完成耗時

要統(tǒng)計任務(wù)完成的耗時，只需要實(shí)現(xiàn)下列的公式即可：

總耗時 = 最后一個任務(wù)的完成時間 - 首個任務(wù)被取得的時間

首先來解決“獲取首個任務(wù)被取得的時間”這個問題。

由于我們是通過 PM2 的 Cluster 模式來啟動應(yīng)用的，且從 Redis 隊列中讀取任務(wù)是個異步操作，因此在多進(jìn)程運(yùn)行的情況下無法直接保證從隊列中讀取任務(wù)的先后順序，必須通過一個額外的標(biāo)記來判斷。其原理如下圖：

如圖所示，綠色的 worker 由于無法保證運(yùn)行的先后順序，所以編號用問號來表示。當(dāng)?shù)谝粋€任務(wù)被取得時，把黃色的標(biāo)記值從 false 設(shè)置成 true。當(dāng)且僅當(dāng)黃色的標(biāo)記值為 false 時才會設(shè)置時間。這樣一來，當(dāng)其他任務(wù)被取得時，由于黃色的標(biāo)記值已經(jīng)是 true 了，因此無法設(shè)置時間，所以我們便能得到首個任務(wù)被取得的時間。

在本文的例子中，黃色的標(biāo)記值和首個任務(wù)被取得的時間也被存放在 Redis 中，分別被命名為 localtasksSETFIRST 和 localtasksBEGINTIME。

原理已經(jīng)弄懂，但是在實(shí)踐中還有一個地方值得注意。我們知道，從 Redis 中讀寫數(shù)據(jù)也是一個異步操作。由于我們有多個 worker 但只有一個 Redis，那么在讀取黃色標(biāo)記值的時候很可能會出現(xiàn)“沖突”的問題。舉個例子，當(dāng) worker-1 修改標(biāo)記值為 true 的同時， worker-2 正好在讀取標(biāo)記值。由于時間的關(guān)系，可能 worker-2 讀到的標(biāo)記值依然是 false，那么這就沖突了。為了解決這個問題，我們可以使用 node-redlock 這個工具來實(shí)現(xiàn)“鎖”的操作。

顧名思義，“鎖”的操作可以理解為當(dāng) worker-1 讀取并修改標(biāo)記值的時候，不允許其他 worker 讀取該值，也就是把標(biāo)記值給鎖住了。當(dāng) worker-1 完成標(biāo)記值的修改時會釋放鎖，此時才允許其他的 worker 去讀取該標(biāo)記值。

node-redlock 是 Redis 分布式鎖 Redlock 算法的 JavaScript 實(shí)現(xiàn)，關(guān)于該算法的講解可參考 https://redis.io/topics/distlock。值得注意的是，在 node-redlock 在使用的過程中，如果要鎖一個已存在的 key，就必須為該 key 添加一個前綴 locks:，否則會報錯。

回到 utils.ts，編寫一個 setBeginTime() 的工具函數(shù)： 

exportconst setBeginTime = async(redlock: Redlock) => {  
 // 讀取標(biāo)記值前先把它鎖住  
 constlock= await redlock.lock(`lock:${TASK_NAME}_SET_FIRST`, 1000)  
 const setFirst = await getRedisValue(`${TASK_NAME}_SET_FIRST`)  
 // 當(dāng)且僅當(dāng)標(biāo)記值不等于 true 時，才設(shè)置起始時間  
 if(setFirst !== 'true') {  
     console.log(`${pm2tips} Get the first task!`)  
 await setRedisValue(`${TASK_NAME}_SET_FIRST`, 'true')  
 await setRedisValue(`${TASK_NAME}_BEGIN_TIME`, `${new Date().getTime()}`)  
 }  
 // 完成標(biāo)記值的讀寫操作后，釋放鎖  
 awaitlock.unlock().catch(e => e)  
 } 

然后把它添加到 taskHandler() 函數(shù)里面即可： 

exportdefaultasyncfunction tasksHandler() {  
  +  // 獲取第一個任務(wù)被取得的時間  
  +  await setBeginTime(redlock)  
  // 從隊列中取出一個任務(wù)  
  const task = await popTask()  
  // 處理任務(wù)  
  await handleTask(task)  
  // 遞歸運(yùn)行  
  await tasksHandler()  
  } 

接下來解決“最后一個任務(wù)的完成時間”這個問題。

類似上一個問題，由于任務(wù)執(zhí)行的先后順序無法保證，異步操作的完成時間也無法保證，因此我們也需要一個額外的標(biāo)識來記錄任務(wù)的完成情況。在 Redis 中創(chuàng)建一個初始值為 0 的標(biāo)識 localtasksCURINDEX，當(dāng) worker 完成一個任務(wù)就讓標(biāo)識加。由于任務(wù)隊列的初始長度是已知的（為 TASKAMOUNT 常量，也寫入了 Redis 的 localtasksTOTAL 中），因此當(dāng)標(biāo)識的值等于隊列初始長度的值時，即可表明所有任務(wù)都已經(jīng)完成。

如圖所示，被完成的任務(wù)都會讓黃色的標(biāo)識加一，任何時候只要判斷到標(biāo)識的值等于隊列的初始長度值，即可表明任務(wù)已經(jīng)全部完成。

回到 taskHandler() 函數(shù)，加入下列內(nèi)容：   

exportdefaultasyncfunction tasksHandler() {  
   +  // 獲取標(biāo)識值和隊列初始長度  
   +  let curIndex = Number(await getRedisValue(`${TASK_NAME}_CUR_INDEX`))  
   +  const taskAmount = Number(await getRedisValue(`${TASK_NAME}_TOTAL`))  
   +  // 等待新任務(wù)  
   +  if(taskAmount === 0) {  
   +    console.log(`${pm2tips} Wating new tasks...`)  
   +    await sleep(2000)  
   +    await tasksHandler()  
   +    return  
   +  }  
   +  // 判斷所有任務(wù)已經(jīng)完成  
   +  if(curIndex === taskAmount) {  
   +    const beginTime = await getRedisValue(`${TASK_NAME}_BEGIN_TIME`)  
   +    // 獲取總耗時  
   +    const cost = newDate().getTime() - Number(beginTime)  
   +    console.log(`${pm2tips} All tasks were completed! Time cost: ${cost}ms. ${beginTime}`)  
   +    // 初始化 Redis 的一些標(biāo)識值  
   +    await setRedisValue(`${TASK_NAME}_TOTAL`, '0')  
   +    await setRedisValue(`${TASK_NAME}_CUR_INDEX`, '0')  
   +    await setRedisValue(`${TASK_NAME}_SET_FIRST`, 'false')  
   +    await delRedisKey(`${TASK_NAME}_BEGIN_TIME`)  
   +    await sleep(2000)  
   +    await tasksHandler()  
   }  
   // 獲取第一個任務(wù)被取得的時間  
   await setBeginTime(redlock)  
   // 從隊列中取出一個任務(wù)  
   const task = await popTask()  
   // 處理任務(wù)  
   await handleTask(task)  
   + // 任務(wù)完成后需要為標(biāo)識位加一  
   +  try{  
   +    constlock= await redlock.lock(`lock:${TASK_NAME}_CUR_INDEX`, 1000)  
   +    curIndex = await getCurIndex()  
   +    await setCurIndex(curIndex + 1)  
   +    awaitlock.unlock().catch((e) => e)  
   +  } catch(e) {  
   +    console.log(e)  
   +  }  
   +  // recursion  
   +  await tasksHandler()  
   +}  
   // 遞歸運(yùn)行  
   await tasksHandler()  
   } 

到這一步為止，我們已經(jīng)解決了獲取“最后一個任務(wù)的完成時間”的問題，再結(jié)合前面的首個任務(wù)被取得的時間，便能得出運(yùn)行的總耗時。

最后來看一下實(shí)際的運(yùn)行效果。我們循例往隊列里面添加了 task-1 到 task-20 這 20 個任務(wù)，然后啟動 4 個進(jìn)程來跑：

運(yùn)行狀況良好。從運(yùn)行結(jié)果來看，4 個進(jìn)程處理 20 個平均耗時 2 秒的任務(wù)，只需要 10 秒的時間，完全符合設(shè)想。

五、小結(jié)

當(dāng)面對海量的異步任務(wù)需要處理的時候，多進(jìn)程 + 任務(wù)隊列的方式是一個不錯的解決方式。本文通過探索 Redis + NodeJS 結(jié)合的方式，構(gòu)造出了一個異步任務(wù)隊列處理系統(tǒng)，能較好地完成最初方案的設(shè)想，但依然有很多問題需要改進(jìn)。比如說當(dāng)任務(wù)出錯了應(yīng)該怎么辦，系統(tǒng)能否支持不同類型的任務(wù)，能否運(yùn)行多個隊列等等，都是值得思考的問題。