如何基于 Go 語言設計一個簡潔優(yōu)雅的分布式任務系統(tǒng)
在當今云計算與微服務盛行的時代，分布式任務系統(tǒng)已成為支撐大規(guī)模業(yè)務的核心基礎設施。今天就來為大家分享下如何基于 Go 語言從零設計和實現(xiàn)一個架構簡潔且擴展性強的分布式任務系統(tǒng)。
一、前置概念
本文會設計并實現(xiàn)一個分布式任務系統(tǒng)，這里我們要先明確兩個概念。
- 分布式：在我們將要實現(xiàn)的分布式任務系統(tǒng)中，分布式是指我們的服務可以部署多個副本，這樣才能確保服務更加穩(wěn)定。
- 任務：這里的任務是指異步任務，可能是定時或需要周期性運行的任務。
有了這兩個前置概念，我們再來分析下在 Go 中如何實現(xiàn)分布式和如何處理異步任務。
二、異步任務
在 Go 中，要處理異步任務有多種方式，比如原生支持的 time.Sleep、time.Timer 或 time.Ticke，再比如一些第三方包 go-co-op/gocron/v2、robfig/cron/v3 或 bamzi/jobrunner 等。本項目在調研過后決定采用 robfig/cron/v3 包（以下簡稱 cron）來處理異步任務，原因如下：
- cron 是一個非常流行的包，支持標準 crontab 表達式（并且可精確到秒），支持時區(qū)、任務鏈等高級功能。
- 提供秒級精度的任務調度。
- 輕量級，且能輕松應對各種復雜的定時任務場景。
對于 cron 包的使用，可以參考我的另一篇文章「在 Go 中使用 cron 執(zhí)行定時任務」，里面有詳細說明。
三、分布式
既然我們的任務系統(tǒng)是分布式的，那么必然要考慮并發(fā)安全問題。當多個副本同時讀寫系統(tǒng)資源時，很容易產(chǎn)生競態(tài)問題。在分布式場景中，解決競態(tài)問題最常用的手段當然是分布式鎖。
Go 中的分布式鎖解決方案也很多，常見的有基于 etcd、Redis、ZooKeeper 等中間件來實現(xiàn)的，因為 Redis 在系統(tǒng)中更加常用，所以本項目采用基于 Redis 實現(xiàn)分布式鎖的解決方案。Go 中有兩個比較常用的第三方包 bsm/redislock 和 go-redsync/redsync 都是基于 Redis 的分布式鎖實現(xiàn)。本項目在調研過后決定采用 go-redsync/redsync 包（以下簡稱 redsync），原因如下：
- redsync 遵循 Redis 官方推薦的 Redlock 算法，支持多節(jié)點，容忍部分節(jié)點故障，避免單點問題。
- 通過多數(shù)派機制確保鎖的全局唯一性，降低鎖沖突風險。
- redsync 是 Redis 官方 唯一推薦的 Go Redis 分布式鎖解決方案，由 Redis 社區(qū)背書，長期維護，可靠性高。
對于 redsync 包的使用，可以參考我的另一篇文章「在 Go 中如何使用分布式鎖解決并發(fā)問題？」，里面有詳細說明。
四、分布式任務系統(tǒng)
現(xiàn)在我們對分布式任務系統(tǒng)中的分布式和任務都有了明確的認識，并且找到了解決方案。那么接下來就可以設計并實現(xiàn)分布式任務系統(tǒng)了。
1. 功能介紹
我們要實現(xiàn)的分布式任務系統(tǒng)叫 nightwatch，nightwatch 是守夜、值班的意思，那么這套系統(tǒng)的功能也就一目了然了，就是用來 24 小時不停機的執(zhí)行異步任務的。
nightwatch 要實現(xiàn)的主要功能如下：
現(xiàn)在我們有一個系統(tǒng)，用戶可以在 Web 頁面通過表單提交一個“任務”到關系型數(shù)據(jù)庫的任務表中。然后 nightwatch 系統(tǒng)會定時的掃描任務表，取出待執(zhí)行的任務，并根據(jù)任務中的配置，到 Kubernetes 中拉起 Job 資源對象，真正的執(zhí)行任務。此外，nightwatch 還會取出已經(jīng)開始執(zhí)行的任務，然后去 Kubernetes 中獲取當前任務對應的 Job 實時狀態(tài)，并回寫到數(shù)據(jù)庫中。直到 Kubernetes 中的 Job 執(zhí)行完成（或失?。琻ightwatch 會標記 Job 在數(shù)據(jù)庫表中的任務狀態(tài)為完成（或失?。．斎蝿諣顟B(tài)為完成（或失?。瑒t任務任務終止，nightwatch 不再掃描出這種狀態(tài)的數(shù)據(jù)。
系統(tǒng)整體架構如下：
nightwatch-component
nightwatch 是系統(tǒng)中一個非常核心的組件，用來控制任務的執(zhí)行，并同步任務狀態(tài)。
2. 架構設計
現(xiàn)在我們知道了 nightwatch 的作用，那么就可以設計其實現(xiàn)架構了。
nightwatch 架構設計如下：
nightwatch-architecture
首先，我們需要思考一個問題，分布式鎖應該在何時使用？
在分布式任務系統(tǒng)中，我們有兩種方式使用分布式鎖來保證并發(fā)安全。一種是在執(zhí)行具體的定時任務時，多個副本之間進行競爭，誰搶到鎖，誰就可以執(zhí)行任務，未搶到鎖的副本可以選擇性的跳過此次執(zhí)行周期。另一種是在 nigthwatch 啟動時，就開始搶鎖，多個副本之間誰搶到鎖，誰就去執(zhí)行任務調度，未搶到鎖的副本則進行周期性的嘗試搶鎖操作，如果當前執(zhí)行任務調度的副本被終止，那么其他副本就有機會搶到鎖，并執(zhí)行任務調度。
這兩種方式個各自有不同的使用場景，第一種方式的優(yōu)勢是能夠實現(xiàn)多副本之間的負載均衡，多個副本都在工作，都有可能搶到鎖并執(zhí)行任務，不過這種方式不能嚴格控制執(zhí)行任務的間隔時間，比較適合對間隔時間要求不嚴格的任務。第二種方式實際上只有一個副本在執(zhí)行任務調度，其他副本是空載狀態(tài)，是主備設計，這種方式的好處是能夠嚴格控制任務執(zhí)行的間隔時間。
nigthwatch 采用第二種方式來使用分布式鎖保證并發(fā)安全。所以在 nigthwatch 的架構設計中，在啟動 nigthwatch 時，先將所有的定時任務注冊到任務調度器中，接著就會進行搶鎖操作，只有搶到鎖的副本才能夠執(zhí)行任務調度。未搶到鎖，則使用一個循環(huán)周期性的嘗試搶鎖，直到搶鎖成功。對于搶到鎖的副本，當注冊的任務定時策略達到時，任務調度器就會執(zhí)行任務。架構圖中的 task 就是我們要實現(xiàn)的異步任務，也是主要業(yè)務邏輯，task 組件會從數(shù)據(jù)庫表中讀取任務，然后在 Kubernetes 中啟動 Job，并同步數(shù)據(jù)庫和 Kubernetes 資源之間的狀態(tài)。
3. 目錄結構
我們現(xiàn)在已經(jīng)設計好了 nigthwatch 的架構，可以動手進行開發(fā)實現(xiàn)了。
以下是 nigthwatch 項目的目錄和文件：
$ tree nightwatch
nightwatch # 項目目錄
├── README.md # README 文件
├── assets # 項目相關的資源目錄
│ ├── docker-compose.yaml # 用于啟動項目依賴的 MariaDB 和 Redis
│ └── schema.sql # 測試數(shù)據(jù) SQL
├── cmd # 項目啟動入口
│ └── main.go
├── go.mod
├── go.sum
├── internal # 項目內部包
│ ├── logger.go # 定制日志
│ ├── nightwatch.go # nightwatch 的實現(xiàn)和啟動入口
│ └── watcher # 任務接口和實現(xiàn)
│ ├── all # 任務注冊入口
│ │ └── all.go
│ ├── config.go # 任務配置
│ ├── task # 任務實現(xiàn)，一個可以定時同步 MariaDB 和 Kubernetes 任務狀態(tài)的示例程序
│ │ ├── task.go
│ │ └── watcher.go
│ └── watcher.go # 任務接口
└── pkg # 項目公共包
├── db # 數(shù)據(jù)庫實例
│ ├── mysql.go
│ └── redis.go
├── meta # 元信息
│ └── where.go # MariaDB where 查詢條件元信息封裝
├── model # 任務模型
│ └── task.go
├── store # 數(shù)據(jù)庫操作接口
│ ├── helper.go
│ ├── store.go
│ └── task.go
└── util # 工具包
└── reflect
└── reflect.go
14 directories, 21 files這里主要的目錄和文件我都標明了其用途，不必完全記住，你先有個印象，大概知道整個項目的結構。
4. 調用鏈路
為了便于你理解代碼，我畫了一張 nigthwatch 項目的調用鏈路圖：
nightwatch-flow
這個調用鏈路圖指明了 nigthwatch 項目中所有目錄之間的代碼調用關系。根據(jù)這張圖，可以看出這是一個非常簡潔的架構。cmd 中的入口函數(shù)main會調用internal中的nigthwatch包，nigthwatch是分布式系統(tǒng)實現(xiàn)的關鍵所在，這里實現(xiàn)了任務的注冊和調度，watcher定義了任務的接口，task就是任務的具體實現(xiàn)，task的業(yè)務邏輯中會依賴store層來讀寫數(shù)據(jù)庫，所以store會依賴model和db。
5. 代碼實現(xiàn)
接下來就進入到真正的編碼階段了。
首先我們需要為 nigthwatch 項目的業(yè)務設計一張任務表，建表 SQL 語句如下：
https://github.com/jianghushinian/blog-go-example/blob/main/nightwatch/assets/schema.sql
CREATE TABLE IF NOTEXISTS `task` (
`id` bigint(20) NOT NULL AUTO_INCREMENT,
`name` varchar(45) NOT NULLDEFAULT'' COMMENT '任務名稱',
`namespace` varchar(45) NOT NULLDEFAULT'' COMMENT 'k8s namespace 名稱',
`info` TEXT NOT NULL COMMENT '任務 k8s 相關信息',
`status` varchar(45) NOT NULLDEFAULT'' COMMENT '任務狀態(tài)',
`user_id` bigint(20) NOT NULLDEFAULT'0' COMMENT '用戶 ID',
`created_at` datetime NOT NULLDEFAULTCURRENT_TIMESTAMP,
`updated_at` datetime NOT NULLDEFAULTCURRENT_TIMESTAMPONUPDATECURRENT_TIMESTAMP,
PRIMARY KEY (`id`),
UNIQUE KEY `uk_name_namespace` (`name`, `namespace`)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COMMENT='任務表';為了簡化你理解項目的成本，這里僅定義了最小需要字段。
同時我們可以插入兩條測試數(shù)據(jù)，用來后續(xù)項目功能的驗證：
INSERT INTO `task` (`id`, `name`, `namespace`, `info`, `status`, `user_id`) VALUES (1, 'demo-task-1', 'default', '{"image":"alpine","command":["sleep"],"args":["60"]}', 'Normal', 1);
INSERT INTO `task` (`id`, `name`, `namespace`, `info`, `status`, `user_id`) VALUES (2, 'demo-task-2', 'demo', '{"image":"busybox","command":["sleep"],"args":["3600"]}', 'Normal', 2);拿 ID為1的task數(shù)據(jù)舉例，任務名是demo-task-1，namespace是default，鏡像是alpine，執(zhí)行命令是sleep，命令參數(shù)是60，狀態(tài)為Normal表示待執(zhí)行。當 nigthwatch 服務掃描到這條數(shù)據(jù)時，就會在 Kubernetes 中default這個namespace下創(chuàng)建一個name為demo-task-1的 Job，其啟動鏡像為alpine，啟動命令為sleep 60，即睡眠60 秒然后退出。
現(xiàn)在有了數(shù)據(jù)庫表和測試數(shù)據(jù)，我們來看看 nigthwatch 代碼是如何實現(xiàn)的。
入口文件 cmd/main.go 實現(xiàn)如下：
https://github.com/jianghushinian/blog-go-example/blob/main/nightwatch/cmd/main.go
package main
import (
"flag"
"log/slog"
"path/filepath"
"time"
genericapiserver "k8s.io/apiserver/pkg/server"
"k8s.io/client-go/kubernetes"
"k8s.io/client-go/tools/clientcmd"
"k8s.io/client-go/util/homedir"
"github.com/jianghushinian/blog-go-example/nightwatch/internal"
"github.com/jianghushinian/blog-go-example/nightwatch/pkg/db"
)
func main() {
slog.SetLogLoggerLevel(slog.LevelDebug)
var kubecfg *string
if home := homedir.HomeDir(); home != "" {
kubecfg = flag.String("kubeconfig", filepath.Join(home, ".kube", "config"), "Optional absolute path to kubeconfig")
} else {
kubecfg = flag.String("kubeconfig", "", "Absolute path to kubeconfig")
}
config, err := clientcmd.BuildConfigFromFlags("", *kubecfg)
if err != nil {
slog.Error(err.Error())
return
}
config.QPS = 50
config.Burst = 100
clientset, err := kubernetes.NewForConfig(config)
if err != nil {
slog.Error(err.Error())
return
}
cfg := nightwatch.Config{
MySQLOptions: &db.MySQLOptions{
Host: "127.0.0.1:33306",
Username: "root",
Password: "nightwatch",
Database: "nightwatch",
MaxIdleConnections: 100,
MaxOpenConnections: 100,
MaxConnectionLifeTime: time.Duration(10) * time.Second,
},
RedisOptions: &db.RedisOptions{
Addr: "127.0.0.1:36379",
Username: "",
Password: "nightwatch",
Database: 0,
MaxRetries: 3,
MinIdleConns: 0,
DialTimeout: 5 * time.Second,
ReadTimeout: 3 * time.Second,
WriteTimeout: 3 * time.Second,
PoolSize: 10,
},
Clientset: clientset,
}
nw, err := cfg.New()
if err != nil {
slog.Error(err.Error())
return
}
stopCh := genericapiserver.SetupSignalHandler()
nw.Run(stopCh)
}因為 main.go 非常重要，是整個程序的入口，所以我就把完整代碼都貼出來了，包括 import 部分，這是為了讓你對項目文件之間的依賴關系有一個更清晰的認知，后續(xù)講解的其他模塊我就只會貼出核心代碼。
main 函數(shù)的核心功能如下：
首先會初始化各種依賴包，初始化 Kubernetes clientset 用于后續(xù)操作 Job，初始化 MySQL 用于從中讀取任務和更新任務狀態(tài)，初始化 Redis 用于實現(xiàn)分布式鎖。接著會使用這些初始化的對象創(chuàng)建一個配置對象 nightwatch.Config。然后使用 cfg.New() 創(chuàng)建一個 nightwatch 實例對象 nw。最后調用 nw.Run(stopCh) 啟動服務。這里為了做優(yōu)雅退出，還引用了 Kubernetes genericapiserver 優(yōu)雅退出機制。
這里涉及到的 Kubernetes clientset、MySQL 和 Redis 相關的具體配置細節(jié)我就不詳細講解了，咱們還是將主要精力聚焦在 nigthwatch 的主脈絡上。
接下來看下 cfg.New() 代碼實現(xiàn)如下：
https://github.com/jianghushinian/blog-go-example/blob/main/nightwatch/internal/nightwatch.go
// New 通過配置構造一個 nightWatch 對象
func (c *Config) New() (*nightWatch, error) {
rdb, err := db.NewRedis(c.RedisOptions)
if err != nil {
slog.Error(err.Error(), "Failed to create Redis client")
returnnil, err
}
logger := newCronLogger()
runner := cron.New(
cron.WithSeconds(),
cron.WithLogger(logger),
cron.WithChain(cron.SkipIfStillRunning(logger), cron.Recover(logger)),
)
pool := goredis.NewPool(rdb)
lockOpts := []redsync.Option{
redsync.WithRetryDelay(50 * time.Microsecond),
redsync.WithTries(3),
redsync.WithExpiry(defaultExpiration),
}
locker := redsync.New(pool).NewMutex(lockName, lockOpts...)
cfg, err := c.CreateWatcherConfig()
if err != nil {
returnnil, err
}
nw := &nightWatch{runner: runner, locker: locker, config: cfg}
if err := nw.addWatchers(); err != nil {
returnnil, err
}
return nw, nil
}*Config.New 方法會通過配置信息構造一個 nightWatch 對象并返回。這里的 runner 就是異步任務調度器，使用 cron 包實現(xiàn)，用來調度和執(zhí)行定時任務。并且這個方法內部還實例化了一個 redsync 分布式鎖對象 locker。nightWatch 對象就是通過 runner、locker 和 cfg 來構造的。
這里的核心部分是 addWatchers 的邏輯，其實現(xiàn)如下：
https://github.com/jianghushinian/blog-go-example/blob/main/nightwatch/internal/nightwatch.go
// 注冊所有 Watcher 實例到 nightWatch
func (nw *nightWatch) addWatchers() error {
for n, w := range watcher.ListWatchers() {
if err := w.Init(context.Background(), nw.config); err != nil {
slog.Error(err.Error(), "Failed to construct watcher", "watcher", n)
return err
}
spec := watcher.Every3Seconds
if obj, ok := w.(watcher.ISpec); ok {
spec = obj.Spec()
}
if _, err := nw.runner.AddJob(spec, w); err != nil {
slog.Error(err.Error(), "Failed to add job to the cron", "watcher", n)
return err
}
}
returnnil
}*nightWatch.addWatchers 方法用來注冊所有 Watcher 對象到調度器 runner 中。
Watcher 是一個接口，定義了異步任務應該實現(xiàn)的方法。Watcher 接口定義如下：
https://github.com/jianghushinian/blog-go-example/blob/main/nightwatch/internal/watcher/watcher.go
type Watcher interface {
Init(ctx context.Context, config *Config) error
cron.Job
}
type ISpec interface {
Spec() string
}
var (
registryLock = new(sync.Mutex)
registry = make(map[string]Watcher)
)
func Register(watcher Watcher) {
registryLock.Lock()
defer registryLock.Unlock()
name := reflectutil.StructName(watcher)
if _, ok := registry[name]; ok {
panic("duplicate watcher entry: " + name)
}
registry[name] = watcher
}
func ListWatchers()map[string]Watcher {
registryLock.Lock()
defer registryLock.Unlock()
return registry
}可以看到，要實現(xiàn)一個異步任務，需要實現(xiàn) Init 方法以及 cron.Job 接口。cron.Job 接口其實只有一個方法定義如下：
type Job interface {
Run()
}只要滿足 Watcher 接口的任務，就可以通過 Register 函數(shù)注冊到 registry 中。ListWatchers 函數(shù)則可以返回注冊到 registry 中全部任務。而 ListWatchers 函數(shù)正是在前文講解的 *nightWatch.addWatchers 方法中調用的。
到目前為止，任務如何被注冊到 nightWatch.runner 的過程我們就串起來了。接下來需要關注的兩個點是，調度器 runner 是何時啟動的，以及是何時調用 Register 函數(shù)注冊任務的。
我們先來看調度器 runner 是何時啟動的：
https://github.com/jianghushinian/blog-go-example/blob/main/nightwatch/internal/nightwatch.go
// Run 執(zhí)行異步任務，此方法會阻塞直到關閉 stopCh
func (nw *nightWatch) Run(stopCh <-chanstruct{}) {
ctx := wait.ContextForChannel(stopCh)
// 循環(huán)加鎖，直到加鎖成功，再去啟動任務
ticker := time.NewTicker(defaultExpiration + (5 * time.Second))
defer ticker.Stop()
for {
err := nw.locker.LockContext(ctx)
if err == nil {
slog.Debug("Successfully acquired lock", "lockName", lockName)
break
}
slog.Debug("Failed to acquire lock", "lockName", lockName, "err", err)
<-ticker.C
}
// 看門狗，實現(xiàn)鎖自動續(xù)約
ticker = time.NewTicker(extendExpiration)
defer ticker.Stop()
gofunc() {
for {
select {
case <-ticker.C:
if ok, err := nw.locker.ExtendContext(ctx); !ok || err != nil {
slog.Debug("Failed to extend lock", "err", err, "status", ok)
}
case <-ctx.Done():
slog.Debug("Exiting lock watchdog")
return
}
}
}()
// 啟動定時任務
nw.runner.Start()
slog.Info("Successfully started nightwatch server")
// 阻塞等待退出信號
<-stopCh
nw.stop()
}在 *nightWatch.Run 方法中，首先會啟動一個無限循環(huán)，定時執(zhí)行嘗試搶鎖操作，直到搶鎖成功。這與前文中講解的 nightwatch 架構設計是一致的。搶到鎖后，就可以執(zhí)行 nw.runner.Start() 啟動調度器，執(zhí)行定時任務了。
此外，在 nightwatch 架構圖中沒有體現(xiàn)的一點是，這里為分布式鎖實現(xiàn)了看門狗機制，用來自動續(xù)約。關于 redsync 分布式鎖的自動續(xù)約，在我的文章「在 Go 中如何使用分布式鎖解決并發(fā)問題？」中有詳細講解。
而這個 Run 方法，就是在 main 函數(shù)中通過 nw.Run(stopCh) 調用的。
我們還剩下一個最后要看的核心邏輯是 task 在何時會調用 Register 注冊到 registry 變量中。
還記得前文中講解的 Watcher 接口么，*taskWatcher 實現(xiàn)了這個接口：
https://github.com/jianghushinian/blog-go-example/blob/main/nightwatch/internal/watcher/task/watcher.go
var _ watcher.Watcher = (*taskWatcher)(nil)
type taskWatcher struct {
store store.IStore
clientset kubernetes.Interface
wg sync.WaitGroup
}
func (w *taskWatcher) Init(ctx context.Context, config *watcher.Config) error {
w.store = config.Store
w.clientset = config.Clientset
returnnil
}
func (w *taskWatcher) Spec() string {
return"@every 30s"
}
func init() {
watcher.Register(&taskWatcher{})
}taskWatcher 就是 task 任務的具體對象，它實現(xiàn)了 watcher.Watcher 接口。可以發(fā)現(xiàn)，Register 函數(shù)是在 init 函數(shù)中調用的，即 task 包被導入時實現(xiàn)自動注冊。
task 包會在 nightwatch/internal/watcher/all/all.go 文件被導入：
https://github.com/jianghushinian/blog-go-example/blob/main/nightwatch/internal/watcher/all/all.go
package all
import (
// 觸發(fā)所有 Watcher 的 init 函數(shù)進行注冊
_ "github.com/jianghushinian/blog-go-example/nightwatch/internal/watcher/task"
)這里以匿名包的方式導入 task 包。如果我們還有其他的任務實現(xiàn)，則同樣可以參考 task 包的注冊方式，在這里以匿名包形式導入，這也是 all 包名的由來，可以注冊全部的任務。
然后會在 nightwatch 中再次以匿名包的方式導入 all 包：
https://github.com/jianghushinian/blog-go-example/blob/main/nightwatch/internal/nightwatch.go
package nightwatch
import (
...
// 觸發(fā) init 函數(shù)
_ "github.com/jianghushinian/blog-go-example/nightwatch/internal/watcher/all"
)我們可以總結出任務的注冊流程是，nightwatch 包導入 all 包，all 包會導入 task 包，task 包的 init 函數(shù)執(zhí)行就會完成注冊。所以在入口文件 main.go 導入 nightwatch 包的時候，就會觸發(fā)任務的注冊。在調用 nw.Run(stopCh) 啟動服務時，所有的任務已經(jīng)注冊完成了。
taskWatcher 對象的核心邏輯當然就是 Run 方法了：
https://github.com/jianghushinian/blog-go-example/blob/main/nightwatch/internal/watcher/task/watcher.go
// Run 運行 task watcher 任務
func (w *taskWatcher) Run() {
w.wg.Add(2)
slog.Debug("Current sync period is start")
// NOTE: 將 Normal 狀態(tài)任務在 Kubernetes 中啟動
gofunc() {
defer w.wg.Done()
ctx := context.Background()
_, tasks, err := w.store.Tasks().List(ctx, meta.WithFilter(map[string]any{
"status": model.TaskStatusNormal,
}))
if err != nil {
slog.Error(err.Error(), "Failed to list tasks")
return
}
var wg sync.WaitGroup
wg.Add(len(tasks))
for _, task := range tasks {
gofunc(task *model.Task) {
defer wg.Done()
job, err := w.clientset.BatchV1().Jobs(task.Namespace).Create(ctx, toJob(task), metav1.CreateOptions{})
if err != nil {
slog.Error(err.Error(), "Failed to create job")
return
}
task.Status = model.TaskStatusPending
if err := w.store.Tasks().Update(ctx, task); err != nil {
slog.Error(err.Error(), "Failed to update task status")
return
}
slog.Info("Successfully created job", "namespace", job.Namespace, "name", job.Name)
}(task)
}
wg.Wait()
}()
// NOTE: 同步中間狀態(tài)的任務在 Kubernetes 中的狀態(tài)到表中
gofunc() {
defer w.wg.Done()
ctx := context.Background()
_, tasks, err := w.store.Tasks().List(ctx, meta.WithFilterNot(map[string]any{
// 排除這幾個狀態(tài)
"status": []model.TaskStatus{model.TaskStatusNormal, model.TaskStatusSucceeded, model.TaskStatusFailed},
}))
if err != nil {
slog.Error(err.Error(), "Failed to list tasks")
return
}
var wg sync.WaitGroup
wg.Add(len(tasks))
for _, task := range tasks {
gofunc(task *model.Task) {
defer wg.Done()
job, err := w.clientset.BatchV1().Jobs(task.Namespace).Get(ctx, task.Name, metav1.GetOptions{})
if err != nil {
slog.Error(err.Error(), "Failed to get task")
return
}
task.Status = toTaskStatus(job)
if err := w.store.Tasks().Update(ctx, task); err != nil {
slog.Error(err.Error(), "Failed to update task status")
return
}
slog.Info("Successfully sync job status to task", "namespace", job.Namespace, "name", job.Name, "status", task.Status)
}(task)
}
wg.Wait()
}()
w.wg.Wait()
slog.Debug("Current sync period is complete")
}Run 方法就是用來實現(xiàn)每個 watcher 對象的業(yè)務邏輯。比如這里就實現(xiàn)了 task 任務的業(yè)務邏輯，它包含兩個功能，在 Run 方法的上半部分代碼中啟動了第一個 goroutine 用來實現(xiàn)將 Normal 狀態(tài)任務在 Kubernetes 中啟動，下半部分代碼中啟動了第二個 goroutine 用來實現(xiàn)同步已運行的任務在 Kubernetes 中的 Job 狀態(tài)到數(shù)據(jù)庫表中。
至此，nightwatch 項目就講解完成了。我們一起實現(xiàn)了一個架構簡潔且擴展性強的分布式任務系統(tǒng)。關于 nightwatch 項目中更多的代碼細節(jié)你可以跳轉到我的 GitHub 倉庫中查看。
五、總結
本文帶大家從技術選型到架構設計再到代碼實現(xiàn)，一步步完成了一個簡潔優(yōu)雅的分布式任務系統(tǒng)。這套系統(tǒng)不僅架構簡潔，擴展也非常方便，我們只需要按照 task 的套路實現(xiàn)更多的異步任務，都可以非常方便的方式注冊到 nightwatch 中。
