From 9cd891556f13db205dbdfd4ba4c5ef4bbcfaf1ea Mon Sep 17 00:00:00 2001 From: fit2cloud-chenyw Date: Thu, 13 Jan 2022 17:28:25 +0800 Subject: [PATCH] =?UTF-8?q?feat:=20=E5=AE=9A=E6=97=B6=E6=8A=A5=E5=91=8A?= =?UTF-8?q?=E4=BD=BF=E7=94=A8=E4=BC=98=E5=85=88=E7=BA=A7=E7=BA=BF=E7=A8=8B?= =?UTF-8?q?=E6=B1=A0?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../base/mapper/ext/ExtTaskMapper.java | 9 ++ .../base/mapper/ext/ExtTaskMapper.xml | 13 ++ .../pool/PriorityThreadPoolExecutor.java | 113 ++++++++++++++++++ .../pool/PriorityThreadPoolProperties.java | 17 +++ .../java/io/dataease/config/AsyncConfig.java | 24 ++++ .../job/sechedule/ScheduleManager.java | 36 +++--- .../job/sechedule/strategy/TaskHandler.java | 44 ++++--- .../strategy/impl/EmailTaskHandler.java | 37 +++++- .../listener/GlobalTaskStartListener.java | 1 + .../plugins/server/XEmailTaskServer.java | 21 +++- 10 files changed, 269 insertions(+), 46 deletions(-) create mode 100644 backend/src/main/java/io/dataease/base/mapper/ext/ExtTaskMapper.java create mode 100644 backend/src/main/java/io/dataease/base/mapper/ext/ExtTaskMapper.xml create mode 100644 backend/src/main/java/io/dataease/commons/pool/PriorityThreadPoolExecutor.java create mode 100644 backend/src/main/java/io/dataease/commons/pool/PriorityThreadPoolProperties.java diff --git a/backend/src/main/java/io/dataease/base/mapper/ext/ExtTaskMapper.java b/backend/src/main/java/io/dataease/base/mapper/ext/ExtTaskMapper.java new file mode 100644 index 0000000000..efec50afcf --- /dev/null +++ b/backend/src/main/java/io/dataease/base/mapper/ext/ExtTaskMapper.java @@ -0,0 +1,9 @@ +package io.dataease.base.mapper.ext; + +public interface ExtTaskMapper { + + int runningCount(Long taskId); + + void resetRunnings(Long taskId); + +} diff --git a/backend/src/main/java/io/dataease/base/mapper/ext/ExtTaskMapper.xml b/backend/src/main/java/io/dataease/base/mapper/ext/ExtTaskMapper.xml new file mode 100644 index 0000000000..213d5ae488 --- /dev/null +++ b/backend/src/main/java/io/dataease/base/mapper/ext/ExtTaskMapper.xml @@ -0,0 +1,13 @@ + + + + + + + + update sys_task_instance set status = -1, info = 'System Interrupt Error' where task_id = #{taskId} and status = 0 + + + diff --git a/backend/src/main/java/io/dataease/commons/pool/PriorityThreadPoolExecutor.java b/backend/src/main/java/io/dataease/commons/pool/PriorityThreadPoolExecutor.java new file mode 100644 index 0000000000..d55d9e8a65 --- /dev/null +++ b/backend/src/main/java/io/dataease/commons/pool/PriorityThreadPoolExecutor.java @@ -0,0 +1,113 @@ +package io.dataease.commons.pool; + +import java.util.concurrent.*; +import java.util.concurrent.atomic.AtomicInteger; +import java.util.concurrent.atomic.AtomicLong; + +import io.dataease.commons.utils.LogUtil; + +public class PriorityThreadPoolExecutor extends ThreadPoolExecutor { + + public static AtomicInteger globaInteger = new AtomicInteger(1); + + private ThreadLocal local = new ThreadLocal() { + @Override + protected Integer initialValue() { + return 1; + } + }; + + public PriorityThreadPoolExecutor(int corePoolSize, int maximumPoolSize, long keepAliveTime, TimeUnit unit) { + super(corePoolSize, maximumPoolSize, keepAliveTime, unit, getWorkQueue()); + } + + public PriorityThreadPoolExecutor(int corePoolSize, int maximumPoolSize, long keepAliveTime, TimeUnit unit, + ThreadFactory threadFactory) { + super(corePoolSize, maximumPoolSize, keepAliveTime, unit, getWorkQueue(), threadFactory); + } + + public PriorityThreadPoolExecutor(int corePoolSize, int maximumPoolSize, long keepAliveTime, TimeUnit unit, + RejectedExecutionHandler handler) { + super(corePoolSize, maximumPoolSize, keepAliveTime, unit, getWorkQueue(), handler); + } + + public PriorityThreadPoolExecutor(int corePoolSize, int maximumPoolSize, long keepAliveTime, TimeUnit unit, + ThreadFactory threadFactory, RejectedExecutionHandler handler) { + super(corePoolSize, maximumPoolSize, keepAliveTime, unit, getWorkQueue(), threadFactory, handler); + } + + protected static PriorityBlockingQueue getWorkQueue() { + return new PriorityBlockingQueue(); + } + + @Override + public void execute(Runnable command) { + int andIncrement = globaInteger.getAndIncrement(); + Integer theadInteger = local.get(); + try { + if (theadInteger == 0) { + this.execute(command, 0); + } else { + this.execute(command, andIncrement); + } + + } finally { + local.set(1); + } + } + + public void execute(Runnable command, int priority) { + super.execute(new PriorityRunnable(command, priority)); + } + + public Future submit(Callable task, int priority) { + local.set(priority); + return super.submit(task); + } + + protected static class PriorityRunnable> + implements Runnable, Comparable> { + private final static AtomicLong seq = new AtomicLong(); + private final long seqNum; + Runnable run; + private int priority; + + public PriorityRunnable(Runnable run, int priority) { + seqNum = seq.getAndIncrement(); + this.run = run; + this.priority = priority; + } + + public int getPriority() { + return priority; + } + + public void setPriority(int priority) { + this.priority = priority; + } + + public Runnable getRun() { + return run; + } + + @Override + public void run() { + LogUtil.info("number " + priority + " is starting..."); + this.run.run(); + } + + @Override + public int compareTo(PriorityRunnable other) { + int res = 0; + if (this.priority == other.priority) { + if (other.run != this.run) {// ASC + res = (seqNum < other.seqNum ? -1 : 1); + } + } else {// DESC + res = this.priority > other.priority ? 1 : -1; + } + return res; + } + } + +} diff --git a/backend/src/main/java/io/dataease/commons/pool/PriorityThreadPoolProperties.java b/backend/src/main/java/io/dataease/commons/pool/PriorityThreadPoolProperties.java new file mode 100644 index 0000000000..efb807ba86 --- /dev/null +++ b/backend/src/main/java/io/dataease/commons/pool/PriorityThreadPoolProperties.java @@ -0,0 +1,17 @@ +package io.dataease.commons.pool; + +import org.springframework.boot.context.properties.ConfigurationProperties; +import org.springframework.stereotype.Component; + +import lombok.Data; + +@ConfigurationProperties(prefix = "detask", ignoreInvalidFields = true) +@Data +@Component +public class PriorityThreadPoolProperties { + + private int corePoolSize = 2; + private int maximumPoolSize = 100; + private int keepAliveTime = 60; + +} diff --git a/backend/src/main/java/io/dataease/config/AsyncConfig.java b/backend/src/main/java/io/dataease/config/AsyncConfig.java index 22a88ef995..8a8786c294 100644 --- a/backend/src/main/java/io/dataease/config/AsyncConfig.java +++ b/backend/src/main/java/io/dataease/config/AsyncConfig.java @@ -1,15 +1,25 @@ package io.dataease.config; +import java.util.concurrent.TimeUnit; + +import javax.annotation.Resource; + 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 io.dataease.commons.pool.PriorityThreadPoolExecutor; +import io.dataease.commons.pool.PriorityThreadPoolProperties; + @EnableAsync(proxyTargetClass = true) @Configuration public class AsyncConfig { + @Resource + private PriorityThreadPoolProperties priorityThreadPoolProperties; + @Bean public AsyncTaskExecutor taskExecutor() { ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor(); @@ -18,4 +28,18 @@ public class AsyncConfig { executor.setMaxPoolSize(10); return executor; } + + @Bean + public PriorityThreadPoolExecutor priorityExecutor() { + int corePoolSize = priorityThreadPoolProperties.getCorePoolSize(); + + int maximumPoolSize = priorityThreadPoolProperties.getMaximumPoolSize(); + + int keepAliveTime = priorityThreadPoolProperties.getKeepAliveTime(); + + PriorityThreadPoolExecutor executor = new PriorityThreadPoolExecutor(corePoolSize, maximumPoolSize, + keepAliveTime, TimeUnit.SECONDS); + return executor; + } + } diff --git a/backend/src/main/java/io/dataease/job/sechedule/ScheduleManager.java b/backend/src/main/java/io/dataease/job/sechedule/ScheduleManager.java index c93e59e11c..3a41798e93 100644 --- a/backend/src/main/java/io/dataease/job/sechedule/ScheduleManager.java +++ b/backend/src/main/java/io/dataease/job/sechedule/ScheduleManager.java @@ -28,7 +28,7 @@ public class ScheduleManager { * @throws SchedulerException */ public void addSimpleJob(JobKey jobKey, TriggerKey triggerKey, Class cls, int repeatIntervalTime, - JobDataMap jobDataMap) throws SchedulerException { + JobDataMap jobDataMap) throws SchedulerException { JobBuilder jobBuilder = JobBuilder.newJob(cls).withIdentity(jobKey); @@ -46,7 +46,8 @@ public class ScheduleManager { scheduler.scheduleJob(jd, trigger); } - public void addSimpleJob(JobKey jobKey, TriggerKey triggerKey, Class cls, int repeatIntervalTime) throws SchedulerException { + public void addSimpleJob(JobKey jobKey, TriggerKey triggerKey, Class cls, int repeatIntervalTime) + throws SchedulerException { addSimpleJob(jobKey, triggerKey, cls, repeatIntervalTime); } @@ -59,7 +60,8 @@ public class ScheduleManager { * @param cron * @param jobDataMap */ - public void addCronJob(JobKey jobKey, TriggerKey triggerKey, Class jobClass, String cron, Date startTime, Date endTime, JobDataMap jobDataMap) { + public void addCronJob(JobKey jobKey, TriggerKey triggerKey, Class jobClass, String cron, Date startTime, + Date endTime, JobDataMap jobDataMap) { try { LogUtil.info("addCronJob: " + triggerKey.getName() + "," + triggerKey.getGroup()); @@ -99,7 +101,8 @@ public class ScheduleManager { } } - public void addCronJob(JobKey jobKey, TriggerKey triggerKey, Class jobClass, String cron, Date startTime, Date endTime) { + public void addCronJob(JobKey jobKey, TriggerKey triggerKey, Class jobClass, String cron, Date startTime, + Date endTime) { addCronJob(jobKey, triggerKey, jobClass, cron, startTime, endTime, null); } @@ -140,7 +143,8 @@ public class ScheduleManager { * @param cron * @throws SchedulerException */ - public void modifyCronJobTime(TriggerKey triggerKey, String cron, Date startTime, Date endTime) throws SchedulerException { + public void modifyCronJobTime(TriggerKey triggerKey, String cron, Date startTime, Date endTime) + throws SchedulerException { LogUtil.info("modifyCronJobTime: " + triggerKey.getName() + "," + triggerKey.getGroup()); @@ -151,7 +155,6 @@ public class ScheduleManager { return; } - /** 方式一 ：调用 rescheduleJob 开始 */ TriggerBuilder triggerBuilder = TriggerBuilder.newTrigger();// 触发器 @@ -279,7 +282,6 @@ public class ScheduleManager { } } - public static void startJobs(Scheduler sched) { try { sched.start(); @@ -289,7 +291,6 @@ public class ScheduleManager { } } - public void shutdownJobs(Scheduler sched) { try { if (!sched.isShutdown()) { @@ -312,7 +313,7 @@ public class ScheduleManager { * @throws SchedulerException */ public void addOrUpdateSimpleJob(JobKey jobKey, TriggerKey triggerKey, Class clz, - int intervalTime, JobDataMap jobDataMap) throws SchedulerException { + int intervalTime, JobDataMap jobDataMap) throws SchedulerException { if (scheduler.checkExists(triggerKey)) { modifySimpleJobTime(triggerKey, intervalTime); @@ -323,7 +324,7 @@ public class ScheduleManager { } public void addOrUpdateSingleJob(JobKey jobKey, TriggerKey triggerKey, Class clz, - Date date, JobDataMap jobDataMap) throws SchedulerException { + Date date, JobDataMap jobDataMap) throws SchedulerException { if (scheduler.checkExists(triggerKey)) { modifySingleJobTime(triggerKey, date); } else { @@ -333,15 +334,15 @@ public class ScheduleManager { } public void addOrUpdateSingleJob(JobKey jobKey, TriggerKey triggerKey, Class clz, - Date date) throws SchedulerException { + Date date) throws SchedulerException { addOrUpdateSingleJob(jobKey, triggerKey, clz, date, null); } - public void addOrUpdateSimpleJob(JobKey jobKey, TriggerKey triggerKey, Class clz, int intervalTime) throws SchedulerException { + public void addOrUpdateSimpleJob(JobKey jobKey, TriggerKey triggerKey, Class clz, int intervalTime) + throws SchedulerException { addOrUpdateSimpleJob(jobKey, triggerKey, clz, intervalTime, null); } - /** * 添加或修改 cronJob * @@ -352,7 +353,8 @@ public class ScheduleManager { * @param jobDataMap * @throws SchedulerException */ - public void addOrUpdateCronJob(JobKey jobKey, TriggerKey triggerKey, Class jobClass, String cron, Date startTime, Date endTime, JobDataMap jobDataMap) throws SchedulerException { + public void addOrUpdateCronJob(JobKey jobKey, TriggerKey triggerKey, Class jobClass, String cron, Date startTime, + Date endTime, JobDataMap jobDataMap) throws SchedulerException { LogUtil.info("AddOrUpdateCronJob: " + jobKey.getName() + "," + triggerKey.getGroup()); @@ -363,7 +365,8 @@ public class ScheduleManager { } } - public void addOrUpdateCronJob(JobKey jobKey, TriggerKey triggerKey, Class jobClass, String cron, Date startTime, Date endTime) throws SchedulerException { + public void addOrUpdateCronJob(JobKey jobKey, TriggerKey triggerKey, Class jobClass, String cron, Date startTime, + Date endTime) throws SchedulerException { addOrUpdateCronJob(jobKey, triggerKey, jobClass, cron, startTime, endTime, null); } @@ -398,7 +401,8 @@ public class ScheduleManager { if (!CronExpression.isValidExpression(cron)) { DataEaseException.throwException("cron :" + cron + " error"); } - return TriggerBuilder.newTrigger().withIdentity("Calculate Date").withSchedule(CronScheduleBuilder.cronSchedule(cron)).build(); + return TriggerBuilder.newTrigger().withIdentity("Calculate Date") + .withSchedule(CronScheduleBuilder.cronSchedule(cron)).build(); } diff --git a/backend/src/main/java/io/dataease/job/sechedule/strategy/TaskHandler.java b/backend/src/main/java/io/dataease/job/sechedule/strategy/TaskHandler.java index c685930a17..87dff10bdf 100644 --- a/backend/src/main/java/io/dataease/job/sechedule/strategy/TaskHandler.java +++ b/backend/src/main/java/io/dataease/job/sechedule/strategy/TaskHandler.java @@ -13,9 +13,6 @@ import java.util.Date; public abstract class TaskHandler implements InitializingBean { - private static final String[] week = {"SUNDAY", "MONDAY", "TUESDAY", "WEDNESDAY", "THURSDAY", "FRIDAY", "SATURDAY"}; - - public void addTask(ScheduleManager scheduleManager, GlobalTaskEntity taskEntity) throws Exception { // 1。首先看看是否过期 Long endTime = taskEntity.getEndTime(); @@ -30,12 +27,11 @@ public abstract class TaskHandler implements InitializingBean { if (ObjectUtils.isNotEmpty(taskEntity.getEndTime())) { new Date(taskEntity.getEndTime()); } - Class executor = this.getClass(); + Class executor = this.getClass(); String cron = cron(taskEntity); scheduleManager.addOrUpdateCronJob(jobKey, triggerKey, executor, cron, start, end, jobDataMap(taskEntity)); } - protected abstract JobDataMap jobDataMap(GlobalTaskEntity taskEntity); private String cron(GlobalTaskEntity taskEntity) { @@ -54,36 +50,34 @@ public abstract class TaskHandler implements InitializingBean { instance.setTime(date); if (taskEntity.getRateType() == 0) { - return - instance.get(Calendar.SECOND) + " " + - instance.get(Calendar.MINUTE) + " " + - instance.get(Calendar.HOUR_OF_DAY) + " * * ?"; + return instance.get(Calendar.SECOND) + " " + + instance.get(Calendar.MINUTE) + " " + + instance.get(Calendar.HOUR_OF_DAY) + " * * ?"; } if (taskEntity.getRateType() == 1) { - return - instance.get(Calendar.SECOND) + " " + - instance.get(Calendar.MINUTE) + " " + - instance.get(Calendar.HOUR_OF_DAY) + " ? * " + - getDayOfWeek(instance); + return instance.get(Calendar.SECOND) + " " + + instance.get(Calendar.MINUTE) + " " + + instance.get(Calendar.HOUR_OF_DAY) + " ? * " + + getDayOfWeek(instance); } if (taskEntity.getRateType() == 2) { - return - instance.get(Calendar.SECOND) + " " + - instance.get(Calendar.MINUTE) + " " + - instance.get(Calendar.HOUR_OF_DAY) + " " + - instance.get(Calendar.DATE) + " * ?"; + return instance.get(Calendar.SECOND) + " " + + instance.get(Calendar.MINUTE) + " " + + instance.get(Calendar.HOUR_OF_DAY) + " " + + instance.get(Calendar.DATE) + " * ?"; } return null; } + public abstract void resetRunningInstance(Long taskId); + private String getDayOfWeek(Calendar instance) { int index = instance.get(Calendar.DAY_OF_WEEK); - index = (index + 1) % 7; + index = (index + 1) % 7; return String.valueOf(index); } - public void removeTask(ScheduleManager scheduleManager, GlobalTaskEntity taskEntity) { JobKey jobKey = new JobKey(taskEntity.getTaskId().toString()); TriggerKey triggerKey = new TriggerKey(taskEntity.getTaskId().toString()); @@ -95,14 +89,16 @@ public abstract class TaskHandler implements InitializingBean { scheduleManager.fireNow(jobKey); } - - //判断任务是否过期 + // 判断任务是否过期 public Boolean taskExpire(Long endTime) { - if (ObjectUtils.isEmpty(endTime)) return false; + if (ObjectUtils.isEmpty(endTime)) + return false; Long now = System.currentTimeMillis(); return now > endTime; } + protected abstract Boolean taskIsRunning(Long taskId); + @Override public void afterPropertiesSet() throws Exception { String beanName = null; diff --git a/backend/src/main/java/io/dataease/job/sechedule/strategy/impl/EmailTaskHandler.java b/backend/src/main/java/io/dataease/job/sechedule/strategy/impl/EmailTaskHandler.java index d84ba736de..12ba9352a2 100644 --- a/backend/src/main/java/io/dataease/job/sechedule/strategy/impl/EmailTaskHandler.java +++ b/backend/src/main/java/io/dataease/job/sechedule/strategy/impl/EmailTaskHandler.java @@ -5,6 +5,7 @@ import io.dataease.auth.entity.TokenInfo; import io.dataease.auth.service.AuthUserService; import io.dataease.auth.service.impl.AuthUserServiceImpl; import io.dataease.auth.util.JWTUtils; +import io.dataease.base.mapper.ext.ExtTaskMapper; import io.dataease.commons.utils.CommonBeanFactory; import io.dataease.commons.utils.LogUtil; import io.dataease.commons.utils.ServletUtils; @@ -20,6 +21,7 @@ import io.dataease.service.system.EmailService; import org.apache.commons.lang3.ObjectUtils; import org.apache.commons.lang3.StringUtils; import org.quartz.*; +import org.springframework.scheduling.annotation.Async; import org.springframework.stereotype.Service; import javax.annotation.Resource; @@ -46,6 +48,16 @@ public class EmailTaskHandler extends TaskHandler implements Job { return jobDataMap; } + public EmailTaskHandler proxy() { + return CommonBeanFactory.getBean(EmailTaskHandler.class); + } + + @Override + protected Boolean taskIsRunning(Long taskId) { + ExtTaskMapper extTaskMapper = CommonBeanFactory.getBean(ExtTaskMapper.class); + return extTaskMapper.runningCount(taskId) > 0; + } + @Override public void execute(JobExecutionContext context) throws JobExecutionException { // 插件没有加载 空转 @@ -54,11 +66,16 @@ public class EmailTaskHandler extends TaskHandler implements Job { JobDataMap jobDataMap = context.getJobDetail().getJobDataMap(); GlobalTaskEntity taskEntity = (GlobalTaskEntity) jobDataMap.get("taskEntity"); + ScheduleManager scheduleManager = SpringContextUtil.getBean(ScheduleManager.class); if (taskExpire(taskEntity.getEndTime())) { - ScheduleManager scheduleManager = SpringContextUtil.getBean(ScheduleManager.class); removeTask(scheduleManager, taskEntity); return; } + if (taskIsRunning(taskEntity.getTaskId())) { + LogUtil.info("Skip synchronization task: {} ,due to task status is {}", + taskEntity.getTaskId(), "running"); + return; + } GlobalTaskInstance taskInstance = buildInstance(taskEntity); Long instanceId = saveInstance(taskInstance); @@ -67,10 +84,15 @@ public class EmailTaskHandler extends TaskHandler implements Job { XpackEmailTemplateDTO emailTemplate = (XpackEmailTemplateDTO) jobDataMap.get("emailTemplate"); SysUserEntity creator = (SysUserEntity) jobDataMap.get("creator"); LogUtil.info("start execute send panel report task..."); - sendReport(taskInstance, emailTemplate, creator); + proxy().sendReport(taskInstance, emailTemplate, creator); } + @Override + public void resetRunningInstance(Long taskId) { + ExtTaskMapper extTaskMapper = CommonBeanFactory.getBean(ExtTaskMapper.class); + extTaskMapper.resetRunnings(taskId); + } public Long saveInstance(GlobalTaskInstance taskInstance) { EmailXpackService emailXpackService = SpringContextUtil.getBean(EmailXpackService.class); @@ -99,11 +121,12 @@ public class EmailTaskHandler extends TaskHandler implements Job { emailXpackService.saveInstance(taskInstance); } - + @Async("priorityExecutor") public void sendReport(GlobalTaskInstance taskInstance, XpackEmailTemplateDTO emailTemplateDTO, - SysUserEntity user) { + SysUserEntity user) { EmailXpackService emailXpackService = SpringContextUtil.getBean(EmailXpackService.class); try { + String panelId = emailTemplateDTO.getPanelId(); String url = panelUrl(panelId); String token = tokenByUser(user); @@ -116,11 +139,15 @@ public class EmailTaskHandler extends TaskHandler implements Job { String recipients = emailTemplateDTO.getRecipients(); byte[] content = emailTemplateDTO.getContent(); EmailService emailService = SpringContextUtil.getBean(EmailService.class); + String contentStr = ""; if (ObjectUtils.isNotEmpty(content)) { contentStr = new String(content, "UTF-8"); } - emailService.sendWithImage(recipients, emailTemplateDTO.getTitle(), contentStr, bytes); + emailService.sendWithImage(recipients, emailTemplateDTO.getTitle(), + contentStr, bytes); + + Thread.sleep(10000); success(taskInstance); } catch (Exception e) { error(taskInstance, e); diff --git a/backend/src/main/java/io/dataease/listener/GlobalTaskStartListener.java b/backend/src/main/java/io/dataease/listener/GlobalTaskStartListener.java index c7361ee93b..c8de41274a 100644 --- a/backend/src/main/java/io/dataease/listener/GlobalTaskStartListener.java +++ b/backend/src/main/java/io/dataease/listener/GlobalTaskStartListener.java @@ -32,6 +32,7 @@ public class GlobalTaskStartListener implements ApplicationListener { TaskHandler taskHandler = TaskStrategyFactory.getInvokeStrategy(task.getTaskType()); try { + taskHandler.resetRunningInstance(task.getTaskId()); taskHandler.addTask(scheduleManager, task); } catch (Exception e) { e.printStackTrace(); diff --git a/backend/src/main/java/io/dataease/plugins/server/XEmailTaskServer.java b/backend/src/main/java/io/dataease/plugins/server/XEmailTaskServer.java index 9ecd645704..fd8eb56a74 100644 --- a/backend/src/main/java/io/dataease/plugins/server/XEmailTaskServer.java +++ b/backend/src/main/java/io/dataease/plugins/server/XEmailTaskServer.java @@ -3,6 +3,7 @@ package io.dataease.plugins.server; import com.github.pagehelper.Page; import com.github.pagehelper.PageHelper; import io.dataease.commons.exception.DEException; +import io.dataease.commons.pool.PriorityThreadPoolExecutor; import io.dataease.commons.utils.*; import io.dataease.plugins.common.entity.GlobalTaskEntity; import io.dataease.plugins.common.entity.GlobalTaskInstance; @@ -23,6 +24,9 @@ import org.springframework.beans.factory.annotation.Autowired; import org.springframework.web.bind.annotation.*; import java.util.List; +import java.util.concurrent.Future; + +import javax.annotation.Resource; @Api(tags = "xpack：定时报告") @RequestMapping("/plugin/task") @@ -32,6 +36,9 @@ public class XEmailTaskServer { @Autowired private ScheduleService scheduleService; + @Resource + private PriorityThreadPoolExecutor priorityExecutor; + @PostMapping("/queryTasks/{goPage}/{pageSize}") public Pager> queryTask(@PathVariable int goPage, @PathVariable int pageSize, @RequestBody XpackGridRequest request) { @@ -85,7 +92,19 @@ public class XEmailTaskServer { String token = ServletUtils.getToken(); String fileId = null; try { - fileId = emailXpackService.print(url, token, buildPixel(request.getPixel())); + Future future = priorityExecutor.submit(() -> { + try { + return emailXpackService.print(url, token, buildPixel(request.getPixel())); + } catch (Exception e) { + LogUtil.error(e.getMessage(), e); + DEException.throwException("预览失败，请联系管理员"); + } + return null; + }, 0); + Object object = future.get(); + if (ObjectUtils.isNotEmpty(object)) { + fileId = object.toString(); + } } catch (Exception e) { LogUtil.error(e.getMessage(), e); DEException.throwException("预览失败，请联系管理员");