69久久久久久久,国产精品久久久久一区欢美无码,午夜婷婷在线播放,日韩AV免费在线,成人性爱免费在线观看,yy4080一级毛片一成人,日本操B视频,手机免费看A V

（給前端大學(xué)加星標(biāo)，提升前端技能.）

作者：jrainlau
https://segmentfault.com/a/1190000037567355

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

總耗時(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 文件并寫(xiě)入如下內(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<string| null> => 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ú)寫(xiě)一個(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ù)的名稱(chēng) 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 中寫(xiě)一段 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 中讀寫(xiě)數(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，編寫(xiě)一個(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)記值的讀寫(xiě)操作后，釋放鎖
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)題。

類(lèi)似上一個(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 常量，也寫(xiě)入了 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)能否支持不同類(lèi)型的任務(wù)，能否運(yùn)行多個(gè)隊(duì)列等等，都是值得思考的問(wèn)題。

??愛(ài)心三連擊
點(diǎn)分享
點(diǎn)點(diǎn)贊
點(diǎn)在看

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

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

二、使用 NodeJS 操作 Redis

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

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

五、小結(jié)

??愛(ài)心三連擊