如何基于 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 中。
