前言

無(wú)論是在大數(shù)據(jù)處理領(lǐng)域，還是在消息處理領(lǐng)域，任務(wù)系統(tǒng)都有一個(gè)很關(guān)鍵的能力 - 任務(wù)觸發(fā)去重的保障。這個(gè)能力對(duì)于一些準(zhǔn)確性要求極高的場(chǎng)景中（如金融等）是必不可少的。作為 Serverless 化任務(wù)處理平臺(tái)，Serverless Task 也需要提供這類(lèi)保障，在用戶應(yīng)用層面及自身系統(tǒng)內(nèi)部?jī)蓚€(gè)維度具備任務(wù)的準(zhǔn)確觸發(fā)語(yǔ)義。本文主要針對(duì)消息處理可靠性這一主題來(lái)介紹函數(shù)計(jì)算內(nèi)部的一些技術(shù)細(xì)節(jié)，并展示如何在實(shí)際應(yīng)用中使用函數(shù)計(jì)算所提供的這方面能力來(lái)增強(qiáng)任務(wù)執(zhí)行的可靠性。

淺談任務(wù)去重

在討論異步消息處理系統(tǒng)時(shí)，消息處理的基本語(yǔ)義是無(wú)法繞開(kāi)的話題。在一個(gè)異步的消息處理系統(tǒng)（任務(wù)系統(tǒng)）中，一條消息的處理流程簡(jiǎn)化如下圖所示：

圖 1

用戶下發(fā)任務(wù) - 進(jìn)入隊(duì)列 - 任務(wù)處理單元監(jiān)聽(tīng)并獲取消息 - 調(diào)度到實(shí)際 worker 執(zhí)行

在任務(wù)消息整個(gè)的流轉(zhuǎn)過(guò)程中，任何組件（環(huán)節(jié)）可能出現(xiàn)的宕機(jī)等問(wèn)題會(huì)導(dǎo)致消息的錯(cuò)誤傳遞。一般的任務(wù)系統(tǒng)會(huì)提供至多 3 個(gè)層級(jí)的消息處理語(yǔ)義：

●At-Most-Once：保證消息最多被傳遞一次。當(dāng)出現(xiàn)網(wǎng)絡(luò)分區(qū)、系統(tǒng)組件宕機(jī)時(shí)，可能出現(xiàn)消息丟失；

●At-Least-Once：保證消息至少被傳遞一次。消息傳遞鏈路支持錯(cuò)誤重試，利用消息重發(fā)機(jī)制保證下游一定收到上游消息，但是在宕機(jī)或者網(wǎng)絡(luò)分區(qū)的場(chǎng)景下，可能導(dǎo)致相同消息傳遞多次。

●Exactly-Once機(jī)制則可以保證消息精確被傳送一次，精確一次并不是意味著在宕機(jī)或網(wǎng)絡(luò)分區(qū)的場(chǎng)景下沒(méi)有重傳，而是重傳對(duì)于接受方的狀態(tài)不產(chǎn)生任何改變，與傳送一次的結(jié)果一樣。在實(shí)際生產(chǎn)中，往往是依賴重傳機(jī)制 & 接收方去重（冪等）來(lái)做到 Exactly Once。

函數(shù)計(jì)算能夠提供任務(wù)分發(fā)的 Exactly Once 語(yǔ)義，即無(wú)論在何種情況下，重復(fù)的任務(wù)將被系統(tǒng)認(rèn)為是相同的觸發(fā)，進(jìn)而只進(jìn)行一次的任務(wù)分發(fā)。

結(jié)合圖 1，如果要做到任務(wù)去重，系統(tǒng)至少需要提供兩個(gè)維度的保障：

1、系統(tǒng)側(cè)保障：任務(wù)調(diào)度系統(tǒng)自身的 failover 不影響消息的傳遞正確性及唯一性；

2、提供給用戶一種機(jī)制，可以做到整個(gè)業(yè)務(wù)邏輯的觸發(fā)去重語(yǔ)義。

下面，我們將結(jié)合簡(jiǎn)化的 Serverless Task 系統(tǒng)架構(gòu)，談一談函數(shù)計(jì)算是如何做到上面的能力的。

函數(shù)計(jì)算異步任務(wù)觸發(fā)去重的實(shí)現(xiàn)

函數(shù)計(jì)算的任務(wù)系統(tǒng)架構(gòu)如下圖所示

圖 2

首先，用戶調(diào)用函數(shù)計(jì)算 API 下發(fā)一個(gè)任務(wù)（步驟 1）進(jìn)入系統(tǒng)的 API-Server 中，API-Server 進(jìn)行校驗(yàn)后將消息傳入內(nèi)部隊(duì)列（步驟 2.1）。后臺(tái)有一個(gè)異步模塊實(shí)時(shí)監(jiān)聽(tīng)內(nèi)部隊(duì)列（步驟 2.2），之后調(diào)用資源管理模塊獲取運(yùn)行時(shí)資源（步驟 2.2-2.3）。獲取運(yùn)行時(shí)資源后，調(diào)度模塊將任務(wù)數(shù)據(jù)下發(fā)到 VM 級(jí)別的客戶端中（步驟 3.1），并由客戶端將任務(wù)轉(zhuǎn)發(fā)至實(shí)際的用戶運(yùn)行資源（步驟 3.2）。為了做到上文中所提到的兩個(gè)維度的保障，我們需要在以下層面進(jìn)行支持：

1、系統(tǒng)側(cè)保障：在步驟 2.1 - 3.1 中，任何一個(gè)中間過(guò)程的 Failover 只能觸發(fā)一次步驟 3.2 的執(zhí)行，即只會(huì)調(diào)度一次用戶實(shí)例的運(yùn)行；

2、用戶側(cè)應(yīng)用級(jí)別去重能力：能夠支持用戶多次反復(fù)執(zhí)行步驟 1，但實(shí)際只會(huì)觸發(fā)一次 步驟 3.2 的執(zhí)行。

系統(tǒng)側(cè)優(yōu)雅升級(jí) & Failover 時(shí)的任務(wù)分發(fā)去重保證

當(dāng)用戶的消息進(jìn)入函數(shù)計(jì)算系統(tǒng)中（即完成步驟 2.1）后，用戶的請(qǐng)求將收到 HTTP 狀態(tài)碼 202 的 Response，用戶可以認(rèn)為已經(jīng)成功提交一次任務(wù)。從該任務(wù)消息進(jìn)入 MQ 起，其生命周期便由 Scheduler 維護(hù)，所以 Scheduler 的穩(wěn)定性及 MQ 的穩(wěn)定性將直接影響系統(tǒng) Exactly Once 的實(shí)現(xiàn)方案。

在大多數(shù)開(kāi)源消息系統(tǒng)中（如 MQ、Kafka）一般都提供消息多副本存儲(chǔ)及唯一消費(fèi)的語(yǔ)義。函數(shù)計(jì)算所使用的消息隊(duì)列（最底層為 RocketMQ）也是同樣的，底層存儲(chǔ)的 3 副本實(shí)現(xiàn)使得我們無(wú)需關(guān)注消息存儲(chǔ)方面的穩(wěn)定性。除此之外，函數(shù)計(jì)算所使用的的消息隊(duì)列還具有以下特性：

1、消費(fèi)的唯一性：每一個(gè)隊(duì)列中的每一條消息當(dāng)被消費(fèi)后，會(huì)進(jìn)入“不可見(jiàn)模式”。在此模式下，其他消費(fèi)者無(wú)法獲取該消息；

2、每條消息的實(shí)際消費(fèi)者需要實(shí)時(shí)更新該模式的不可見(jiàn)時(shí)間；當(dāng)消費(fèi)者消費(fèi)完成后，需要顯示的刪除該消息。因此，消息在隊(duì)列中的的整個(gè)生命周期如下圖所示：

圖 3

Scheduler 主要負(fù)責(zé)消息的處理，其任務(wù)主要有以下幾個(gè)部分組成：

1、根據(jù)函數(shù)計(jì)算負(fù)載均衡模塊的調(diào)度策略，監(jiān)聽(tīng)自身所負(fù)責(zé)的隊(duì)列；

2、當(dāng)隊(duì)列中出現(xiàn)消息后，拉取消息，并在內(nèi)存中維持一個(gè)狀態(tài)：直到消息消費(fèi)完成（用戶實(shí)例返回函數(shù)執(zhí)行結(jié)果）前，不斷更新消息的可見(jiàn)時(shí)間，確保消息不會(huì)再次在隊(duì)列中出現(xiàn)；

3、當(dāng)任務(wù)執(zhí)行完成后，顯示刪除該消息。

在隊(duì)列的調(diào)度模型方面，函數(shù)計(jì)算對(duì)于普通用戶采用“單隊(duì)列”的管理模式；即每一個(gè)用戶的所有異步執(zhí)行請(qǐng)求由一個(gè)獨(dú)立隊(duì)列相互隔離，并且由一個(gè) Scheduler 固定負(fù)責(zé)。這個(gè)負(fù)載的映射關(guān)系由函數(shù)計(jì)算的負(fù)載均衡服務(wù)進(jìn)行管理，如下圖所示（我們?cè)诤罄m(xù)文章中還會(huì)更為詳細(xì)的介紹這部分內(nèi)容）：

圖 4

當(dāng) Scheduler 1 發(fā)生宕機(jī)或升級(jí)時(shí)，任務(wù)由兩種執(zhí)行狀態(tài)：

1、如果消息還未傳遞到用戶的執(zhí)行實(shí)例中（圖 2 中的步驟 3.1 ~ 3.2），那么當(dāng)這臺(tái) Scheduler 負(fù)責(zé)的隊(duì)列被其他 Scheduler 拾起后，消息將在消費(fèi)可見(jiàn)期后再次出現(xiàn)，因此 Scheduler 2 將再次獲取該消息，做到后續(xù)的觸發(fā)。

2、如果消息已經(jīng)開(kāi)始執(zhí)行（步驟 3.2），當(dāng)消息在 Scheduler 2 中再次出現(xiàn)后，我們依賴用戶 VM 中的 Agent 進(jìn)行狀態(tài)管理。此時(shí) Scheduler 2 將向?qū)?yīng)的 Agent 發(fā)送執(zhí)行請(qǐng)求；此時(shí) Agent 發(fā)現(xiàn)該消息已經(jīng)存在于內(nèi)存中，那么將直接忽略執(zhí)行請(qǐng)求，并將執(zhí)行的結(jié)果在執(zhí)行后通過(guò)此鏈接告知 Scheduler 2，進(jìn)而完成 Failover 的恢復(fù)。

用戶側(cè)業(yè)務(wù)級(jí)別的分發(fā)去重實(shí)現(xiàn)

函數(shù)計(jì)算系統(tǒng)能夠做到對(duì)于單點(diǎn)故障下的每條消息準(zhǔn)確的消費(fèi)能力，但是如果用戶側(cè)對(duì)于同一條業(yè)務(wù)數(shù)據(jù)反復(fù)觸發(fā)函數(shù)執(zhí)行的話，函數(shù)計(jì)算無(wú)法識(shí)別不同消息是否在邏輯上是同一個(gè)任務(wù)。這種情況往往發(fā)生在網(wǎng)絡(luò)分區(qū)。在圖 2 中，如果用戶調(diào)用 1 發(fā)生超時(shí)，此時(shí)有可能有兩種情況：

1、消息未到達(dá)函數(shù)計(jì)算系統(tǒng)，任務(wù)未成功提交；

2、消息已經(jīng)到達(dá)函數(shù)計(jì)算并入隊(duì)，任務(wù)提交成功，但由于超時(shí)用戶無(wú)法得知提交成功的信息。

大多數(shù)情況下用戶會(huì)對(duì)此次的提交進(jìn)行重試。如果是第 2 種情況，那么同一個(gè)任務(wù)將被提交并執(zhí)行多次。因此函數(shù)計(jì)算需要提供一種機(jī)制，保證這種場(chǎng)景下業(yè)務(wù)的準(zhǔn)確性。

函數(shù)計(jì)算提供了 TaskID 這一任務(wù)概念（StatefulAsyncInvocationID）。該 ID 全局唯一。用戶每次提交任務(wù)均可以指定這樣一個(gè) ID。當(dāng)發(fā)生請(qǐng)求超時(shí)時(shí)，用戶可以進(jìn)行無(wú)限次重試。所有的重復(fù)重試將在函數(shù)計(jì)算側(cè)進(jìn)行校驗(yàn)。函數(shù)計(jì)算內(nèi)部使用 DB 對(duì)任務(wù) Meta 數(shù)據(jù)進(jìn)行存儲(chǔ)；當(dāng)有相同 ID 進(jìn)入系統(tǒng)時(shí)該次請(qǐng)求將被拒絕，并返回 400 錯(cuò)誤。此時(shí)客戶端即可得知任務(wù)的提交情況。

在實(shí)際使用中以 Go SDK 為例，您可以編輯如下觸發(fā)任務(wù)的代碼：

import fc "github.com/aliyun/fc-go-sdk"

func SubmitJob() {
    invokeInput := fc.NewInvokeFunctionInput("ServiceName", "FunctionName")
    invokeInput = invokeInput.WithAsyncInvocation().WithStatefulAsyncInvocationID("TaskUUID")
    invokeOutput, err := fcClient.InvokeFunction(invokeInput)
    ...
}

便提交了一個(gè)獨(dú)一無(wú)二的任務(wù)。

總結(jié)

本文介紹了函數(shù)計(jì)算 Serverless Task 對(duì)于任務(wù)觸發(fā)去重的相關(guān)技術(shù)細(xì)節(jié)，以便支持對(duì)于任務(wù)執(zhí)行準(zhǔn)確性有嚴(yán)格要求的場(chǎng)景。在使用 Serverless Task 后，您無(wú)需擔(dān)心任何系統(tǒng)組件的 Failover，您每次提交的任務(wù)將被準(zhǔn)確執(zhí)行一次。為了支持業(yè)務(wù)側(cè)語(yǔ)義的分發(fā)去重，您可以在提交任務(wù)時(shí)設(shè)置任務(wù)的全局唯一 ID，使用函數(shù)計(jì)算提供的能力幫您對(duì)任務(wù)進(jìn)行去重處理。