言情小说君子以泽,性爱有声小说在线收听,盗墓笔记小说txt下载

新聞中心

這里有您想知道的互聯(lián)網(wǎng)營(yíng)銷解決方案

Redis+Node.js實(shí)現(xiàn)一個(gè)能處理海量數(shù)據(jù)的異步任務(wù)隊(duì)列系統(tǒng)

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

創(chuàng)新互聯(lián)2013年開(kāi)創(chuàng)至今，是專業(yè)互聯(lián)網(wǎng)技術(shù)服務(wù)公司，擁有項(xiàng)目成都做網(wǎng)站、成都網(wǎng)站設(shè)計(jì)、成都外貿(mào)網(wǎng)站建設(shè)網(wǎng)站策劃，項(xiàng)目實(shí)施與項(xiàng)目整合能力。我們以讓每一個(gè)夢(mèng)想脫穎而出為使命，1280元隆堯做網(wǎng)站,已為上家服務(wù),為隆堯各地企業(yè)和個(gè)人服務(wù),聯(lián)系電話:13518219792

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

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

總耗時(shí)時(shí)間 = 2,500,000 秒 ≈ 695 小時(shí) ≈ 29 天

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

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

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

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

其基本原理如下圖所示：

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

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

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

二、使用 NodeJS 操作 Redis

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

Docker 拉取 Redis 鏡像

 
 
 
   
  
  
  docker pull redis:latest

Docker 啟動(dòng) Redis

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

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

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

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

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

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

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

為了在 Redis 中創(chuàng)建任務(wù)隊(duì)列，我們可以單獨(dú)寫一個(gè) createTasks.ts 的腳本，用于往隊(duì)列中塞入自定義的任務(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 中獲取了各個(gè) Redis 操作的方法，以及任務(wù)的名稱 TASKNAME （此處為 localtasks）和任務(wù)的總數(shù) TASKAMOUNT（此處為 20 個(gè)）。通過(guò) LPUSH 方法往 TASKNAME 的 List 當(dāng)中塞入內(nèi)容為 task-1 到 task-20 的任務(wù)，如圖所示：

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

首先新建一個(gè) index.ts 文件，作為整個(gè)異步任務(wù)隊(duì)列處理系統(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)行該文件時(shí)，會(huì)自動(dòng)連接 Redis，并且在 ready 狀態(tài)時(shí)執(zhí)行任務(wù)處理器 taskHandler()。

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

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

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

其實(shí)到了這一步為止，整個(gè)異步任務(wù)隊(duì)列處理系統(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() {    
  
  
      // 從隊(duì)列中取出一個(gè)任務(wù)    
  
  
      const task = await popTask()    
  
  
      // 處理任務(wù)    
  
  
      await handleTask(task)    
  
  
      // 遞歸運(yùn)行    
  
  
      await tasksHandler()    
  
  
      }

最后，我們使用 PM2 啟動(dòng) 4 個(gè)進(jìn)程，來(lái)試著跑一下整個(gè)項(xiàng)目：

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

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

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

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

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

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

首先來(lái)解決“獲取首個(gè)任務(wù)被取得的時(shí)間”這個(gè)問(wèn)題。

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

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

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

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

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

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

回到 utils.ts，編寫一個(gè) 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í)，才設(shè)置起始時(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() {    
  
  
    +  // 獲取第一個(gè)任務(wù)被取得的時(shí)間    
  
  
    +  await setBeginTime(redlock)    
  
  
    // 從隊(duì)列中取出一個(gè)任務(wù)    
  
  
    const task = await popTask()    
  
  
    // 處理任務(wù)    
  
  
    await handleTask(task)    
  
  
    // 遞歸運(yùn)行    
  
  
    await tasksHandler()    
  
  
    }

接下來(lái)解決“最后一個(gè)任務(wù)的完成時(shí)間”這個(gè)問(wèn)題。

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

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

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

 
 
 
   
  
  
  exportdefaultasyncfunction tasksHandler() {    
  
  
     +  // 獲取標(biāo)識(shí)值和隊(duì)列初始長(zhǎng)度    
  
  
     +  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`)    
  
  
     +    // 獲取總耗時(shí)    
  
  
     +    const cost = newDate().getTime() - Number(beginTime)    
  
  
     +    console.log(`${pm2tips} All tasks were completed! Time cost: ${cost}ms. ${beginTime}`)    
  
  
     +    // 初始化 Redis 的一些標(biāo)識(shí)值    
  
  
     +    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()    
  
  
     }    
  
  
     // 獲取第一個(gè)任務(wù)被取得的時(shí)間    
  
  
     await setBeginTime(redlock)    
  
  
     // 從隊(duì)列中取出一個(gè)任務(wù)    
  
  
     const task = await popTask()    
  
  
     // 處理任務(wù)    
  
  
     await handleTask(task)    
  
  
     + // 任務(wù)完成后需要為標(biāo)識(shí)位加一    
  
  
     +  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)解決了獲取“最后一個(gè)任務(wù)的完成時(shí)間”的問(wèn)題，再結(jié)合前面的首個(gè)任務(wù)被取得的時(shí)間，便能得出運(yùn)行的總耗時(shí)。

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

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

五、小結(jié)

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

當(dāng)前題目：Redis+Node.js實(shí)現(xiàn)一個(gè)能處理海量數(shù)據(jù)的異步任務(wù)隊(duì)列系統(tǒng)
本文來(lái)源：http://m.fisionsoft.com.cn/article/dhecohp.html

新聞中心

其他資訊