Dcron.Start() //启动分布式定时任务 Dcron.Stop()//停止定时任务

会产生数据竞争

代码 ` //分布式定时任务路由 Dcron.Start() //启动分布式定时任务

quit := make(chan os.Signal) signal.Notify(quit, syscall.SIGHUP, syscall.SIGINT, syscall.SIGQUIT, syscall.SIGTERM) sign := <-quit //阻塞等待结束信号 utils.Logger().Info().Msgf("%v %v %v %+v", "SERVER_RUN", "server_close_sig", "", sign)

//停止定时任务 Dcron.Stop()

================== WARNING: DATA RACE Read at 0x00c000594038 by goroutine 46: github.com/libi/dcron.(*NodePool).tickerUpdatePool() C:/Users/yunwe/go/pkg/mod/github.com/libi/[email protected]/node_pool.go:77 +0xfb github.com/libi/dcron.(*NodePool).StartPool.func1() C:/Users/yunwe/go/pkg/mod/github.com/libi/[email protected]/node_pool.go:57 +0x39

Previous write at 0x00c000594038 by main goroutine: github.com/libi/dcron.(*Dcron).Stop() C:/Users/yunwe/go/pkg/mod/github.com/libi/[email protected]/dcron.go:159 +0x39 main.main() D:/code_qifan/jump_live/main.go:81 +0x486

Goroutine 46 (running) created at: github.com/libi/dcron.(*NodePool).StartPool() C:/Users/yunwe/go/pkg/mod/github.com/libi/[email protected]/node_pool.go:57 +0x1e5 github.com/libi/dcron.(*Dcron).Start() C:/Users/yunwe/go/pkg/mod/github.com/libi/[email protected]/dcron.go:134 +0x84 main.main() D:/code_qifan/jump_live/main.go:69 +0x271

`

If two machines start the cron tab at the same time, the cron tab will stop after a while, the code is as follows

func main() { drv, _ := dredis.NewDriver(&dredis.Conf{ Host: "10.203.169.19", Port: 1841, Password: "005f4ce0b221abf", }, redis.DialConnectTimeout(time.Second*10)) dcronDemo := dcron.NewDcron("server1", drv, cron.WithSeconds()) //添加多个任务 启动多个节点时 任务会均匀分配给各个节点

dcronDemo.AddFunc("s1 test1", "0 */1 * * * *", func() {
	fmt.Println("执行 service1 test1 任务", time.Now().Format("15:04:05"))
})
dcronDemo.AddFunc("s1 test2", "0 */1 * * * *", func() {
	fmt.Println("执行 service1 test2 任务", time.Now().Format("15:04:05"))
})


dcronDemo.AddFunc("s1 test3", "0 */1 * * * *", func() {
	fmt.Println("执行 service1 test3 任务", time.Now().Format("15:04:05"))
})
dcronDemo.Start()


//测试120分钟后退出
time.Sleep(120 * time.Minute)

}

建议将Duration放到可配置项中,不然 func (np *NodePool) tickerUpdatePool() { tickers := time.NewTicker(time.Second * defaultDuration) for range tickers.C { if np.dcron.isRun { np.updatePool() } } } 直接for循环扫描redis,对性能有影响.

启动新的节点，如果任务中有小于 10s（defaultDuration） 的定时任务， 可能会导致有的任务会同时在两个节点中运行。

解决方案： 在 dcron.Start() 中至少等待 defaultDuration 时间后再运行 d.cr.Start()。但是此方案可能会导致有的任务停滞运行一段时间。

背景

项目中的定时任务越来越多，为了防止任务重复执行曾经使用过的方案：

• 只启用了一个节点。
• 固定循环间隔，使用分布式事务锁。

第一种方案没有容错机制，当单个节点宕机，所有定时任务都无法正常执行。

第二种方案不能跟cron一样灵活设定时间，比如需要设定每天1点执行就必须借助数据库或者其他存储手段去轮询，非常低效。

在对比了市面上主流的分布式定时任务库后，发现要不就是过重，要不就是使用复杂或者不能使用golang无缝接入。所以萌生了开发一个分布式定时任务库。

要解决的痛点主要包括：

1. 高一致可靠性
2. 能复用现在架构内的存储系统，redis或者mysql。
3. 引入足够方便，使用方法足够简单。

原理

在参考了gojob源码后想到了全新的解决思路：

将所有节点存入公共存储（目前基本所有项目都使用redis作为缓存库，所以首先开发了redis支持)后使用一致性hash算法来选举出执行单个任务的节点来保证唯一性，所有节点都按照写入的cron预执行，在任务执行入口处根据一致性hash算法来判断该任务是否应该由当前节点执行。

版本：master 本地测试

leaseRespChan 满了

func (e *EtcdDriver) SetHeartBeat(nodeID string) {
	leaseID, err := e.putKeyWithLease(nodeID, nodeID)
	if err != nil {
		log.Printf("putKeyWithLease error: %v", err)
		return
	}

	leaseRespChan, err := e.cli.KeepAlive(context.Background(), leaseID)

	if err != nil {
		log.Printf("keepalive error:%v", err)
		return
	}

         // 尝试修复
	go func (){
		for {
			select {
			case resp := <-leaseRespChan:
				if resp == nil {
					log.Printf("ectd cli keepalive unexpected nil")
				}
			case <-time.After(businessTimeout):
				log.Printf("ectd cli keepalive timeout")
			}
		}
	} ()
}

我的代码：
drvRedis, _ := redis.NewDriver(&redis.Conf{ Addr: conf.RedisSetting.RedisHost, Password: conf.RedisSetting.RedisPassword, //Wait: true, }) if err := drvRedis.Ping(); err != nil { log.Fatal(err.Error()) return } log.Println("init redis ======") dCronEngine := dcron.NewDcron("cron_server1", drvRedis, cron.WithSeconds())

tg := new(product_platform.TaskGroup)
//每小时执行一次
//dcr.AddFunc("TestTask", "0 0 * * * *", tg.TestTask)
dCronEngine.AddFunc("GetBipData", "0 0 * * * *", tg.GetBipData)
dCronEngine.AddFunc("GetTeaData", "0 0 * * * *", tg.GetTeaData)
dCronEngine.Start()

返回的错误日志： 2022-05-27 20:55:06.386 INFO runtime/proc.go:255 init redis ====== [dcron] 2022/05/27 20:55:06 INFO: addJob 'GetBipData' : 0 0 * * * * [dcron] 2022/05/27 20:55:06 INFO: addJob 'GetTeaData' : 0 0 * * * * [dcron] 2022/05/27 20:55:06 ERR: dcron start node pool error ERR unkown command or protocal error

部署到linux服务器上就出现错误了，本地没有什么事，请问这是什么问题，执行到start方法的时候就出错了？

从mysql批量加入定时，是加一个for然后 for (⋯⋯) { dcron.AddFunc("test n+1","*/3 * * * *",func(){ fmt.Println("执行 test n+1 任务",time.Now().Format("15:04:05")) }) } 这样子添加吗？

1. fix filename spell error
2. change the log interface
3. use rwmutex take the place of mutex
4. use an independent interface dlog.Logger instead of interface {Print ... }

兄弟，既然是分布式定时器库，我觉得想法可以进一步展开。

1. 新增个http这其他rpc接口，支持远程注册定时任务&参数，定时call相关接口
2. 周期式执行/递推式执行周期任务是否可选
3. 大量线上场景并非周期式定时任务，一段时间后触发的任务占绝大多数，能否支持
4. 海量定时任务还可以继续优化，例如只缓存未来10min钟内要执行的定时任务，降低内存压力