SpringBoot中使用redis实现相对可靠的分布式定时任务，适用于订单场景

字数统计: 2.1k字   |   阅读时长≈ 9分

重要的前言

先说说为什么重复造轮子，当时的情况是只需要一个简单的延时任务，做到不丢失即可，引入分布式定时任务框架又觉得太重，因此手动造了个基于redis的简单可靠定时任务。

适用场景，此种做法在单机部署情况下是肯定没问题的，在分布式部署情况下可能会出现问题，比如同一个服务同时重启了，又刚好同时读取redis中的定时任务，还刚好就在那几十毫秒的情况下没有检测到redis锁，此时会导致任务重复读取。

因此，在定时任务的代码中需要容忍可能会出现重复执行的情况，一般做法是用订单状态判断，所以如果不是非常严格的场景，这套做法是够用了 (复杂的定时任务建议用QUARTZ)。

上代码

废话结束，接下来上代码，首先创建定时任务的实体类，保存定时任务执行时间、订单ID、订单类型等。

定时任务实体类

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
public class DelayedInfo implements Delayed {
    private String orderId;
    /**
     * 任务类型（订单类型）
     */
    private Integer delayType;
    /**
     * 当前时间
     */
    private long currentTime;
    /**
     * 任务执行时间=当前时间 + 延时任务执行时间
     */
    private long startTime;
    /**
     * 延时时间,单位为秒
     */
    private long timeOut;
    public DelayedInfo(){}

    public DelayedInfo(String orderId, long currentTime, long timeOut) {
        this.orderId = orderId;
        this.currentTime = currentTime;
        this.startTime = currentTime + timeOut * 1000l;
        this.timeOut=timeOut;
        this.delayType=Contract.ORDER_TYPE_CHARGE;
    }
    
    public DelayedInfo(String orderId, long currentTime, long timeOut,int orderType) {
        this.orderId = orderId;
        this.currentTime = currentTime;
        this.startTime = currentTime + timeOut * 1000l;
        this.timeOut=timeOut;
        this.delayType=orderType;
    }

    @Override
    public long getDelay(TimeUnit unit) {
        return unit.convert(startTime - System.currentTimeMillis(), unit);
    }

    @Override
    public int compareTo(Delayed other) {
        if (other == this){
            return 0;
        }
        if(other instanceof DelayedInfo){
            DelayedInfo otherRequest = (DelayedInfo)other;
            long otherStartTime = otherRequest.getStartTime();
            return (int)(this.startTime - otherStartTime);
        }
        return 0;
    }

    public String getOrderId() {
        return orderId;
    }

    public void setOrderId(String orderId) {
        this.orderId = orderId;
    }

    public long getStartTime() {
        return startTime;
    }

    public void setStartTime(long startTime) {
        this.startTime = startTime;
    }

    public long getCurrentTime() {
        return currentTime;
    }

    public void setCurrentTime(long currentTime) {
        this.currentTime = currentTime;
    }

    @Override
    public String toString() {
        return JacksonUtil.toJSon(this);
    }

	public long getTimeOut() {
		return timeOut;
	}

	public void setTimeOut(long timeOut) {
		this.timeOut = timeOut;
	}

	public Integer getDelayType() {
		return delayType;
	}

	public void setDelayType(Integer delayType) {
		this.delayType = delayType;
	}
    
}

其中delayType、OrderType就是订单类型的枚举标识，可自己定义标识。

定时任务父类

再创建一个定时任务的抽象类，它用来做一些每个定时任务都会有的操作，比如添加定时任务，删除定时任务，扫描redis：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
public abstract class DelayService {
	private DelayQueue<DelayedInfo> delayQueue = new DelayQueue<DelayedInfo>();
	protected Logger logger = LoggerFactory.getLogger(this.getClass());
	StringRedisTemplate redisTemplate;
	public DelayService(StringRedisTemplate redisTemplate) {
		this.redisTemplate = redisTemplate;
	}

	/**
	 * @Title: delayProcess
	 * @Description: 延时任务的具体实现
	 * @author reinershir
	 * @date 2020年12月11日
	 * @param info 定时任务的参数
	 * @return 如返回true，表示任务执行完成，将会从redis中删除定时任务，返回false会重新等待一轮时间再次执行
	 */
	public abstract boolean delayProcess(DelayedInfo info);


	public boolean remove(DelayedInfo order) {
		redisTemplate.delete(getRedisKey(order));
		return delayQueue.remove(order);
	}

	/**
	 * @Title: getRedisKey
	 * @Description: 拼接redis存储的key
	 * @author reinershir
	 * @date 2020年12月14日
	 * @param info
	 * @return
	 */
	private String getRedisKey(DelayedInfo info) {
		return Contract.DELAYED_TASK_KEY + info.getDelayType() + "_" + info.getOrderId();
	}

	/**
	 * @Title: getRedisKeyPrefix
	 * @Description: 获取redis key前缀，用于扫描
	 * @author reinershir
	 * @date 2020年12月14日
	 * @param orderType 订单类型枚举
	 * @return redis key prefix
	 */
	private String getRedisKeyPrefix(OrderType orderType) {
		return Contract.DELAYED_TASK_KEY + orderType.getType();
	}
	
	public DelayedInfo take() throws InterruptedException {
		return this.delayQueue.take();
	}

	/**
	 * @Title: add
	 * @Description:
	 * @author reinershir
	 * @date 2020年12月11日
	 * @param orderId     订单号
	 * @param currentTime 当前时间，直接用System.currentTimeMillis
	 * @param timeout     延时执行时间，单位为秒
	 */
	public void add(String orderId, long currentTime, long timeout) {
		DelayedInfo info = new DelayedInfo(orderId, currentTime, timeout);
		delayQueue.put(info);
		redisTemplate.opsForValue().set(getRedisKey(info), com.hz.dianlailai.utils.JacksonUtil.toJSon(info));
	}
	
	public void add(DelayedInfo info) {
		delayQueue.put(info);
		redisTemplate.opsForValue().set(getRedisKey(info), com.hz.dianlailai.utils.JacksonUtil.toJSon(info));
	}
	
	public void addDelayTask(String orderId, Integer delayType, long timeout) {
		DelayedInfo info = new DelayedInfo(orderId, System.currentTimeMillis(), timeout);
		info.setDelayType(delayType);
		this.add(info);
	}
    
    public void putTask(DelayedInfo info) {
		delayQueue.put(info);
	}

	public void remove(String orderId) {
		delayQueue.forEach((info)->{
			if (info.getOrderId().equals(orderId)) {
				remove(info);
				return;
			}
		});
	}

	// 扫描redis
	public void redisScan(OrderType orderType) {
		final String prefix = getRedisKeyPrefix(orderType) + "*";
		new Thread(new Runnable() {
			@Override
			public void run() {
				logger.info("开始扫描定时任务,类型：{}", orderType);
				Set<String> keys = redisTemplate.keys(prefix + "*");
				if (!CollectionUtils.isEmpty(keys)) {
					// 写到DelayQueue
					for (String key : keys) {
						String val = redisTemplate.opsForValue().get(key);
						//不可使用订阅发布模式，因为KEY失效时服务器挂了定时任务就会丢失
						DelayedInfo info = JacksonUtil.readValue(val, DelayedInfo.class);
						if (info != null) {
							//获取任务的启动时间
							long startTime = info.getDelay(TimeUnit.MILLISECONDS);
							if(startTime<1) {
								//如果执行时间已经过期（即应该要马上执行的任务），加个5秒的锁，保证在执行时没有其它程序读取即可，（所以此定时非完全可靠，需要定时任务中的代码支持重复执行的情况）
								startTime = 5000l;
							}
							//给该任务设置一个状态，表示已经有程序在读取该定时任务了
								if(redisTemplate.opsForValue().setIfAbsent("TASK_LOCK_"+key, info.getOrderId(), startTime, TimeUnit.MILLISECONDS)) {
									logger.info("找到定时任务,任务类型：{},编号：{}，时间：{}分钟", orderType, info.getOrderId(),info.getDelay(TimeUnit.MILLISECONDS)/1000/60);
									DelayService.this.putTask(info);
									//清除占用状态
									redisTemplate.delete("TASK_LOCK_"+key);
								}
							
						}else {
							redisTemplate.delete(key);
						}
					}
				}
				logger.info("类型：{} 定时任务扫描完毕！", orderType);
			}
		}).start();
	}
}

订单定时任务处理类

OK,上面已经建好了父类，接下来可以创建执行业务代码的订单延时任务了，此处我还加了个小功能，那就是执行结果返回true才真的结束任务，如果返回false会再等待一次相同的时间再执行，最多允许失败3次。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
@Service
public class OrderDelayService extends DelayService{

	public OrderDelayService(StringRedisTemplate redisTemplate) {
		super(redisTemplate);
	}

	@Override
	public boolean delayProcess(DelayedInfo info) {
		logger.info("执行检查订单定时任务，订单编号:{}",info.getOrderId());
		ConsummerOrder order = orderService.getByOrderNo(info.getOrderId());
		if(order!=null) {
			return false;
		}
		
		return true;
	}

}

创建定时任务执行类

定时任务创建好后需要有个东西来执行它，因此我创建一个循环执行的线程不断判断是否有任务，如果没有任务，该线程阻塞。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
/**
 * 定时任务监听类，用于不停获取队列里的任务
 * @author reinershir
 *
 */
public class DelayServiceListener extends Thread{
	
	private Logger logger = LoggerFactory.getLogger(getClass());
	
	Map<String, Integer> countStore = new HashMap<>();
	DelayService delayService;
	public DelayServiceListener(DelayService delayService) {
		this.delayService=delayService;
	}

	@Override
	public void run() {
		while (true) {
			try {
				//阻塞，等待任务添加进队列
				DelayedInfo info = delayService.take();
				if (delayService.delayProcess(info)) {
					// 执行完成，删除任务
					delayService.remove(info);
				} else {
					// 记录重试次数，超时3次不再重复执行
					Integer count = countStore.get(info.getOrderId());
					if (count == null) {
						count = 1;
						countStore.put(info.getOrderId(), count);
					} else {
						if (count < 3) {
							countStore.put(info.getOrderId(), ++count);
						} else {
							countStore.remove(info.getOrderId());
						}
					}
					if (count < 3) {
						// 未执行完成，等待下一轮
						delayService.add(info.getOrderId(), System.currentTimeMillis(), info.getTimeOut());
					}
				}
			} catch (Exception e) {
				logger.error("DelayService执行出错！", e);
			}
		}
	}
}


创建定时任务业务处理类

在这里处理你的业务逻辑，代码示例：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
/**
 * 延时任务
 * @author reinershir
 *
 */
@Service
public class MyDelayService extends DelayService{

	@Autowired
	public MyDelayService(StringRedisTemplate redisTemplate) {
		super(redisTemplate);
	}

	@Override
	public boolean delayProcess(DelayedInfo info) {
		logger.info("执行延时任务，订单编号:{}",info.getOrderId());

		return true;
	}

}

最后一个！添加配置类

我们需要在spring启动完成后执行扫描代码，因此还需要一个配置类

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
@Configuration
public class DelayTaskConfig implements ApplicationListener<ApplicationReadyEvent>{

	 @Autowired
	 RedisTemplate<String, String> redisTemplate;

	@Override
	public void onApplicationEvent(ApplicationReadyEvent event) {
		ApplicationContext applicationContext = event.getApplicationContext();
        if (applicationContext != null) {
        	MyDelayService MyDelayService = applicationContext.getBean(MyDelayService.class);
        	//扫描保存在redis里的定时任务
        	MyDelayService.redisScan(此处为自定的订单类型);
        	new DelayServiceListener(MyDelayService).start();
        }

		
	}

}

大功造成，在代码中动态添加定时任务的用法如下，只需注入并调用add方法即可：

1
2
3
4
5
@Autowired
MyDelayService MyDelayService;

//在你的代码中调用添加定时任务方法
MyDelayService.add(订单编号, 订单类型,延时执行时间，单位为秒);

最后redis依赖别忘记加了哦

1
2
3
4
5
6
7
8
9
10
11
12
13
14
<dependency>
	<groupId>org.springframework.boot</groupId>
	<artifactId>spring-boot-starter-data-redis</artifactId>
	<exclusions>
		<exclusion>
			<groupId>org.springframework.boot</groupId>
			<artifactId>spring-boot-starter</artifactId>
		</exclusion>
		<exclusion>
			<groupId>org.slf4j</groupId>
			<artifactId>slf4j-api</artifactId>
		</exclusion>
	</exclusions>
</dependency>

扫一扫，分享到微信

tag:

    缺失模块。
    1、请确保node版本大于6.2
    2、在博客根目录（注意不是yilia-plus根目录）执行以下命令：
    npm i hexo-generator-json-content --save

    3、在根目录_config.yml里添加配置： 

      jsonContent:
    meta: false
    pages: false
    posts:
      title: true
      date: true
      path: true
      text: false
      raw: false
      content: false
      slug: false
      updated: false
      comments: false
      link: false
      permalink: false
      excerpt: false
      categories: false
      tags: true

关于我：多年工作经验，熟悉微高并发服务架构设计与开发，有AI客户机器人开发相关经验。
项目或内推机会请联系邮箱： [email protected]
Update By 2023-6-14