定时任务的多机处理问题常用方案

ゝ一世哀愁。 2021-11-30 00:34 1247阅读 0赞

业务需要用到定时任务，但是必须考虑到定时任务在集群中可能在同一时刻被多次启动，这不是我们想要的，所以下面收集了一些常用方案，供大家参考。

## 问题 ##

先来说说定时任务在集群中需要解决的问题：

1、如果集群中每台机器都启动定时任务，容易造成数据重复处理的问题

2、如果采用定时任务开关的方式，只一台机器的开关on，其他机器的开关off，可以避免数据重复处理的问题，但是存在单点故障的问题。

## 方案 ##

**①指定执行定时任务的机器**  
在多台机器中选择一台执行定时任务，每次执行的时候回判断当前机器和指定的机器是否一致，一致才会执行。

这种方法可以避免执行多次的情况，但是最明显的缺点就是单点故障问题，当这台指定的机器挂了以后，任务就不会执行了。

**②通过动态的加载配置文件来实现**  
例如：新建多个Timmer的配置文件，对任务进行分配。针对不同的机器加载不同的配置文件，即不同的机器执行不同的任务。

这种方法缺陷就是要有至少两种不同的配置文件，那么维护起来就很麻烦。

**③将定时任务单独执行**

将程序中牵扯到的定时任务，从主程序中剥离出来单独执行。

这种方法缺陷就是增加了开发和维护的成本，如果不是大型项目的话，一般不建议这么做。

**④监听程序**  
对程序进行监听，监听是否有重复执行的定时器任务，有的话则不执行相关的业务逻辑。

**⑤从数据库中读取定时任务**  
由于都是连接同一个数据库，给数据库里的定时任务打上相应的标记，来区别有无重复执行定时任务。

create table timed_task (
    type int(3) not null,
    status tinyint(1) not null,
    exec_timestamp bigint(20) not null,
    interval_time int(11) not null,
    PRIMARY KEY (type)
    );

这个表是用来表示当前查询的任务是否可以执行，如果项目中有很多定时任务，通过type字段可以将多个不同类型的定时器进行区分，从而实现表的共用。

type  				# 将多个不同类型的定时器进行区分
    status 				# 状态，是否可以执行定时任务，0为可执行，1为不可执行
    exec_timestamp		# 上一次定时任务的执行时间
    interval_time		# 时间阈值，以秒为单位

![在这里插入图片描述][20190812000854351.]

这里主要说一下字段`interval_time`，主要用来检测执行定时任务的节点有无出现故障或宕机和防止短时间内定时任务被重复执行的情况。

可以设立一个检查机制，当执行节点出现故障或宕机的时候没有及时把字段status重新设置为0，让这个定时任务重新处于可执行状态的话，那么可以根据字段exec\_timestamp和字段interval\_time和当前时间进行比较。如果当前时间是大于字段exec\_timestamp和字段interval\_time之和的，而字段status始终为1的话，说明出现了故障或宕机，那么就可以去检查程序哪里出现了问题。

定时任务在集群中的执行，因为每台机器可能存在一个几秒的时间差，所以有可能出现  
定时任务已经被某台机器处理过了，处理时间只用了几毫秒，所以字段status被重置为0，让这个定时任务重新处于可执行的状态，但是因为几秒钟的偏差，这时某台机器才到了定时任务的指定定时时间，发现数据库中这个定时任务的字段status为0处于可执行的状态，那么会又重新执行了一次这个定时任务，出现数据重复处理的情况。所以这个字段`interval_time`也可以用来防止重复执行的问题，当前时间如果小于字段exec\_timestamp和字段interval\_time之和的话，并且字段status为0，说明短期内这个定时任务是已经被执行过一次的，那么就不允许再次执行了。

**⑥使用redis的过期机制和分布式锁**  
借用redis的分布式锁来解决数据重复处理的问题，为了防止锁被长久锁定，或者防止客户端崩掉了没有删掉锁，可以用expire加入过期时间。

## 个人实践 ##

我个人是使用第六种方案，就是使用redis的分布式锁来解决定时任务的多机处理问题的。下面会有简单的实现代码，但由于篇幅所限，所以只能将代码放在一个文件下，但这在开发中是不合理的结构，此处只供参考。

注意：下面的代码实践是针对redis单实例的情况下的，而在redis主从架构或redis集群下实现分布式锁会在另一篇文章讲到

package main
    import (
    	"github.com/robfig/cron/v3"
    	"github.com/go-redis/redis"
    	"time"
    	"fmt"
    	"math/rand"
    	"strconv"
    	"crypto/md5"
    )
    
    // 分布式锁的结构体
    type redisLock struct { 
    	key   string
    	value string
    }
    
    var Cron *cron.Cron			// 用到第三方库提供的定时任务，很方便就解决了定时任务的难题， 避免重复造轮子
    var client = redis.NewClient(&redis.Options{ 
    	Addr:     "127.0.0.1:6379",
    	Password: "", // no password set
    	DB:       0,          // use default DB
    })
    
    func StartCronJob() { 
    	Cron = cron.New()
    	cronJobTime := "* * * * *"	// 每分钟执行一次
    
    	_, err := Cron.AddFunc(cronJobTime, cronJob)
    	if err != nil { 
    		fmt.Println(err)
    		return
    	}
    	Cron.Start()		
    }
    
    func cronJob() { 
    	r := redisLock{ }
    	hasLock, err := r.GrabLock(time.Now().Unix(), 101)		// 抢锁
    	if err != nil { 
    		return
    	}
    	if !hasLock { 
    		return
    	}
    	defer r.ReleaseLock()			// 抢到锁的才有资格释放锁
    	fmt.Println("hello world, ", time.Now())
    }
    
    // 抢锁
    func (r *redisLock) GrabLock(dateStamp int64, tp int) (hasLock bool, err error) { 
    	// 随机睡眠（0-20ms），在集群中，往往由于存在几ms的时间偏差，所以可能会导致定时任务
    	// 总是被某几台机器执行，这就很不均衡了，所以随机睡眠，让每台机器都有机会抢到锁去执行定时任务
    	rand.Seed(time.Now().UnixNano())
    	time.Sleep(time.Duration(rand.Intn(20)) * time.Millisecond)
    
    	// key之所以这样设置就是因为如果存在很多定时任务的时候，可以更好的区别开来不同的定时任务
    	key := "cron_job:" + strconv.FormatInt(dateStamp, 10) + ":" + strconv.Itoa(tp)
    
    	// value之所以这样设置是为了在所有获取锁请求的客户端里保持唯一，就是用来保证能安全地释放锁，这个很重要，因为这可以避免误删其他客户端得到的锁
    	ds := strconv.FormatInt(dateStamp, 10) + strconv.FormatInt(int64(tp), 10) + strconv.Itoa(rand.Intn(100))
    	value := fmt.Sprintf("%x", md5.Sum([]byte(ds)))
    	expireTime := 300
    
    	r.key = key
    	r.value = value
    
    	// 上锁
    	hasLock, err = client.SetNX(key, value, time.Duration(expireTime)*time.Second).Result()
    	if err != nil { 
    		fmt.Println(err)
    		return
    	}
    	return
    }
    
    // 释放锁
    func (r *redisLock) ReleaseLock() { 
    	// 为保证原子性，使用lua脚本去删除key
    	delScript := `if redis.call("get", KEYS[1]) == ARGV[1] then return redis.call("del", KEYS[1]) else return 0 end`
    	keys := []string{ r.key}
    	result, err := client.Eval(delScript, keys, r.value).Result()
    	if err != nil || result == 0 { 
    		fmt.Println(err)
    		return
    	}
    }
    
    func main() { 
    	StartCronJob()
    }

首先借助于redis的`setnx`命令来操作，`setnx`本身针对key赋值的时候会判断redis中是否存在这个key，如果有返回-1, 如果没有的话，会直接`set`键值。那`setnx`命令跟`set`命令有啥区别？ `setnx`是原子操作，而`set`不能保证原子性。

超时时间是为了防止如果某个抢到锁的客户端在还没释放锁之前redis就宕机了，那么这个锁就永远不会被释放，其他客户端就抢不到锁，结果就是死锁。

那么为什么执行`eval()`方法可以确保原子性，源于Redis的特性，下面是官网对`eval`命令的解释：在`eval`命令执行`Lua`脚本的时候，`Lua`脚本将被当成一个命令去执行，并且直到`eval`命令执行完成，Redis才会执行其他命令。

最常见的释放锁的方式就是直接使用`del()`方法删除锁，这种不先判断锁的拥有者而直接释放锁的方式，会导致一个客户端释放了另一个客户端的锁。举个例子：A客户端拿到了锁，被某个操作阻塞了很长时间，然后这个锁过了超时时间后被redis自动释放了，然后这个A客户端还以为这个锁还是自己的，就去释放锁了，可实际上锁已经被B客户端拿到了，这样就导致A客户端释放了B客户端的锁。所以单纯的用`DEL`指令有可能造成一个客户端删除了其他客户端的锁。  
![在这里插入图片描述][20190813214446700.]  
所以，为了防止把别人的锁释放了，释放锁之前必须检查一下，当前的value是不是自己设置进去的value，如果不是，就说明锁不是自己的了，不能释放。

## 参考 ##

[http://blog.sina.com.cn/s/blog\_3fba24680102vhb4.html][http_blog.sina.com.cn_s_blog_3fba24680102vhb4.html]  
[https://blog.csdn.net/crazy\_yinchao/article/details/77837357][https_blog.csdn.net_crazy_yinchao_article_details_77837357]  
[https://www.bbsmax.com/A/amd08lBjdg/][https_www.bbsmax.com_A_amd08lBjdg]

[20190812000854351.]: /images/20211129/30b464f5ad8d42e197ad2ef93cce9d6b.png
[20190813214446700.]: /images/20211129/858ae7ee163b4d4ab17cca040a729157.png
[http_blog.sina.com.cn_s_blog_3fba24680102vhb4.html]: http://blog.sina.com.cn/s/blog_3fba24680102vhb4.html
[https_blog.csdn.net_crazy_yinchao_article_details_77837357]: https://blog.csdn.net/crazy_yinchao/article/details/77837357
[https_www.bbsmax.com_A_amd08lBjdg]: https://www.bbsmax.com/A/amd08lBjdg/