基于Golang設(shè)計(jì)一套可控的定時(shí)任務(wù)系統(tǒng)

更新時(shí)間：2023年07月25日 14:38:24 作者：第八共同體

這篇文章主要為大家學(xué)習(xí)介紹了如何基于Golang設(shè)計(jì)一套可控的定時(shí)任務(wù)系統(tǒng)，文中的示例代碼講解詳細(xì)，感興趣的小伙伴可以跟隨小編一起學(xué)習(xí)一下

現(xiàn)在的系統(tǒng)設(shè)計(jì)中，有許多規(guī)律性的功能特征需要用到定時(shí)任務(wù)來(lái)完成，比如每分鐘需要執(zhí)行一次清理數(shù)據(jù)的任務(wù)，每個(gè)月的第一天，需要處理一項(xiàng)什么任務(wù)等等這種，還有一種規(guī)律性的任務(wù)不是以時(shí)間間隔為第一維度切割的，而是如果任務(wù)執(zhí)行完成，不管成功與否，都間隔一段時(shí)間執(zhí)行一次任務(wù)等等。

像上面說(shuō)描述的任務(wù)的特征，都需要我們?nèi)ブ芷谛缘膱?zhí)行任務(wù)主體，如果沒(méi)辦法對(duì)定時(shí)任務(wù)進(jìn)行嚴(yán)格的控制管理，在生產(chǎn)環(huán)境下是非常危險(xiǎn)的。比如上周發(fā)生的一起生產(chǎn)事故，定時(shí)任務(wù)去拉取某服務(wù)器的數(shù)據(jù)，因?yàn)槌绦虍惓＃瑢?dǎo)致產(chǎn)生龐大的協(xié)程拉取對(duì)方數(shù)據(jù)，致使對(duì)方數(shù)據(jù)庫(kù)崩潰。

所以在享受一件技術(shù)帶來(lái)好處的同時(shí)，要嘗試在可控的范圍內(nèi)使用，才是我們調(diào)庫(kù)人員的基本操守，接下里，我們就一起探究一下，我是如何設(shè)計(jì)一套可控的定時(shí)任務(wù)系統(tǒng)的。

功能點(diǎn)：

任務(wù)的自動(dòng)注冊(cè)
任務(wù)的信息管理，包括任務(wù)的cron表達(dá)式，任務(wù)的詳情信息備注
手動(dòng)控制任務(wù)的啟動(dòng)和執(zhí)行
任務(wù)的實(shí)例管理，可查看單次運(yùn)行實(shí)例的任務(wù)日志
任務(wù)優(yōu)先級(jí)
保證至少執(zhí)行一次
...

整體結(jié)構(gòu)如下：

系統(tǒng)分為四個(gè)主要部分，從下到上，依次為：

任務(wù)的執(zhí)行體，完成任務(wù)的主要功能，實(shí)現(xiàn)為單個(gè)rpc方法，方便進(jìn)行獨(dú)立部署，或者手動(dòng)執(zhí)行指定次數(shù)時(shí)進(jìn)行單獨(dú)調(diào)用的邏輯實(shí)現(xiàn)
Worker服務(wù)節(jié)點(diǎn)，該節(jié)點(diǎn)單獨(dú)部署，對(duì)所有調(diào)度的任務(wù)進(jìn)行執(zhí)行
Schedule服務(wù)節(jié)點(diǎn)，該節(jié)點(diǎn)也是單獨(dú)部署，對(duì)Cronweb管理的所有任務(wù)進(jìn)行同步
Cronweb節(jié)點(diǎn)，任務(wù)的信息管理能力，可手動(dòng)停止任務(wù)的執(zhí)行，可調(diào)整任務(wù)的執(zhí)行cron表達(dá)式，可定制任務(wù)執(zhí)行的超時(shí)重啟次數(shù)等等信息

最終，在調(diào)研了大多數(shù)任務(wù)管理的庫(kù)后，選擇了asynq.選擇什么庫(kù)，亦或是自己造輪子，都是根據(jù)業(yè)務(wù)需求來(lái)定的，以能滿足功能需求為首要標(biāo)準(zhǔn)。

Asynq是一個(gè)用于排隊(duì)任務(wù)并與woker異步處理的庫(kù)。它由Redis支持，設(shè)計(jì)為可擴(kuò)展但易于啟動(dòng)。

大致概述Asynq的工作方式：

Client將任務(wù)放在隊(duì)列上
Server從隊(duì)列中刪除任務(wù)，并為每個(gè)任務(wù)啟動(dòng)一個(gè)woker goroutine
任務(wù)是由多名woker同時(shí)處理的

任務(wù)隊(duì)列被用作跨多個(gè)計(jì)算機(jī)分配工作的機(jī)制。系統(tǒng)可以由多個(gè)worker servers 和brokers組成，使位于高可用性和水平規(guī)模。

上面兩段話為asynq的簡(jiǎn)單介紹，剛看起來(lái)可能會(huì)有點(diǎn)抽象，那接下來(lái)，我們先來(lái)詳細(xì)的介紹一下這個(gè)庫(kù)，然后看看我們是如何對(duì)它進(jìn)行封裝和調(diào)整的。

看一個(gè)簡(jiǎn)單的實(shí)例

package main
import (
    "log"
    "github.com/hibiken/asynq"
    "your/app/package/tasks"
)
const redisAddr = "127.0.0.1:6379"
func main() {
    srv := asynq.NewServer(
        asynq.RedisClientOpt{Addr: redisAddr},
        asynq.Config{
            // Specify how many concurrent workers to use
            Concurrency: 10,
            // Optionally specify multiple queues with different priority.
            Queues: map[string]int{
                "critical": 6,
                "default":  3,
                "low":      1,
            },
            // See the godoc for other configuration options
        },
    )
    // mux maps a type to a handler
    mux := asynq.NewServeMux()
    mux.HandleFunc(tasks.TypeEmailDelivery, tasks.HandleEmailDeliveryTask)
    mux.Handle(tasks.TypeImageResize, tasks.NewImageProcessor())
    // ...register other handlers...
    if err := srv.Run(mux); err != nil {
        log.Fatalf("could not run server: %v", err)
    }
}

上面這塊代碼實(shí)現(xiàn)了一個(gè)簡(jiǎn)單的server，可以看到，

asynq庫(kù)深度使用redis組件，所以你的系統(tǒng)如何沒(méi)有接入redis那就要重新考慮了，當(dāng)然現(xiàn)在的系統(tǒng)中redis的使用還是很普遍的。繼續(xù)看，實(shí)例化Server時(shí)，可以指定有多少個(gè)并發(fā)任務(wù)來(lái)處理調(diào)度的task.此項(xiàng)可根據(jù)運(yùn)行調(diào)整進(jìn)行調(diào)整。
另外的Queues是隊(duì)列的配置，此處提供了三個(gè)隊(duì)列，這里的隊(duì)列你可以根據(jù)自己的需要進(jìn)行調(diào)整，值為隊(duì)列的優(yōu)先級(jí)。
此處需要注意的是，如果你指定了嚴(yán)格模式時(shí)，低優(yōu)先級(jí)的任務(wù)，只有等到高優(yōu)先級(jí)的任務(wù)執(zhí)行完成后，才能執(zhí)行。如果高優(yōu)先級(jí)的隊(duì)列中一直有任務(wù)，那么低優(yōu)先級(jí)的任務(wù)可能會(huì)得不到執(zhí)行的機(jī)會(huì)。默認(rèn)的嚴(yán)格模式并沒(méi)有打開，如果你需要此模式，進(jìn)需要提供Config結(jié)構(gòu)體中StrictPriority屬性設(shè)置為true.
然后就是類似于http server中的路由設(shè)置，你可在這里指定需要的不同任務(wù)
最后啟動(dòng)了一個(gè)asynq的Server

任務(wù)的執(zhí)行邏輯

package tasks
import (
    "context"
    "encoding/json"
    "fmt"
    "log"
    "time"
    "github.com/hibiken/asynq"
)
// A list of task types.
const (
    TypeEmailDelivery   = "email:deliver"
    TypeImageResize     = "image:resize"
)
type EmailDeliveryPayload struct {
    UserID     int
    TemplateID string
}
type ImageResizePayload struct {
    SourceURL string
}
//----------------------------------------------
// Write a function NewXXXTask to create a task.
// A task consists of a type and a payload.
//----------------------------------------------
func NewEmailDeliveryTask(userID int, tmplID string) (*asynq.Task, error) {
    payload, err := json.Marshal(EmailDeliveryPayload{UserID: userID, TemplateID: tmplID})
    if err != nil {
        return nil, err
    }
    return asynq.NewTask(TypeEmailDelivery, payload), nil
}
func NewImageResizeTask(src string) (*asynq.Task, error) {
    payload, err := json.Marshal(ImageResizePayload{SourceURL: src})
    if err != nil {
        return nil, err
    }
    // task options can be passed to NewTask, which can be overridden at enqueue time.
    return asynq.NewTask(TypeImageResize, payload, asynq.MaxRetry(5), asynq.Timeout(20 * time.Minute)), nil
}
//---------------------------------------------------------------
// Write a function HandleXXXTask to handle the input task.
// Note that it satisfies the asynq.HandlerFunc interface.
//
// Handler doesn't need to be a function. You can define a type
// that satisfies asynq.Handler interface. See examples below.
//---------------------------------------------------------------
func HandleEmailDeliveryTask(ctx context.Context, t *asynq.Task) error {
    var p EmailDeliveryPayload
    if err := json.Unmarshal(t.Payload(), &p); err != nil {
        return fmt.Errorf("json.Unmarshal failed: %v: %w", err, asynq.SkipRetry)
    }
    log.Printf("Sending Email to User: user_id=%d, template_id=%s", p.UserID, p.TemplateID)
    // Email delivery code ...
    return nil
}
// ImageProcessor implements asynq.Handler interface.
type ImageProcessor struct {
    // ... fields for struct
}
func (processor *ImageProcessor) ProcessTask(ctx context.Context, t *asynq.Task) error {
    var p ImageResizePayload
    if err := json.Unmarshal(t.Payload(), &p); err != nil {
        return fmt.Errorf("json.Unmarshal failed: %v: %w", err, asynq.SkipRetry)
    }
    log.Printf("Resizing image: src=%s", p.SourceURL)
    // Image resizing code ...
    return nil
}
func NewImageProcessor() *ImageProcessor {
	return &ImageProcessor{}
}

上面的代碼定義了不同的任務(wù)類型，以及任務(wù)在調(diào)度過(guò)程中的Payload信息。這個(gè)payload是一個(gè)可以利用的點(diǎn)，在你需要實(shí)現(xiàn)更加高級(jí)的控制能力的時(shí)候，這里我們先簡(jiǎn)單的看看如何使用的即可。

NewEmailDeliveryTask

和

NewImageResizeTask

其實(shí)就是定義了兩個(gè)不同的任務(wù)類型，在實(shí)現(xiàn)時(shí)，你可以指定不同的屬性，比如最大的重試次數(shù)，以及單個(gè)任務(wù)的實(shí)例的超時(shí)時(shí)間等等。

接下來(lái)的兩段代碼，我摘出來(lái)著重說(shuō)一下：

//---------------------------------------------------------------
// Write a function HandleXXXTask to handle the input task.
// Note that it satisfies the asynq.HandlerFunc interface.
//
// Handler doesn't need to be a function. You can define a type
// that satisfies asynq.Handler interface. See examples below.
//---------------------------------------------------------------
func HandleEmailDeliveryTask(ctx context.Context, t *asynq.Task) error {
    var p EmailDeliveryPayload
    if err := json.Unmarshal(t.Payload(), &p); err != nil {
        return fmt.Errorf("json.Unmarshal failed: %v: %w", err, asynq.SkipRetry)
    }
    log.Printf("Sending Email to User: user_id=%d, template_id=%s", p.UserID, p.TemplateID)
    // Email delivery code ...
    return nil
}
// ImageProcessor implements asynq.Handler interface.
type ImageProcessor struct {
    // ... fields for struct
}
func (processor *ImageProcessor) ProcessTask(ctx context.Context, t *asynq.Task) error {
    var p ImageResizePayload
    if err := json.Unmarshal(t.Payload(), &p); err != nil {
        return fmt.Errorf("json.Unmarshal failed: %v: %w", err, asynq.SkipRetry)
    }
    log.Printf("Resizing image: src=%s", p.SourceURL)
    // Image resizing code ...
    return nil
}
func NewImageProcessor() *ImageProcessor {
	return &ImageProcessor{}
}

這其實(shí)就是說(shuō)明了如何將一個(gè)task對(duì)象轉(zhuǎn)換為可以在Server中進(jìn)行路由注冊(cè)的工具方法。你可以把它當(dāng)做是http HandlerFunc一樣對(duì)待，當(dāng)然了，你也可以使用第二種方式，自己定義一個(gè)結(jié)構(gòu)體，然后實(shí)現(xiàn)ProcessTask方法.

Client如何讓任務(wù)進(jìn)行調(diào)度的,三種不同場(chǎng)景下的使用方式

package main
import (
    "log"
    "time"
    "github.com/hibiken/asynq"
    "your/app/package/tasks"
)
const redisAddr = "127.0.0.1:6379"
func main() {
    client := asynq.NewClient(asynq.RedisClientOpt{Addr: redisAddr})
    defer client.Close()
    // ------------------------------------------------------
    // Example 1: Enqueue task to be processed immediately.
    //            Use (*Client).Enqueue method.
    // ------------------------------------------------------
    task, err := tasks.NewEmailDeliveryTask(42, "some:template:id")
    if err != nil {
        log.Fatalf("could not create task: %v", err)
    }
    info, err := client.Enqueue(task)
    if err != nil {
        log.Fatalf("could not enqueue task: %v", err)
    }
    log.Printf("enqueued task: id=%s queue=%s", info.ID, info.Queue)
    // ------------------------------------------------------------
    // Example 2: Schedule task to be processed in the future.
    //            Use ProcessIn or ProcessAt option.
    // ------------------------------------------------------------
    info, err = client.Enqueue(task, asynq.ProcessIn(24*time.Hour))
    if err != nil {
        log.Fatalf("could not schedule task: %v", err)
    }
    log.Printf("enqueued task: id=%s queue=%s", info.ID, info.Queue)
    // ----------------------------------------------------------------------------
    // Example 3: Set other options to tune task processing behavior.
    //            Options include MaxRetry, Queue, Timeout, Deadline, Unique etc.
    // ----------------------------------------------------------------------------
    task, err = tasks.NewImageResizeTask("https://example.com/myassets/image.jpg")
    if err != nil {
        log.Fatalf("could not create task: %v", err)
    }
    info, err = client.Enqueue(task, asynq.MaxRetry(10), asynq.Timeout(3 * time.Minute))
    if err != nil {
        log.Fatalf("could not enqueue task: %v", err)
    }
    log.Printf("enqueued task: id=%s queue=%s", info.ID, info.Queue)
}

上面這部分代碼實(shí)現(xiàn)了三種場(chǎng)景下的使用方法，你可以立即調(diào)用，也可以在未來(lái)的某個(gè)時(shí)間點(diǎn)調(diào)用，還可以更加詳盡的控制任務(wù)執(zhí)行。

上面的代碼僅僅是關(guān)于asynq的簡(jiǎn)單的一個(gè)介紹。在生產(chǎn)環(huán)境下，如何使用呢，一般情況下，我們會(huì)提供一個(gè)provider.provider來(lái)提供配置的源，源可以是文件，也可以是Mysql還可以是其他存儲(chǔ)源，最重要的是需要實(shí)現(xiàn)對(duì)應(yīng)的方法。下面一個(gè)文件源為例來(lái)說(shuō)明如何實(shí)現(xiàn)一個(gè)源

// FileBasedConfigProvider implements asynq.PeriodicTaskConfigProvider interface.
type FileBasedConfigProvider struct {
? ? ? ? filename string
}
type PeriodicTaskConfigContainer struct {
? ? ? ? Configs []*Config `yaml:"configs"`
}
type Config struct {
? ? ? ? Cronspec string `yaml:"cronspec"`
? ? ? ? TaskType string `yaml:"task_type"`
}  
// Parses the yaml file and return a list of PeriodicTaskConfigs.
func (p *FileBasedConfigProvider) GetConfigs() ([]*asynq.PeriodicTaskConfig, error) {
? ? ? ? data, err := os.ReadFile(p.filename)
? ? ? ? if err != nil {
? ? ? ? ? ? ? ? return nil, err
? ? ? ? }
? ? ? ? var c PeriodicTaskConfigContainer
? ? ? ? if err := yaml.Unmarshal(data, &c); err != nil {
? ? ? ? ? ? ? ? return nil, err
? ? ? ? }
? ? ? ? var configs []*asynq.PeriodicTaskConfig
? ? ? ? for _, cfg := range c.Configs {
? ? ? ? ? ? ? ? configs = append(configs, &asynq.PeriodicTaskConfig{Cronspec: cfg.Cronspec, Task: asynq.NewTask(cfg.TaskType, nil)})
? ? ? ? }
? ? ? ? return configs, nil
}

如何使用

        provider := &FileBasedConfigProvider{filename: "./periodic_task_config.yml"}
? ? ? ? mgr, err := asynq.NewPeriodicTaskManager(
? ? ? ? ? ? ? ? asynq.PeriodicTaskManagerOpts{
? ? ? ? ? ? ? ? ? ? ? ? RedisConnOpt: asynq.RedisClientOpt{
? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? Addr: ? ? "127.0.0.1:6379",
? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? Password: "123456",
? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? DB: ? ? ? 1,
? ? ? ? ? ? ? ? ? ? ? ? },
? ? ? ? ? ? ? ? ? ? ? ? PeriodicTaskConfigProvider: provider, ? ? ? ? // this provider object is the interface to your config source
? ? ? ? ? ? ? ? ? ? ? ? SyncInterval: ? ? ? ? ? ? ? 10 * time.Second, // this field specifies how often sync should happen
? ? ? ? ? ? ? ? })
? ? ? ? if err != nil {
? ? ? ? ? ? ? ? log.Fatal(err)
? ? ? ? }
? ? ? ? if err := mgr.Run(); err != nil {
? ? ? ? ? ? ? ? log.Fatal(err)
? ? ? ? }

在任務(wù)調(diào)度的邏輯中，我們一般會(huì)實(shí)現(xiàn)一個(gè)調(diào)度器來(lái)進(jìn)行任務(wù)調(diào)度，而不是單個(gè)任務(wù)進(jìn)行獨(dú)立的調(diào)度，例如：

? ? ? ? loc, err := time.LoadLocation("Asia/Shanghai")
? ? ? ? if err != nil {
            panic(err)
        }
? ? ? ? scheduler := asynq.NewScheduler(
? ? ? ? ? ? ? asynq.RedisClientOpt{
? ? ? ? ? ? ? ? ? ? ? Addr: ? ? "127.0.0.1:6379",
? ? ? ? ? ? ? ? ? ? ? Password: "123456",
? ? ? ? ? ? ? ? ? ? ? DB: ? ? ? 1,
? ? ? ? ? ? ? },
? ? ? ? ? ? ? &asynq.SchedulerOpts{
? ? ? ? ? ? ? ? ? ? ? Location: loc,
? ? ? ? ? ? ? },
? ? ? ?  )
? ? ? ?  task := asynq.NewTask("example_task", nil)
? ? ? ?  // You can use cron spec string to specify the schedule.
? ? ? ?  entryID, err := scheduler.Register("*/1 * * * *", task)
? ? ? ?  if err != nil {
? ? ? ?  ? ? log.Fatal(err)
? ? ? ?  }
? ? ? ?  fmt.Println(entryID)
? ? ? ?  if err := scheduler.Run(); err != nil {
? ? ? ? ? ? ? log.Fatal(err)
? ? ? ?  }

該庫(kù)還提供了一個(gè)工具用于監(jiān)控任務(wù)的運(yùn)行情況。Asynqmon是一個(gè)基于web的工具，用于監(jiān)控和管理Asynq隊(duì)列和任務(wù)。下面是Web UI的一些截圖

以上就是asynq的全部介紹，下面，看看我對(duì)任務(wù)系統(tǒng)的改動(dòng)

首先，增加任務(wù)的注冊(cè)和發(fā)現(xiàn)

客戶端在啟動(dòng)時(shí)，注冊(cè)節(jié)點(diǎn)信息到etcd,并且注冊(cè)該rpc服務(wù)器提供的所有遠(yuǎn)程調(diào)用方法，供web頁(yè)面建立cron任務(wù)時(shí)，進(jìn)行選擇執(zhí)行相應(yīng)的定時(shí)方法，如果采用任務(wù)對(duì)列不需要提供調(diào)用方法，框架自己提供
調(diào)度器定期拉取etcd注冊(cè)信息，維護(hù)web端和客戶端的rpc連接池，方便web端調(diào)度器進(jìn)行rpc連接和rpc調(diào)用

訂閱任務(wù)取消的信號(hào)，對(duì)任務(wù)取消的信號(hào)進(jìn)行處理,在頁(yè)面增加任務(wù)啟停的功能

pubsub := AsyncClient.Subscribe(context.Background(), "asynq:cancel")
	cancelCh := pubsub.Channel()
	ctx, cancel := context.WithCancel(context.Background())
	defer cancel()
	now := time.Now()
	ch1 := make(chan int)
	stop := make(chan bool)
	go func(ctx context.Context, xxx string) {
		err := xxxx
		if err != nil {
			ch1 <- 2
			logx.Error(fmt.Sprintf("本次定時(shí)任務(wù)執(zhí)行失敗,失敗原因: %s", err))
			return
		}
		 for i := 0; i < 100; i++ {
		 	select {
		 	default:
		 		time.Sleep(1 * time.Second)
		 		logx.Info("我是任務(wù)主流程，我在運(yùn)行中...")
		 	case <-stop:
		 		logx.Info("收到退出信號(hào)，任務(wù)結(jié)束...")
		 		return
		 	}
		 }
		usedTime := time.Since(now)
		msg := fmt.Sprintf("本次定時(shí)任務(wù)成功結(jié)束執(zhí)行,用時(shí): %f秒", usedTime.Seconds())
		// Signal the goroutine has completed
		ch1 <- 1
	}(ctx, in.Args)
	for {
		select {
		case cd := <-ch1:
			if cd == 1 {
				return &CommonReply{Message: "本次定時(shí)任務(wù)成功結(jié)束執(zhí)行"}, nil
			} else {
				return &CommonReply{Message: "本次定時(shí)任務(wù)執(zhí)行失敗"}, nil
			}
		case msg := <-cancelCh:
			if msg.Payload == in.TaskId {
				// stop <- true
				respMsg := fmt.Sprintf("任務(wù)[%s]已被取消.Exist.", msg.Payload)
				logx.Info(respMsg)
				return &CommonReply{Message: respMsg}, nil
			}
		}
	}

針對(duì)自身系統(tǒng)的功能需求，可以對(duì)asynq庫(kù)已有的功能進(jìn)行更多的豐富和改進(jìn)，同時(shí)，你也可以借鑒其他優(yōu)秀的庫(kù)，對(duì)現(xiàn)有的庫(kù)進(jìn)行豐富和改造。

以上就是基于Golang設(shè)計(jì)一套可控的定時(shí)任務(wù)系統(tǒng)的詳細(xì)內(nèi)容，更多關(guān)于Golang定時(shí)任務(wù)系統(tǒng)的資料請(qǐng)關(guān)注腳本之家其它相關(guān)文章！

您可能感興趣的文章: