基于golang的輕量級工作流框架Fastflow
更新時間：2022年05月24日 10:31:48 作者：ShiningRush
這篇文章主要介紹了基于golang的輕量級工作流框架Fastflow,fastflow 執(zhí)行任務的過程會涉及到幾個概念：Dag, Task, Action, DagInstance，本文給大家分享完整流程，需要的朋友可以參考下
Fastflow 是什么？用一句話來定義它：一個 基于golang協程、支持水平擴容的分布式高性能工作流框架。
它具有以下特點：
- 易用性：工作流模型基于 DAG 來定義，同時還提供開箱即用的 API，你可以隨時通過 API 創(chuàng)建、運行、暫停工作流等，在開發(fā)新的原子能力時還提供了開箱即用的分布式鎖功能
- 高性能：得益于 golang 的協程 與 channel 技術，fastflow 可以在單實例上并行執(zhí)行數百、數千乃至數萬個任務
- 可觀測性：fastflow 基于 Prometheus 的 metrics 暴露了當前實例上的任務執(zhí)行信息，比如并發(fā)任務數、任務分發(fā)時間等。
- 可伸縮性：支持水平伸縮，以克服海量任務帶來的單點瓶頸，同時通過選舉 Leader 節(jié)點來保障各個節(jié)點的負載均衡
- 可擴展性：fastflow 準備了部分開箱即用的任務操作，比如 http請求、執(zhí)行腳本等，同時你也可以自行定義新的節(jié)點動作，同時你可以根據上下文來決定是否跳過節(jié)點(skip)
- 輕量：它僅僅是一個基礎框架，而不是一個完整的產品，這意味著你可以將其很低成本融入到遺留項目而無需部署、依賴另一個項目，這既是它的優(yōu)點也是缺點——當你真的需要一個開箱即用的產品時（比如 airflow），你仍然需要少量的代碼開發(fā)才能使用
為什么要開發(fā) Fastflow
組內有很多項目都涉及復雜的任務流場景，比如離線任務，集群上下架，容器遷移等，這些場景都有幾個共同的特點：
流程耗時且步驟復雜，比如創(chuàng)建一個 k8s 集群，需要幾十步操作，其中包含腳本執(zhí)行、接口調用等，且相互存在依賴關系。
任務量巨大，比如容器平臺每天都會有幾十萬的離線任務需要調度執(zhí)行、再比如我們管理數百個K8S集群，幾乎每天會有集群需要上下節(jié)點、遷移容器等。
我們嘗試過各種解法：
- 硬編碼實現：雖然工作量較小，但是只能滿足某個場景下的特定工作流，沒有可復用性。
- airflow：我們最開始的離線任務引擎就是基于這個來實現的，不得不承認它的功能很全，也很方便，但是存在幾個問題
- 由 python 編寫的，我們希望團隊維護的項目能夠統一語言，更有助于提升工作效率，雖然對一個有經驗的程序員來說多語言并不是問題，但是頻繁地在多個語言間來回切換其實是不利于高效工作的
- airflow 的任務執(zhí)行是以 進程 來運行的，雖然有更好的隔離性，但是顯然因此而犧牲了性能和并發(fā)度。
- 公司內的工作流平臺：你可能想象不到一個世界前十的互聯網公司，他們內部一個經歷了數年線上考證的運維用工作流平臺，會脆弱到承受不了上百工作流的并發(fā)，第一次壓測就直接讓他們的服務癱瘓，進而影響到其他業(yè)務的運維任務。據團隊反饋稱是因為我們的工作流組成太復雜，一個流包含數十個任務節(jié)點才導致了這次意外的服務過載，隨后半年這個團隊重寫了一個新的v2版本。
當然 Github 上也還有其他的任務流引擎，我們也都評估過，無法滿足需求。比如 kubeflow 是基于 Pod 執(zhí)行任務的，比起 進程 更為重量，還有一些項目，要么就是沒有經過海量數據的考驗，要么就是沒有考慮可伸縮性，面對大量任務的執(zhí)行無法水平擴容。
Concept
工作流模型
fastflow 的工作流模型基于 DAG(Directed acyclic graph),下圖是一個簡單的 DAG 示意圖：
在這個圖中，首先 A 節(jié)點所定義的任務會被執(zhí)行，當 A 執(zhí)行完畢后，B、C兩個節(jié)點所定義的任務將同時被觸發(fā)，而只有 B、C 兩個節(jié)點都執(zhí)行成功后，最后的 D 節(jié)點才會被觸發(fā)，這就是 fastflow 的工作流模型。
工作流的要素
fastflow 執(zhí)行任務的過程會涉及到幾個概念：Dag, Task, Action, DagInstance
Dag
描述了一個完整流程，它的每個節(jié)點被稱為 Task，它定義了各個 Task 的執(zhí)行順序和依賴關系，你可以通過編程 or yaml 來定義它
一個編程式定義的DAG
dag := &entity.Dag{
BaseInfo: entity.BaseInfo{
ID: "test-dag",
},
Name: "test",
Tasks: []entity.Task{
{ID: "task1", ActionName: "PrintAction"},
{ID: "task2", ActionName: "PrintAction", DependOn: []string{"task1"}},
{ID: "task3", ActionName: "PrintAction", DependOn: []string{"task2"}},
},
}對應的yaml如下：
id: "test-dag" name: "test" tasks: - id: "task1" actionName: "PrintAction" - id: ["task2"] actionName: "PrintAction" dependOn: ["task1"] - id: "task3" actionName: "PrintAction" dependOn: ["task2"]
同時 Dag 可以定義這個工作流所需要的參數，以便于在各個 Task 去消費它：
id: "test-dag"
name: "test"
vars:
fileName:
desc: "the file name"
defaultValue: "file.txt"
filePath:
desc: "the file path"
defaultValue: "/tmp/"
tasks:
- id: "task1"
actionName: "PrintAction"
params:
writeName: "{{fileName}}"
writePath: "{{filePath}}"Task
它定義了這個節(jié)點的具體工作，比如是要發(fā)起一個 http 請求，或是執(zhí)行一段腳本等，這些不同動作都通過選擇不同的 Action 來實現，同時它也可以定義在何種條件下需要跳過 or 阻塞該節(jié)點。
下面這段yaml演示了 Task 如何根據某些條件來跳過運行該節(jié)點。
id: "test-dag" name: "test" vars: fileName: desc: "the file name" defaultValue: "file.txt" tasks: - id: "task1" actionName: "PrintAction" preCheck: - act: skip #you can set "skip" or "block" conditions: - source: vars # source could be "vars" or "share-data" key: "fileName" op: "in" values: ["warn.txt", "error.txt"]
Task 的狀態(tài)有以下幾個：
- init: Task已經初始化完畢，等待執(zhí)行
- running: 正在運行中
- ending: 當執(zhí)行 Action 的 Run 所定義的內容后，會進入到該狀態(tài)
- retrying: 任務重試中
- failed: 執(zhí)行失敗
- success: 執(zhí)行成功
- blocked: 任務已阻塞，需要人工啟動
- skipped: 任務已跳過
Action
Action 是工作流的核心，定義了該節(jié)點將執(zhí)行什么操作，fastflow攜帶了一些開箱即用的Action，但是一般你都需要根據具體的業(yè)務場景自行編寫，它有幾個關鍵屬性：
- Name: Required Action的名稱，不可重復，它是與 Task 關聯的核心
- Run: Required 需要執(zhí)行的動作，fastflow 將確保該動作僅會被執(zhí)行 一次(ExactlyOnce)
- RunBefore: Optional 在執(zhí)行 Run 之前運行，如果有一些前置動作，可以在這里執(zhí)行，RunBefore 有可能會被執(zhí)行多次。
- RunAfter: Optional 在執(zhí)行 Run 之后運行，一些長時間執(zhí)行的任務內容建議放在這里，只要 Task 尚未結束，節(jié)點發(fā)生故障重啟時仍然會繼續(xù)執(zhí)行這部分內容，
- RetryBefore:Optional 在重試失敗的任務節(jié)點，可以提前執(zhí)行一些清理的動作
自行開發(fā)的 Action 在使用前都必須先注冊到 fastflow，如下所示：
type PrintParams struct {
Key string
Value string
}
type PrintAction struct {
}
// Name define the unique action identity, it will be used by Task
func (a *PrintAction) Name() string {
return "PrintAction"
}
func (a *PrintAction) Run(ctx run.ExecuteContext, params interface{}) error {
cinput := params.(*ActionParam)
fmt.Println("action start: ", time.Now())
fmt.Println(fmt.Sprintf("params: key[%s] value[%s]", cinput.Key, cinput.Value))
return nil
}
func (a *PrintAction) ParameterNew() interface{} {
return &PrintParams{}
}
func main() {
...
// Register action
fastflow.RegisterAction([]run.Action{
&PrintAction{},
})
...
}
DagInstance
當你開始運行一個 Dag 后，則會為本次執(zhí)行生成一個執(zhí)行記錄，它被稱為 DagInstance，當它生成以后，會由 Leader 實例將其分發(fā)到一個健康的 Worker，再由其解析、執(zhí)行。
實例類型與Module
首先 fastflow 是一個分布式的框架，意味著你可以部署多個實例來分擔負載，而實例被分為兩類角色：
- Leader：此類實例在運行過程中只會存在一個，從 Worker 中進行選舉而得出，它負責給 Worker 實例分發(fā)任務，也會監(jiān)聽長時間得不到執(zhí)行的任務將其調度到其他節(jié)點等
- Worker：此類實例會存在復數個，它們負責解析 DAG 工作流并以
協程執(zhí)行其中的任務
而不同節(jié)點能夠承擔不同的功能，其背后是不同的 模塊 在各司其職，不同節(jié)點所運行的模塊如下圖所示：
NOTE
- Leader 實例本質上是一個承擔了
仲裁者角色的 Worker，因此它也會分擔工作負載。 - 為了實現更均衡的負載，以及獲得更好的可擴展性，fastflow 沒有選擇加鎖競爭的方式來實現工作分發(fā)
從上面的圖看，Leader 實例會比 Worker 實例多運行一些模塊用于執(zhí)行中仲裁者相關的任務，模塊之間的協作關系如下圖所示：
其中各個模塊的職責如下：
- Keeper: 每個節(jié)點都會運行 負責注冊節(jié)點到存儲中，保持心跳，同時也會周期性嘗試競選 Leader，防止上任 Leader 故障后阻塞系統，這個模塊同時也提供了 分布式鎖 功能，我們也可以實現不同存儲的 Keeper 來滿足特定的需求，比如 Etcd or Zookeepper，目前支持的 Keeper 實現只有 Mongo
- Store: 每個節(jié)點都會運行 負責解耦 Worker 對底層存儲的依賴，通過這個組件，我們可以實現利用 Mongo, Mysql 等來作為 fastflow 的后端存儲，目前僅實現了 Mongo
- Parser：Worker 節(jié)點運行 負責監(jiān)聽分發(fā)到自己節(jié)點的任務，然后將其 DAG 結構重組為一顆 Task 樹，并渲染好各個任務節(jié)點的輸入，接下來通知 Executor 模塊開始執(zhí)行 Task
- Commander：每個節(jié)點都會運行 負責封裝一些常見的指令，如停止、重試、繼續(xù)等，下發(fā)到節(jié)點去運行
- Executor： Worker 節(jié)點運行 按照 Parser 解析好的 Task 樹以 goroutine 運行單個的 Task
- Dispatcher：Leader節(jié)點才會運行 負責監(jiān)聽等待執(zhí)行的 DAG，并根據 Worker 的健康狀況均勻地分發(fā)任務
- WatchDog：Leader節(jié)點才會運行 負責監(jiān)聽執(zhí)行超時的 Task 將其更新為失敗，同時也會重新調度那些一直得不到執(zhí)行的 DagInstance 到其他 Worker
Tips
以上模塊的分布機制僅僅只是 fastflow 的默認實現，你也可以自行決定實例運行的模塊，比如在 Leader 上不再運行 Worker 的實例，讓其專注于任務調度。
GetStart
更多例子請參考項目下面的
examples目錄
準備一個Mongo實例
如果已經你已經有了可測試的實例，可以直接替換為你的實例，如果沒有的話，可以使用Docker容器在本地跑一個，指令如下：
docker run -d --name fastflow-mongo --network host mongo
運行 fastflow
運行以下示例
package main
import (
"fmt"
"log"
"time"
"github.com/shiningrush/fastflow"
mongoKeeper "github.com/shiningrush/fastflow/keeper/mongo"
"github.com/shiningrush/fastflow/pkg/entity/run"
"github.com/shiningrush/fastflow/pkg/mod"
mongoStore "github.com/shiningrush/fastflow/store/mongo"
)
type PrintAction struct {
}
// Name define the unique action identity, it will be used by Task
func (a *PrintAction) Name() string {
return "PrintAction"
}
func (a *PrintAction) Run(ctx run.ExecuteContext, params interface{}) error {
fmt.Println("action start: ", time.Now())
return nil
}
func main() {
// Register action
fastflow.RegisterAction([]run.Action{
&PrintAction{},
})
// init keeper, it used to e
keeper := mongoKeeper.NewKeeper(&mongoKeeper.KeeperOption{
Key: "worker-1",
// if your mongo does not set user/pwd, youshould remove it
ConnStr: "mongodb://root:pwd@127.0.0.1:27017/fastflow?authSource=admin",
Database: "mongo-demo",
Prefix: "test",
})
if err := keeper.Init(); err != nil {
log.Fatal(fmt.Errorf("init keeper failed: %w", err))
}
// init store
st := mongoStore.NewStore(&mongoStore.StoreOption{
// if your mongo does not set user/pwd, youshould remove it
ConnStr: "mongodb://root:pwd@127.0.0.1:27017/fastflow?authSource=admin",
Database: "mongo-demo",
Prefix: "test",
})
if err := st.Init(); err != nil {
log.Fatal(fmt.Errorf("init store failed: %w", err))
}
go createDagAndInstance()
// start fastflow
if err := fastflow.Start(&fastflow.InitialOption{
Keeper: keeper,
Store: st,
// use yaml to define dag
ReadDagFromDir: "./",
}); err != nil {
panic(fmt.Sprintf("init fastflow failed: %s", err))
}
}
func createDagAndInstance() {
// wait fast start completed
time.Sleep(time.Second)
// run some dag instance
for i := 0; i < 10; i++ {
_, err := mod.GetCommander().RunDag("test-dag", nil)
if err != nil {
log.Fatal(err)
}
time.Sleep(time.Second * 10)
}
}程序運行目錄下的test-dag.yaml
id: "test-dag" name: "test" tasks: - id: "task1" actionName: "PrintAction" - id: "task2" actionName: "PrintAction" dependOn: ["task1"] - id: "task3" actionName: "PrintAction" dependOn: ["task2"]
Basic
Task與Task之間的通信
由于任務都是基于 goroutine 來執(zhí)行，因此任務之間的 context 是共享的，意味著你完全可以使用以下的代碼：
func (a *UpAction) Run(ctx run.ExecuteContext, params interface{}) error {
ctx.WithValue("key", "value")
return nil
}
func (a *DownAction) Run(ctx run.ExecuteContext, params interface{}) error {
val := ctx.Context().Value("key")
return nil
}但是注意這樣做有個弊端：當節(jié)點重啟時，如果任務尚未執(zhí)行完畢，那么這部分內容會丟失。
如果不想因為故障or升級而丟失你的更改，可以使用 ShareData 來傳遞進行通信，ShareData 是整個 在整個 DagInstance 的生命周期都會共享的一塊數據空間，每次對它的寫入都會通過 Store 組件持久化，以確保數據不會丟失，用法如下：
func (a *UpAction) Run(ctx run.ExecuteContext, params interface{}) error {
ctx.ShareData().Set("key", "value")
return nil
}
func (a *DownAction) Run(ctx run.ExecuteContext, params interface{}) error {
val := ctx.ShareData().Get("key")
return nil
}任務日志
fastflow 還提供了 Task 粒度的日志記錄，這些日志都會通過 Store 組件持久化，用法如下：
func (a *Action) Run(ctx run.ExecuteContext, params interface{}) error {
ctx.Trace("some message")
return nil
}使用Dag變量
上面的文章中提到，我們可以在 Dag 中定義一些變量，在創(chuàng)建工作流時可以對這些變量進行賦值，比如以下的Dag，定義了一個名為 `fileName 的變量
id: "test-dag" name: "test" vars: fileName: desc: "the file name" defaultValue: "file.txt"
隨后我們可以使用 Commander 組件來啟動一個具體的工作流：
mod.GetCommander().RunDag("test-id", map[string]string{
"fileName": "demo.txt",
})這樣本次啟動的工作流的變量則被賦值為 demo.txt，接下來我們有兩種方式去消費它
1.帶參數的Action
id: "test-dag"
name: "test"
vars:
fileName:
desc: "the file name"
defaultValue: "file.txt"
tasks:
- id: "task1"
action: "PrintAction"
params:
# using {{var}} to consume dag's variable
fileName: "{{fileName}}"PrintAction.go:
type PrintParams struct {
FileName string `json:"fileName"`
}
type PrintAction struct {
}
// Name define the unique action identity, it will be used by Task
func (a *PrintAction) Name() string {
return "PrintAction"
}
func (a *PrintAction) Run(ctx run.ExecuteContext, params interface{}) error {
cinput := params.(*ActionParam)
fmt.Println(fmt.Sprintf("params: file[%s]", cinput.FileName, cinput.Value))
return nil
}
func (a *PrintAction) ParameterNew() interface{} {
return &PrintParams{}
}2.編程式讀取
fastflow 也提供了相關函數來獲取 Dag 變量
func (a *Action) Run(ctx run.ExecuteContext, params interface{}) error {
// get variable by name
ctx.GetVar("fileName")
// iterate variables
ctx.IterateVars(func(key, val string) (stop bool) {
...
})
return nil
}分布式鎖
如前所述，你可以在直接使用 Keeper 模塊提供的分布式鎖，如下所示：
...
mod.GetKeeper().NewMutex("mutex key").Lock(ctx.Context(),
mod.LockTTL(time.Second),
mod.Reentrant("worker-key1"))
...其中:
LockTTL表示你持有該鎖的TTL，到期之后會自動釋放，默認30sReentrant用于需要實現可重入的分布式鎖的場景，作為持有場景的標識，默認為空，表示該鎖不可重入 歡迎轉載，注明出處即可。如果你覺得這篇博文幫助到你了，請點下右下角的推薦讓更多人看到它。
到此這篇關于基于golang的輕量級工作流框架Fastflow的文章就介紹到這了,更多相關go Fastflow內容請搜索腳本之家以前的文章或繼續(xù)瀏覽下面的相關文章希望大家以后多多支持腳本之家！
相關文章
這篇文章主要介紹了go格式“占位符”, 輸入輸出,類似python的input,本文給大家介紹的非常詳細，具有一定的參考借鑒價值，需要的朋友可以參考下2019-04-04
golang用melody搭建輕量的websocket服務的示例代碼
在Go中,可以使用gin和melody庫來搭建一個輕量級的WebSocket服務,gin是一個流行的Web框架,而melody是一個用于處理WebSocket的庫,本文給大家演示如何使用gin和melody搭建WebSocket服務,感興趣的朋友一起看看吧2023-10-10
在現代 Web 應用中,郵件通知是一個不可或缺的功能,本文將深入解析一個基于 Go 語言 smtp 協議和 email 庫的郵件發(fā)送工具,需要的可以了解下2025-04-04
這篇文章主要介紹了golang語言中for循環(huán)語句用法,實例分析了for循環(huán)遍歷的使用技巧,具有一定參考借鑒價值,需要的朋友可以參考下2015-01-01
備忘錄模式是一種行為設計模式, 允許在不暴露對象實現細節(jié)的情況下保存和恢復對象之前的狀態(tài),本文主要通過一些圖片來給大家介紹一下Go的備忘錄模式,需要的朋友可以參考下2023-08-08
開發(fā)語言上Go成為高并發(fā)業(yè)務開發(fā)的主流語言,再加上云原生技術底座的驅動,Go語言風光無限,下面這篇文章主要給大家介紹了關于Go你不得不知道的一些實用小技巧,需要的朋友可以參考下2022-11-11
這篇文章主要給大家介紹了關于Golang遞歸獲取目錄下所有文件的相關資料,文中通過實例代碼介紹的非常詳細,對大家的學習或者工作具有一定的參考學習價值,需要的朋友可以參考下2023-02-02
這篇文章主要介紹了VScode 中編譯多個Go文件的實現方法，本文通過實例圖文并茂的形式給大家介紹的非常詳細,需要的朋友可以參考下2021-08-08
本文主要為大家詳細介紹了Golang中slice的底層實現，文中通過示例代碼介紹的非常詳細，具有一定的參考價值，感興趣的小伙伴們可以參考一下2023-02-02
這篇文章主要介紹了Golang編程實現刪除字符串中出現次數最少字符的方法,涉及Go語言字符串遍歷與運算相關操作技巧,需要的朋友可以參考下2017-01-01