导航：首页 > 开发技术 >

如何使用分布式定时任务

发表于：2025-11-10 作者：千家信息网编辑

千家信息网最后更新 2025年11月10日，本篇内容介绍了"如何使用分布式定时任务"的有关知识，在实际案例的操作过程中，不少人都会遇到这样的困境，接下来就让小编带领大家学习一下如何处理这些情况吧！希望大家仔细阅读，能够学有所成！dq 简介dq

千家信息网最后更新 2025年11月10日如何使用分布式定时任务

本篇内容介绍了"如何使用分布式定时任务"的有关知识，在实际案例的操作过程中，不少人都会遇到这样的困境，接下来就让小编带领大家学习一下如何处理这些情况吧！希望大家仔细阅读，能够学有所成！

dq 简介

dq 封装底层 beanstalkd 操作，分布式存储，延迟、定时设置。重启服务可以重新执行，但是消息不会丢失，因为消息的处理都交由 beanstalkd 完成。

可以看出使用非常简单，同时 dq 中使用了 redis setnx 保证了每个消息只被消费一次。但是在生产者端没有使用 redis 做消息存储，这个和前面描述的一致。

对 dq 的整体架构做了简单介绍，下面就开始正式的探索 :hammer:

生产者 example

func main() {        producer := dq.NewProducer([]dq.Beanstalk{                {                        Endpoint: "localhost:11300",                        Tube:     "tube",                },                {                        Endpoint: "localhost:11301",                        Tube:     "tube",                },        })        for i := 1000; i < 1005; i++ {    // Delay：延迟执行                _, err := producer.Delay([]byte(strconv.Itoa(i)), time.Second*5)    // At：在某一个时刻执行                //_, err := producer.At([]byte(strconv.Itoa(i)), time.Now().Add(time.Second*5))                if err != nil {                        fmt.Println(err)                }        }}

从使用上，简单分为两步：

NewProducer(opts)：将本地队列的端口配置和主题配置传入生产者；
producer.Delay()：使用刚创建好的 生产者，调用它的 Delay() 。将需要异步发送的消息传入，Delay 还需要传入延迟执行的时间。

需要说明的是：创建的 producer 是一个接口，Delay() 只是接口其中的一个方法。后续会其他的方法和内部设计。那我们就继续往下探索吧～～～

深入生产者执行流程

下面从 example 的代码进去，看整个函数的调用链。

初始化

dq.NewProducer([]dq.Beanstalk{{opt1}, {opt2}, ...})       // 初始化生产者        |- NewProducerNode(endpoint, tube)                                                            // endpoint,tube 来自传入的配置数组

紧接着就到 producerNode.go ，这个部分就会牵涉到 beanstalk 的初始化：

NewProducerNode(endpoint, tube)        |- conn: newConnection(endpoint, tube)                |- return &connection{}

这就涉及到 beanstalk：connection.conn -> *beanstalk.Conn。

但是在 newConnection() 中并没有对 beanstalk.Conn 进行初始化，这属于 延迟初始化

Delay

首先是生产者端调用 producer.Delay(data, timesecond) ，就把消息插入到内部队列，timesecond 就是延迟执行的时间。我们来看看 Delay() 到底做了什么？

p.Delay(data, timesecond)        |- p.wrap(data, time)                 // 将 data 和 time 包装到一块                |- p.insert(nodeFn)                        |- node.Delay()                       // for rangre p.node 每一个node都执行一遍 `Delay()`

而 p.insert 就是将上一步封装好的 data 传递给 p{cluster} 的每一个node去执行 node.Delay。

在前面的初始化说过，最开始是没有对 conn 进行初始化，那现在要插入数据，总不能不初始化这个 conn 。

node.Delay()                                                                     // 配置中的每个node都执行 `Delay()`        |- node.conn.get()                                 // 获取node中的conn【conn==nil，就初始化一个conn】        |- _, err := conn.Put(data, deplay, opts...)                |- node.conn.reset()                  // 出现err情况下，如OOM/Timeout等情况 -> 关闭conn，防止泄漏

所以最后 Delay 实际上是执行 tube.Put(data, delay)：

tube.Put(data, delay)        |- tube.Conn.cmd("put", ...)                // 生产者发布job

这里就涉及到 beanstalk 的 Put 操作：首先看看生产者 Put 指令参数说明：

put

：优先级，值越小优先级越高，默认为1024；
：延迟 ready 秒数，在这段时间 job 为 delayed 状态；
：time to run ，允许 worker 执行的最大秒数，如果 worker 在这段时间不能 delete，release，bury job，那么当 job 超时，服务器将自动 release 此job；
：job body的长度，不包含\r\n；
： job body data；

OK。那插入 job 成功，响应什么呢？

INSERTED \r\n

返回的 id 是插入 job 的任务标识。到此 Put 分析完毕，跟着代码走一遍：

tube.Put(data, priority, daley, ttr)        |- tube.Conn.cmd("put", ...)        |- tube.Conn.readResp("INSERTED id")|- return id, err                        // 将id返回

这样我们在 example 中直接可以看到的 生产者 执行的操作就介绍完了。上图，图更好说话：

producer interface

那么除了 example 中使用的 Delay() ，还有其余几个方法：

Producer interface {  At(body []byte, at time.Time) (string, error)  Close() error  Delay(body []byte, delay time.Duration) (string, error)  Revoke(ids string) error}