延时任务通知服务的设计及实现（四）-- webhook执行任务

一、本文内容

本文将简单梳理下，延迟任务通知服务的webhook模块实现。

这里的回调接口，请求方式约定为post，参数通过body传递参数。

实现比较简单，先梳理其流程图，再是简单的代码实现。

二、流程图

在这里插入图片描述

三、异步执行任务

对线程池进行自定义其配置。


import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.core.task.AsyncTaskExecutor;
import org.springframework.scheduling.annotation.EnableAsync;
import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;import java.util.concurrent.ThreadPoolExecutor;/*** Async注解的配置.** @author xxx*/
@Configuration
@EnableAsync
public class AsyncConfig {@Bean(name = "taskExecutor")public AsyncTaskExecutor taskExecutor() {ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor();// 核心线程数executor.setCorePoolSize(10);// 最大线程数executor.setMaxPoolSize(50);// 队列容量executor.setQueueCapacity(2000);// 线程名称前缀executor.setThreadNamePrefix("DelayTask-");// 其他配置executor.setKeepAliveSeconds(60);executor.setRejectedExecutionHandler(new ThreadPoolExecutor.CallerRunsPolicy());executor.initialize();return executor;}
}

四、任务通知的代码实现

@Async("taskExecutor")public void handleTask(String transNo, Date notifyDate) {// 分布式锁if (!stringRedisTemplate.opsForValue().setIfAbsent(String.format(LOCK_KEY_TASK_EXECUTE, transNo),"1", 10, TimeUnit.SECONDS)) {log.warn("任务正在执行中, transNo={},notifyDate={}", transNo, notifyDate);return;}// 任务执行前，判断任务的状态NotifyTask notifyTask = this.getNotifyTask(transNo);if (null == notifyTask) {log.warn("任务不存在, transNo={},notifyDate={}", transNo, notifyDate);return;}if (notifyTask.getIsFinished()) {log.warn("任务已执行, transNo={},notifyUrl={},notifyParams={},notifyDate={}",transNo, notifyTask.getNotifyUrl(), notifyTask.getNotifyParams(), notifyDate);return;}// 判断任务的执行时间是否与期望执行时间一致if (null != notifyDate && !DateUtil.isSameTime(notifyDate, notifyTask.getNotifyDate())) {log.warn("任务的执行时间与期望执行时间不一致, transNo={},notifyUrl={},notifyParams={},notifyDate={},expectDate={}",transNo, notifyTask.getNotifyUrl(), notifyTask.getNotifyParams(), notifyDate, notifyTask.getNotifyDate());return;}boolean success = this.callback(notifyTask.getNotifyUrl(), notifyTask.getNotifyParams());if (!success) {// 判断任务是否支持重试if (notifyTask.getIsRetry() && notifyTask.getRetryTimes() < commonConfig.getMaxRetryTimes()) {notifyTask.retry();this.saveOrUpdateNotifyTask(notifyTask);} else {this.removeTaskFromRedis(transNo);}} else {// 更新任务的状态及完成时间notifyTask.finish();this.saveOrUpdateNotifyTask(notifyTask);}}private boolean callback(String notifyUrl, String notifyParams) {if (log.isInfoEnabled()) {log.info("回调接口, notifyUrl={}, notifyParams={}", notifyUrl, notifyParams);}HttpResponse httpResponse = HttpUtil.createPost(notifyUrl).body(notifyParams).setConnectionTimeout(3000).setReadTimeout(5000).execute();return httpResponse.isOk();}