精品欧美一区二区三区在线观看 _久久久久国色av免费观看性色_国产精品久久在线观看_亚洲第一综合网站_91精品又粗又猛又爽_小泽玛利亚一区二区免费_91亚洲精品国偷拍自产在线观看 _久久精品视频在线播放_美女精品久久久_欧美日韩国产成人在线

如何基于 Go 語言設計一個簡潔優雅的分布式任務系統

開發 后端
本文帶大家從技術選型到架構設計再到代碼實現,一步步完成了一個簡潔優雅的分布式任務系統。

在當今云計算與微服務盛行的時代,分布式任務系統已成為支撐大規模業務的核心基礎設施。今天就來為大家分享下如何基于 Go 語言從零設計和實現一個架構簡潔且擴展性強的分布式任務系統。

一、前置概念

本文會設計并實現一個分布式任務系統,這里我們要先明確兩個概念。

  • 分布式:在我們將要實現的分布式任務系統中,分布式是指我們的服務可以部署多個副本,這樣才能確保服務更加穩定。
  • 任務:這里的任務是指異步任務,可能是定時或需要周期性運行的任務。

有了這兩個前置概念,我們再來分析下在 Go 中如何實現分布式和如何處理異步任務。

二、異步任務

在 Go 中,要處理異步任務有多種方式,比如原生支持的 time.Sleep、time.Timer 或 time.Ticke,再比如一些第三方包 go-co-op/gocron/v2、robfig/cron/v3 或 bamzi/jobrunner 等。本項目在調研過后決定采用 robfig/cron/v3 包(以下簡稱 cron)來處理異步任務,原因如下:

  • cron 是一個非常流行的包,支持標準 crontab 表達式(并且可精確到秒),支持時區、任務鏈等高級功能。
  • 提供秒級精度的任務調度。
  • 輕量級,且能輕松應對各種復雜的定時任務場景。

對于 cron 包的使用,可以參考我的另一篇文章「在 Go 中使用 cron 執行定時任務」,里面有詳細說明。

三、分布式

既然我們的任務系統是分布式的,那么必然要考慮并發安全問題。當多個副本同時讀寫系統資源時,很容易產生競態問題。在分布式場景中,解決競態問題最常用的手段當然是分布式鎖。

Go 中的分布式鎖解決方案也很多,常見的有基于 etcd、Redis、ZooKeeper 等中間件來實現的,因為 Redis 在系統中更加常用,所以本項目采用基于 Redis 實現分布式鎖的解決方案。Go 中有兩個比較常用的第三方包 bsm/redislock 和 go-redsync/redsync 都是基于 Redis 的分布式鎖實現。本項目在調研過后決定采用 go-redsync/redsync 包(以下簡稱 redsync),原因如下:

  • redsync 遵循 Redis 官方推薦的 Redlock 算法,支持多節點,容忍部分節點故障,避免單點問題。
  • 通過多數派機制確保鎖的全局唯一性,降低鎖沖突風險。
  • redsync 是 Redis 官方 唯一推薦的 Go Redis 分布式鎖解決方案,由 Redis 社區背書,長期維護,可靠性高。

對于 redsync 包的使用,可以參考我的另一篇文章「在 Go 中如何使用分布式鎖解決并發問題?」,里面有詳細說明。

四、分布式任務系統

現在我們對分布式任務系統中的分布式和任務都有了明確的認識,并且找到了解決方案。那么接下來就可以設計并實現分布式任務系統了。

1. 功能介紹

我們要實現的分布式任務系統叫 nightwatch,nightwatch 是守夜、值班的意思,那么這套系統的功能也就一目了然了,就是用來 24 小時不停機的執行異步任務的。

nightwatch 要實現的主要功能如下:

現在我們有一個系統,用戶可以在 Web 頁面通過表單提交一個“任務”到關系型數據庫的任務表中。然后 nightwatch 系統會定時的掃描任務表,取出待執行的任務,并根據任務中的配置,到 Kubernetes 中拉起 Job 資源對象,真正的執行任務。此外,nightwatch 還會取出已經開始執行的任務,然后去 Kubernetes 中獲取當前任務對應的 Job 實時狀態,并回寫到數據庫中。直到 Kubernetes 中的 Job 執行完成(或失敗),nightwatch 會標記 Job 在數據庫表中的任務狀態為完成(或失敗)。當任務狀態為完成(或失敗),則任務任務終止,nightwatch 不再掃描出這種狀態的數據。

系統整體架構如下:

nightwatch-component

nightwatch 是系統中一個非常核心的組件,用來控制任務的執行,并同步任務狀態。

2. 架構設計

現在我們知道了 nightwatch 的作用,那么就可以設計其實現架構了。

nightwatch 架構設計如下:

nightwatch-architecture

首先,我們需要思考一個問題,分布式鎖應該在何時使用?

在分布式任務系統中,我們有兩種方式使用分布式鎖來保證并發安全。一種是在執行具體的定時任務時,多個副本之間進行競爭,誰搶到鎖,誰就可以執行任務,未搶到鎖的副本可以選擇性的跳過此次執行周期。另一種是在 nigthwatch 啟動時,就開始搶鎖,多個副本之間誰搶到鎖,誰就去執行任務調度,未搶到鎖的副本則進行周期性的嘗試搶鎖操作,如果當前執行任務調度的副本被終止,那么其他副本就有機會搶到鎖,并執行任務調度。

這兩種方式個各自有不同的使用場景,第一種方式的優勢是能夠實現多副本之間的負載均衡,多個副本都在工作,都有可能搶到鎖并執行任務,不過這種方式不能嚴格控制執行任務的間隔時間,比較適合對間隔時間要求不嚴格的任務。第二種方式實際上只有一個副本在執行任務調度,其他副本是空載狀態,是主備設計,這種方式的好處是能夠嚴格控制任務執行的間隔時間。

nigthwatch 采用第二種方式來使用分布式鎖保證并發安全。所以在 nigthwatch 的架構設計中,在啟動 nigthwatch 時,先將所有的定時任務注冊到任務調度器中,接著就會進行搶鎖操作,只有搶到鎖的副本才能夠執行任務調度。未搶到鎖,則使用一個循環周期性的嘗試搶鎖,直到搶鎖成功。對于搶到鎖的副本,當注冊的任務定時策略達到時,任務調度器就會執行任務。架構圖中的 task 就是我們要實現的異步任務,也是主要業務邏輯,task 組件會從數據庫表中讀取任務,然后在 Kubernetes 中啟動 Job,并同步數據庫和 Kubernetes 資源之間的狀態。

3. 目錄結構

我們現在已經設計好了 nigthwatch 的架構,可以動手進行開發實現了。

以下是 nigthwatch 項目的目錄和文件:

$ tree nightwatch
nightwatch # 項目目錄
├── README.md # README 文件
├── assets # 項目相關的資源目錄
│   ├── docker-compose.yaml # 用于啟動項目依賴的 MariaDB 和 Redis
│   └── schema.sql # 測試數據 SQL
├── cmd # 項目啟動入口
│   └── main.go
├── go.mod
├── go.sum
├── internal # 項目內部包
│   ├── logger.go # 定制日志
│   ├── nightwatch.go # nightwatch 的實現和啟動入口
│   └── watcher # 任務接口和實現
│       ├── all # 任務注冊入口
│       │   └── all.go
│       ├── config.go # 任務配置
│       ├── task # 任務實現,一個可以定時同步 MariaDB 和 Kubernetes 任務狀態的示例程序
│       │   ├── task.go
│       │   └── watcher.go
│       └── watcher.go # 任務接口
└── pkg # 項目公共包
    ├── db # 數據庫實例
    │   ├── mysql.go
    │   └── redis.go
    ├── meta # 元信息
    │   └── where.go # MariaDB where 查詢條件元信息封裝
    ├── model # 任務模型
    │   └── task.go
    ├── store # 數據庫操作接口
    │   ├── helper.go
    │   ├── store.go
    │   └── task.go
    └── util # 工具包
        └── reflect
            └── reflect.go

14 directories, 21 files

這里主要的目錄和文件我都標明了其用途,不必完全記住,你先有個印象,大概知道整個項目的結構。

4. 調用鏈路

為了便于你理解代碼,我畫了一張 nigthwatch 項目的調用鏈路圖:

nightwatch-flow

這個調用鏈路圖指明了 nigthwatch 項目中所有目錄之間的代碼調用關系。根據這張圖,可以看出這是一個非常簡潔的架構。cmd 中的入口函數main會調用internal中的nigthwatch包,nigthwatch是分布式系統實現的關鍵所在,這里實現了任務的注冊和調度,watcher定義了任務的接口,task就是任務的具體實現,task的業務邏輯中會依賴store層來讀寫數據庫,所以store會依賴model和db。

5. 代碼實現

接下來就進入到真正的編碼階段了。

首先我們需要為 nigthwatch 項目的業務設計一張任務表,建表 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 '任務狀態',
  `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='任務表';

為了簡化你理解項目的成本,這里僅定義了最小需要字段。

同時我們可以插入兩條測試數據,用來后續項目功能的驗證:

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數據舉例,任務名是demo-task-1,namespace是default,鏡像是alpine,執行命令是sleep,命令參數是60,狀態為Normal表示待執行。當 nigthwatch 服務掃描到這條數據時,就會在 Kubernetes 中default這個namespace下創建一個name為demo-task-1的 Job,其啟動鏡像為alpine,啟動命令為sleep 60,即睡眠60 秒然后退出。

現在有了數據庫表和測試數據,我們來看看 nigthwatch 代碼是如何實現的。

入口文件 cmd/main.go 實現如下:

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 部分,這是為了讓你對項目文件之間的依賴關系有一個更清晰的認知,后續講解的其他模塊我就只會貼出核心代碼。

main 函數的核心功能如下:

首先會初始化各種依賴包,初始化 Kubernetes clientset 用于后續操作 Job,初始化 MySQL 用于從中讀取任務和更新任務狀態,初始化 Redis 用于實現分布式鎖。接著會使用這些初始化的對象創建一個配置對象 nightwatch.Config。然后使用 cfg.New() 創建一個 nightwatch 實例對象 nw。最后調用 nw.Run(stopCh) 啟動服務。這里為了做優雅退出,還引用了 Kubernetes genericapiserver 優雅退出機制。

這里涉及到的 Kubernetes clientset、MySQL 和 Redis 相關的具體配置細節我就不詳細講解了,咱們還是將主要精力聚焦在 nigthwatch 的主脈絡上。

接下來看下 cfg.New() 代碼實現如下:

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 包實現,用來調度和執行定時任務。并且這個方法內部還實例化了一個 redsync 分布式鎖對象 locker。nightWatch 對象就是通過 runner、locker 和 cfg 來構造的。

這里的核心部分是 addWatchers 的邏輯,其實現如下:

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 是一個接口,定義了異步任務應該實現的方法。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
}

可以看到,要實現一個異步任務,需要實現 Init 方法以及 cron.Job 接口。cron.Job 接口其實只有一個方法定義如下:

type Job interface {
    Run()
}

只要滿足 Watcher 接口的任務,就可以通過 Register 函數注冊到 registry 中。ListWatchers 函數則可以返回注冊到 registry 中全部任務。而 ListWatchers 函數正是在前文講解的 *nightWatch.addWatchers 方法中調用的。

到目前為止,任務如何被注冊到 nightWatch.runner 的過程我們就串起來了。接下來需要關注的兩個點是,調度器 runner 是何時啟動的,以及是何時調用 Register 函數注冊任務的。

我們先來看調度器 runner 是何時啟動的:

https://github.com/jianghushinian/blog-go-example/blob/main/nightwatch/internal/nightwatch.go

// Run 執行異步任務,此方法會阻塞直到關閉 stopCh
func (nw *nightWatch) Run(stopCh <-chanstruct{}) {
    ctx := wait.ContextForChannel(stopCh)

    // 循環加鎖,直到加鎖成功,再去啟動任務
    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
    }

    // 看門狗,實現鎖自動續約
    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 方法中,首先會啟動一個無限循環,定時執行嘗試搶鎖操作,直到搶鎖成功。這與前文中講解的 nightwatch 架構設計是一致的。搶到鎖后,就可以執行 nw.runner.Start() 啟動調度器,執行定時任務了。

此外,在 nightwatch 架構圖中沒有體現的一點是,這里為分布式鎖實現了看門狗機制,用來自動續約。關于 redsync 分布式鎖的自動續約,在我的文章「在 Go 中如何使用分布式鎖解決并發問題?」中有詳細講解。

而這個 Run 方法,就是在 main 函數中通過 nw.Run(stopCh) 調用的。

我們還剩下一個最后要看的核心邏輯是 task 在何時會調用 Register 注冊到 registry 變量中。

還記得前文中講解的 Watcher 接口么,*taskWatcher 實現了這個接口:

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 任務的具體對象,它實現了 watcher.Watcher 接口。可以發現,Register 函數是在 init 函數中調用的,即 task 包被導入時實現自動注冊。

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 (
    // 觸發所有 Watcher 的 init 函數進行注冊
    _ "github.com/jianghushinian/blog-go-example/nightwatch/internal/watcher/task"
)

這里以匿名包的方式導入 task 包。如果我們還有其他的任務實現,則同樣可以參考 task 包的注冊方式,在這里以匿名包形式導入,這也是 all 包名的由來,可以注冊全部的任務。

然后會在 nightwatch 中再次以匿名包的方式導入 all 包:

https://github.com/jianghushinian/blog-go-example/blob/main/nightwatch/internal/nightwatch.go

package nightwatch

import (
    ...
    // 觸發 init 函數
    _ "github.com/jianghushinian/blog-go-example/nightwatch/internal/watcher/all"
)

我們可以總結出任務的注冊流程是,nightwatch 包導入 all 包,all 包會導入 task 包,task 包的 init 函數執行就會完成注冊。所以在入口文件 main.go 導入 nightwatch 包的時候,就會觸發任務的注冊。在調用 nw.Run(stopCh) 啟動服務時,所有的任務已經注冊完成了。

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 狀態任務在 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: 同步中間狀態的任務在 Kubernetes 中的狀態到表中
    gofunc() {
        defer w.wg.Done()
        ctx := context.Background()

        _, tasks, err := w.store.Tasks().List(ctx, meta.WithFilterNot(map[string]any{
            // 排除這幾個狀態
            "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 方法就是用來實現每個 watcher 對象的業務邏輯。比如這里就實現了 task 任務的業務邏輯,它包含兩個功能,在 Run 方法的上半部分代碼中啟動了第一個 goroutine 用來實現將 Normal 狀態任務在 Kubernetes 中啟動,下半部分代碼中啟動了第二個 goroutine 用來實現同步已運行的任務在 Kubernetes 中的 Job 狀態到數據庫表中。

至此,nightwatch 項目就講解完成了。我們一起實現了一個架構簡潔且擴展性強的分布式任務系統。關于 nightwatch 項目中更多的代碼細節你可以跳轉到我的 GitHub 倉庫中查看。

五、總結

本文帶大家從技術選型到架構設計再到代碼實現,一步步完成了一個簡潔優雅的分布式任務系統。這套系統不僅架構簡潔,擴展也非常方便,我們只需要按照 task 的套路實現更多的異步任務,都可以非常方便的方式注冊到 nightwatch 中。

責任編輯:趙寧寧 來源: 令飛編程
相關推薦

2016-09-30 10:13:07

分布式爬蟲系統

2024-09-23 04:00:00

java架構分布式系統

2021-11-01 12:25:56

Redis分布式

2023-03-06 08:14:48

MySQLRedis場景

2018-09-06 22:49:31

分布式架構服務器

2024-08-07 08:15:47

2022-08-01 08:01:04

ID發號器系統

2019-07-19 15:51:11

框架選型分布式

2013-09-11 16:02:00

Spark分布式計算系統

2023-09-04 08:45:07

分布式配置中心Zookeeper

2019-01-28 11:46:53

架構運維技術

2023-10-08 10:49:16

搜索系統分布式系統

2018-05-09 09:44:51

Java分布式系統

2019-12-27 16:00:56

分布式事務框架Java

2022-11-06 19:28:02

分布式鎖etcd云原生

2024-05-07 09:00:41

Go語言令牌桶

2024-07-15 08:25:07

2024-10-29 14:32:45

Golang分布式系統

2020-07-30 09:35:09

Redis分布式鎖數據庫

2022-12-29 08:32:50

xxl-job緩存Schedule
點贊
收藏

51CTO技術棧公眾號

成人性生交大片免费看无遮挡aⅴ| 最近免费观看高清韩国日本大全| 国产成人在线播放视频| 伊人成综合网yiren22| 欧美在线不卡一区| 熟妇熟女乱妇乱女网站| 天天综合永久入口| 日韩av电影一区| 欧美猛交免费看| 99久久综合狠狠综合久久止| 国产xxxx振车| 懂色av蜜臀av粉嫩av分享吧| 久久av一区| 久久国产色av| 男人天堂av电影| 国产精一区二区| 欧美视频一区二区三区…| 亚洲一区在线直播| 亚洲欧美一区二区三| 国模无码大尺度一区二区三区| 性欧美激情精品| 日韩精品一区二区三区在线视频| 老汉色老汉首页av亚洲| 欧美久久一二区| 成人黄色av片| av黄色在线| 久久久91精品国产一区二区精品 | 国产精品v欧美精品v日韩精品| 国产在线观看黄色| 欧美日韩视频一区二区三区| 中文字幕亚洲欧美| 人妻在线日韩免费视频| 久久国产精品免费一区二区三区| 在线一区二区三区做爰视频网站| 3d动漫一区二区三区| 国产美女在线观看| 国产精品天干天干在观线| 久久免费看av| 视频二区在线观看| 丁香激情综合五月| 亚洲a中文字幕| 一级做a爱片久久毛片| 久久国产精品久久w女人spa| 午夜精品久久久久久久白皮肤| 五月天色婷婷丁香| 欧美电影免费| 中文精品99久久国产香蕉| 亚洲国产无码精品| 欧美人妖在线观看| 亚洲精品在线一区二区| 欧美三级视频在线| 丰满女人性猛交| 中文字幕在线观看日本| 国产日韩欧美不卡在线| 茄子视频成人在线观看| 欧美美女色图| 国产亚洲精品免费| 日韩精品极品视频在线观看免费| 天堂av在线7| 久久综合久久综合久久综合| 麻豆一区区三区四区产品精品蜜桃| 香蕉av一区二区三区| 成人av中文字幕| 国产精品国产精品国产专区不卡| 亚洲黄色a级片| 成人免费高清视频在线观看| 国产成人av一区二区三区| 乱精品一区字幕二区| 成人午夜av影视| 国内精品二区| 三区在线观看| 久久久不卡网国产精品二区| 亚洲a∨一区二区三区| av网站在线免费观看| 国产精品久久影院| 干日本少妇视频| 6699嫩草久久久精品影院| 天天综合色天天综合色h| 日韩欧美一区二| 三级成人黄色影院| 欧美日韩一级片网站| 午夜福利123| 久久这里只有精品一区二区| 亚洲丝袜在线视频| 久久久久久久久久97| 欧美日韩亚洲国产精品| 69av在线播放| 亚洲视屏在线观看| 国产制服丝袜一区| 国产精品乱子乱xxxx| 免费人成在线观看网站| 亚洲同性同志一二三专区| av在线com| www.日韩一区| 天堂va在线高清一区| 精品999久久久| 中文字幕av久久爽一区| 久久久久久久久久久9不雅视频| 欧美日韩成人在线观看| 黄色片视频免费| 国产乱色国产精品免费视频| 肥熟一91porny丨九色丨| 撸视在线观看免费视频| 亚洲欧美乱综合| www一区二区www免费| 日韩电影精品| 国产视频精品一区二区三区| 天海翼在线视频| 亚洲女同同性videoxma| 91亚洲国产成人精品性色| 午夜av免费在线观看| 日韩毛片视频在线看| 国产淫片av片久久久久久| 精品国产一区二区三区性色av| 精品视频在线导航| 欧美黑人猛猛猛| 日韩精品亚洲一区| 国产综合av一区二区三区| 18+视频在线观看| 欧美色图天堂网| 亚洲国产精品自拍视频| 雨宫琴音一区二区三区| 国产精品午夜国产小视频| 色欲av永久无码精品无码蜜桃 | 精品一区二区日本| 成人黄视频在线观看| 欧美性猛片aaaaaaa做受| 日批在线观看视频| 这里只有精品在线| 国产精品视频xxxx| 可以直接在线观看的av| 午夜视频一区二区| 人妻巨大乳一二三区| 日韩成人激情| 国产精品香蕉av| 国产露出视频在线观看| 狠狠久久亚洲欧美专区| 国产麻豆剧传媒精品国产av| 欧美日本三区| 7777精品伊久久久大香线蕉语言| 色大18成网站www在线观看| 日本高清不卡一区| 给我看免费高清在线观看| 亚洲欧洲日本mm| 国产精品香蕉视屏| 17videosex性欧美| 亚洲国产一区二区三区在线观看| 九九视频在线免费观看| 国产福利不卡视频| 日韩中文在线字幕| 欧州一区二区三区| 久久国产精品偷| 国产伦精品一区二区三区四区| 中文一区在线播放| 一本色道久久亚洲综合精品蜜桃 | 毛片免费看不卡网站| 亚洲国产欧美在线成人app| 久久久久久免费观看| 国v精品久久久网| 欧美一级爱爱视频| 秋霞午夜一区二区三区视频| 欧美人与性动交a欧美精品| 国产福利第一视频| 亚洲午夜国产一区99re久久| 国产a√精品区二区三区四区| 黄色综合网站| 精品网站在线看| www.成人影院| 一本一道久久a久久精品逆3p| 自拍偷拍第八页| 亚洲欧洲www| 免费看91视频| 国产精品日韩欧美一区| 欧美精品亚洲精品| 国产一区二区精品调教| 久久人人爽人人爽人人片亚洲 | 国产真实乱子伦精品视频| 男人c女人视频| 你懂的在线观看一区二区| 国产98色在线| 老司机av在线免费看| 精品国产乱码久久久久久闺蜜 | 免费在线毛片网站| 精品成人免费观看| 五月天婷婷久久| 亚洲国产成人一区二区三区| 色哟哟在线观看视频| 国产婷婷精品| 最新不卡av| 美女视频免费精品| 国产欧美日韩中文字幕在线| 久久亚洲资源| 国产亚洲欧美视频| 99在线小视频| 色天使久久综合网天天| 欧美三级日本三级| 国产亚洲女人久久久久毛片| 中文字幕在线播放一区二区| 日本中文字幕一区二区视频| ijzzijzzij亚洲大全| 香蕉视频一区| 91久久久久久久| 澳门成人av网| 欧美老少做受xxxx高潮| 国产日本在线视频| 亚洲成人av片在线观看| 中文字幕 国产| 亚洲成人激情自拍| 三级黄色录像视频| 久久久久久久免费视频了| 性鲍视频在线观看| 丝袜美腿亚洲综合| 精品成在人线av无码免费看| 99久久综合狠狠综合久久aⅴ| 久久亚洲国产精品日日av夜夜| 国产精品一区二区三区av| 国产精品久久久久久久久久久不卡 | 日本一二三区视频| 日韩理论片中文av| 91禁男男在线观看| 久久你懂得1024| 国产污在线观看| 国产一区欧美二区| 欧美伦理片在线观看| 午夜在线播放视频欧美| 欧美在线一区视频| 欧美精选一区| 综合一区中文字幕| 久久中文字幕av一区二区不卡| 精品亚洲第一| 福利欧美精品在线| 春色成人在线视频| 免费精品一区| 91精品视频在线免费观看| 国产精品99| 国产精品狼人色视频一区| 成人性生活av| 日本韩国欧美精品大片卡二| 成年女人在线看片| 久久久午夜视频| 暖暖在线中文免费日本| 欧美不卡视频一区发布| 黄网站视频在线观看| www.欧美精品| 日本视频在线| 视频在线观看99| 日韩美女网站| 大胆人体色综合| 成人在线观看免费网站| 美女久久久久久久久久久| 女女色综合影院| 久久夜色精品国产| 色图在线观看| 久久久久久久久久久成人| f2c人成在线观看免费视频| 国语自产精品视频在免费| 成人三级高清视频在线看| 国产91精品不卡视频| 欧美成人ⅴideosxxxxx| 国产精品成人av在线| 亚洲精品大片| 成人高清在线观看| 国产一区丝袜| 蜜桃传媒视频麻豆第一区免费观看| 欧美猛男同性videos| 亚洲一区二区三区四区中文| 小处雏高清一区二区三区| 亚洲色婷婷久久精品av蜜桃| 亚洲高清免费| 国产一区亚洲二区三区| 久久精品国产一区二区| 日韩精品视频网址| 成人精品亚洲人成在线| 女同毛片一区二区三区| 中文字幕不卡的av| 99久久婷婷国产综合| 亚洲18女电影在线观看| 国产精品自拍99| 欧美精品一级二级| 丰满人妻一区二区三区四区53| 日韩精品视频在线播放| av在线播放av| 欧美激情三级免费| 日本免费一区二区三区四区| 成人观看高清在线观看免费| 精品福利一区| 亚洲欧美综合一区| 国产精品va| 亚洲乱码国产一区三区| 国产酒店精品激情| 四虎永久免费影院| 亚洲天堂网中文字| 欧美精品一二三四区| 91精品国产一区二区三区| 午夜成人鲁丝片午夜精品| 久久亚洲成人精品| jk漫画禁漫成人入口| 亚洲精品欧美极品| 国产精品一区二区av日韩在线| 在线观看成人免费| 裸体一区二区| 国产免费a级片| 国产精品日韩精品欧美在线| 91香蕉在线视频| 91精品久久久久久久91蜜桃| 日韩av免费观影| 欧美老女人性生活| 国产人妖一区| 久久综合九色综合网站| 欧美精品不卡| 在线观看国产一级片| 91麻豆.com| 免费在线黄色片| 欧美日韩亚洲综合一区| 日韩毛片在线一区二区毛片| 欧美人在线观看| 日韩一区二区三区四区五区 | 精品高清在线| 国产中文字幕二区| 国产精品亚洲午夜一区二区三区| 超碰人人干人人| 色综合久久久久综合99| 日本精品久久久久| 欧美高清电影在线看| 国产一区二区色噜噜| 日韩国产欧美一区| 国产欧美一区二区三区国产幕精品| 人妻激情偷乱视频一区二区三区| 国产精品乱码一区二区三区软件| 91久久国产综合久久91| 精品无人国产偷自产在线| 国产高清视频色在线www| 99r国产精品视频| 四虎成人精品永久免费av九九| 三年中国国语在线播放免费| 久久亚洲精精品中文字幕早川悠里| 五月天综合在线| 精品免费视频一区二区| 三级资源在线| 99蜜桃在线观看免费视频网站| 亚洲精品一二三区区别| 国产5g成人5g天天爽| 亚洲三级免费电影| 国产精品无码在线播放| 精品国产一区二区三区久久| 香蕉久久一区| 椎名由奈jux491在线播放 | 日韩av密桃| 青青在线视频免费| 久久久国产精品午夜一区ai换脸| 91video| 一区二区三区久久精品| 国产精品4hu.www| 亚洲图片都市激情| 久久97超碰色| 欧美日韩免费高清| 蜜桃视频成人m3u8| 日本婷婷久久久久久久久一区二区| 另类图片国产| av永久免费观看| 欧美猛男男办公室激情| 黄色av电影在线播放| 99热国产免费| 一区二区久久| 老司机福利av| 欧美熟乱第一页| 成人看片免费| 激情一区二区三区| 视频一区在线播放| 自拍偷拍第9页| 欧美一级片免费看| caoporn-草棚在线视频最| 久久日韩精品| 久久精品噜噜噜成人av农村| 精品99在线观看| 亚洲乱码国产乱码精品精天堂| 亚洲成人一区在线观看| 2021狠狠干| av电影天堂一区二区在线观看| 国产寡妇亲子伦一区二区三区四区| 中文字幕亚洲自拍| 国模大尺度视频一区二区| 精品国偷自产一区二区三区| 久久一区二区三区国产精品| 亚洲在线观看av| 欧美精品xxx| 欧美中文一区二区| 2025中文字幕| 色婷婷精品大视频在线蜜桃视频| 黄色小网站在线观看| 国产专区一区二区| 久久se精品一区精品二区| 黄色一级视频在线观看| 亚洲欧美在线一区| 欧美视频精品全部免费观看| www.com毛片| 亚洲免费观看高清| 毛片在线播放网站| 99超碰麻豆| 日本成人在线不卡视频|