基于Golang设计一套可控的定时任务系统

更新时间：2023年07月25日 14:38:24 作者：第八共同体

这篇文章主要为大家学习介绍了如何基于Golang设计一套可控的定时任务系统，文中的示例代码讲解详细，感兴趣的小伙伴可以跟随小编一起学习一下

现在的系统设计中，有许多规律性的功能特征需要用到定时任务来完成，比如每分钟需要执行一次清理数据的任务，每个月的第一天，需要处理一项什么任务等等这种，还有一种规律性的任务不是以时间间隔为第一维度切割的，而是如果任务执行完成，不管成功与否，都间隔一段时间执行一次任务等等。

像上面说描述的任务的特征，都需要我们去周期性的执行任务主体，如果没办法对定时任务进行严格的控制管理，在生产环境下是非常危险的。比如上周发生的一起生产事故，定时任务去拉取某服务器的数据，因为程序异常，导致产生庞大的协程拉取对方数据，致使对方数据库崩溃。

所以在享受一件技术带来好处的同时，要尝试在可控的范围内使用，才是我们调库人员的基本操守，接下里，我们就一起探究一下，我是如何设计一套可控的定时任务系统的。

功能点：

任务的自动注册
任务的信息管理，包括任务的cron表达式，任务的详情信息备注
手动控制任务的启动和执行
任务的实例管理，可查看单次运行实例的任务日志
任务优先级
保证至少执行一次
...

整体结构如下：

系统分为四个主要部分，从下到上，依次为：

任务的执行体，完成任务的主要功能，实现为单个rpc方法，方便进行独立部署，或者手动执行指定次数时进行单独调用的逻辑实现
Worker服务节点，该节点单独部署，对所有调度的任务进行执行
Schedule服务节点，该节点也是单独部署，对Cronweb管理的所有任务进行同步
Cronweb节点，任务的信息管理能力，可手动停止任务的执行，可调整任务的执行cron表达式，可定制任务执行的超时重启次数等等信息

最终，在调研了大多数任务管理的库后，选择了asynq.选择什么库，亦或是自己造轮子，都是根据业务需求来定的，以能满足功能需求为首要标准。

Asynq是一个用于排队任务并与woker异步处理的库。它由Redis支持，设计为可扩展但易于启动。

大致概述Asynq的工作方式：

Client将任务放在队列上
Server从队列中删除任务，并为每个任务启动一个woker goroutine
任务是由多名woker同时处理的

任务队列被用作跨多个计算机分配工作的机制。系统可以由多个worker servers 和brokers组成，使位于高可用性和水平规模。

上面两段话为asynq的简单介绍，刚看起来可能会有点抽象，那接下来，我们先来详细的介绍一下这个库，然后看看我们是如何对它进行封装和调整的。

看一个简单的实例

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)
    }
}

上面这块代码实现了一个简单的server，可以看到，

asynq库深度使用redis组件，所以你的系统如何没有接入redis那就要重新考虑了，当然现在的系统中redis的使用还是很普遍的。继续看，实例化Server时，可以指定有多少个并发任务来处理调度的task.此项可根据运行调整进行调整。
另外的Queues是队列的配置，此处提供了三个队列，这里的队列你可以根据自己的需要进行调整，值为队列的优先级。
此处需要注意的是，如果你指定了严格模式时，低优先级的任务，只有等到高优先级的任务执行完成后，才能执行。如果高优先级的队列中一直有任务，那么低优先级的任务可能会得不到执行的机会。默认的严格模式并没有打开，如果你需要此模式，进需要提供Config结构体中StrictPriority属性设置为true.
然后就是类似于http server中的路由设置，你可在这里指定需要的不同任务
最后启动了一个asynq的Server

任务的执行逻辑

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{}
}

上面的代码定义了不同的任务类型，以及任务在调度过程中的Payload信息。这个payload是一个可以利用的点，在你需要实现更加高级的控制能力的时候，这里我们先简单的看看如何使用的即可。

NewEmailDeliveryTask

和

NewImageResizeTask

其实就是定义了两个不同的任务类型，在实现时，你可以指定不同的属性，比如最大的重试次数，以及单个任务的实例的超时时间等等。

接下来的两段代码，我摘出来着重说一下：

//---------------------------------------------------------------
// 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{}
}

这其实就是说明了如何将一个task对象转换为可以在Server中进行路由注册的工具方法。你可以把它当做是http HandlerFunc一样对待，当然了，你也可以使用第二种方式，自己定义一个结构体，然后实现ProcessTask方法.

Client如何让任务进行调度的,三种不同场景下的使用方式

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)
}

上面这部分代码实现了三种场景下的使用方法，你可以立即调用，也可以在未来的某个时间点调用，还可以更加详尽的控制任务执行。

上面的代码仅仅是关于asynq的简单的一个介绍。在生产环境下，如何使用呢，一般情况下，我们会提供一个provider.provider来提供配置的源，源可以是文件，也可以是Mysql还可以是其他存储源，最重要的是需要实现对应的方法。下面一个文件源为例来说明如何实现一个源

// 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)
        }

在任务调度的逻辑中，我们一般会实现一个调度器来进行任务调度，而不是单个任务进行独立的调度，例如：

        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)
         }

该库还提供了一个工具用于监控任务的运行情况。Asynqmon是一个基于web的工具，用于监控和管理Asynq队列和任务。下面是Web UI的一些截图

以上就是asynq的全部介绍，下面，看看我对任务系统的改动

首先，增加任务的注册和发现

客户端在启动时，注册节点信息到etcd,并且注册该rpc服务器提供的所有远程调用方法，供web页面建立cron任务时，进行选择执行相应的定时方法，如果采用任务对列不需要提供调用方法，框架自己提供
调度器定期拉取etcd注册信息，维护web端和客户端的rpc连接池，方便web端调度器进行rpc连接和rpc调用

订阅任务取消的信号，对任务取消的信号进行处理,在页面增加任务启停的功能

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("本次定时任务执行失败,失败原因: %s", err))
			return
		}
		 for i := 0; i < 100; i++ {
		 	select {
		 	default:
		 		time.Sleep(1 * time.Second)
		 		logx.Info("我是任务主流程，我在运行中...")
		 	case <-stop:
		 		logx.Info("收到退出信号，任务结束...")
		 		return
		 	}
		 }
		usedTime := time.Since(now)
		msg := fmt.Sprintf("本次定时任务成功结束执行,用时: %f秒", usedTime.Seconds())
		// Signal the goroutine has completed
		ch1 <- 1
	}(ctx, in.Args)
	for {
		select {
		case cd := <-ch1:
			if cd == 1 {
				return &CommonReply{Message: "本次定时任务成功结束执行"}, nil
			} else {
				return &CommonReply{Message: "本次定时任务执行失败"}, nil
			}
		case msg := <-cancelCh:
			if msg.Payload == in.TaskId {
				// stop <- true
				respMsg := fmt.Sprintf("任务[%s]已被取消.Exist.", msg.Payload)
				logx.Info(respMsg)
				return &CommonReply{Message: respMsg}, nil
			}
		}
	}

针对自身系统的功能需求，可以对asynq库已有的功能进行更多的丰富和改进，同时，你也可以借鉴其他优秀的库，对现有的库进行丰富和改造。

以上就是基于Golang设计一套可控的定时任务系统的详细内容，更多关于Golang定时任务系统的资料请关注脚本之家其它相关文章！

您可能感兴趣的文章:

以go为例探究beyla从环境变量BEYLA_OPEN_PORT发现进程原理
这篇文章主要为大家介绍了以golang进程为例,研究beyla从环境变量BEYLA_OPEN_PORT(即通过端口)发现进程的原理,有需要的朋友可以借鉴参考下,希望能够有所帮助,祝大家多多进步,早日升职加薪
2023-12-12
GO语言基本类型String和Slice,Map操作详解
这篇文章主要为大家介绍了GO语言基本类型String和Slice,Map操作示例详解，有需要的朋友可以借鉴参考下，希望能够有所帮助，祝大家多多进步，早日升职加薪
2022-08-08
golang hack插件开发动态链接库实例探究
这篇文章主要为大家介绍了golang hack插件开发动态链接库实例探究,有需要的朋友可以借鉴参考下,希望能够有所帮助,祝大家多多进步,早日升职加薪
2024-01-01
win10下go mod配置方式
这篇文章主要介绍了win10下go mod配置方式，具有很好的参考价值，希望对大家有所帮助。一起跟随小编过来看看吧
2021-04-04
golang规则引擎gengine用法案例
这篇文章主要为大家介绍了golang 规则引擎gengine用法案例,有需要的朋友可以借鉴参考下,希望能够有所帮助,祝大家多多进步,早日升职加薪
2024-01-01
Go语言之自定义集合Set
本文主要介绍的是Go语言的自定义集合Set，文中介绍的很详细，有需要的可以参考学习。
2016-08-08
Go之集合slice的实现
本文主要介绍了Go之集合slice的实现，文中通过示例代码介绍的非常详细，具有一定的参考价值，感兴趣的小伙伴们可以参考一下
2021-08-08
Golang并发利器sync.Once的用法详解
在某些场景下，我们需要初始化一些资源。有时会采用延迟初始化的方式，在真正需要资源的时候才进行初始化。在这种情况下，Go语言中的sync.Once提供一个优雅且并发安全的解决方案，本文将对其进行详细介绍
2023-04-04
Go中字符串处理 fmt.Sprintf与string.Builder的区别对比分析
在Go语言中,我们通常会遇到两种主要的方式来处理和操作字符串：使用fmt.Sprintf函数和string.Builder类型,本文给大家介绍它们在性能和用法上有一些关键区别,感兴趣的朋友跟随小编一起看看吧
2023-11-11
Go实现基于RSA加密算法的接口鉴权
这篇文章主要介绍了Go实现基于RSA加密算法的接口鉴权，对大家的学习或者工作具有一定的参考学习价值，需要的朋友们下面随着小编来一起学习学习吧
2021-06-06