如何从ThreadPoolTaskScheduler获取执行结果

Question

在我的应用程序中，我从数据库检索一些元数据，并从中构建一个实现 Runnable 接口的 MyJob 类的实例，然后将其提交给 ThreadPoolTaskScheduler。这是执行此操作的代码。

@Scheduled(fixedRateString = "${check.rate.minutes}", 
           timeUnit = TimeUnit.MINUTES)
public void checkForJobs() {
    LOG.info("Executing check at {}", Instant.now().toString());
    int avail = threadPool.getPoolSize() - threadPool.getActiveCount();
    // will retrieve jobs metadata to fill vacancies
    List<MetaJob> availableJobs = this.repository.getAvailableJobs(avail);
    if(!(availableJobs == null || availableJobs.isEmpty())) {
        LOG.info("Retrieved {} jobs for scheduling", availableJobs.size());
        for (MetaJob job : availableJobs) {
            scheduleJob(job);
        }
    } else {
        LOG.info("No available jobs at this moment");
    }
}

private void scheduleJob(MetaJob job) {
    JobSchedule schedule = job.getSchedule();
    JobRunner jobRunner = getExecutor(job); // Creating Runnable out of metadata
    if(schedule == null) {
        threadPool.schedule(jobRunner, Instant.now());
    } else {
        Trigger trigger;
        if(schedule.getType() == SchedType.CRON) {
            trigger = new CronTrigger(schedule.getExpression());
        } else {
            trigger = new PeriodicTrigger(Duration.parse(schedule.getExpression()));
        }
        threadPool.schedule(jobRunner, trigger);
    }
}

我的工作正在按预期执行，但我需要处理其完成或失败的结果。 IE。如果执行次数超过元数据中指定的限制，我需要取消后续执行。我还需要使用存储在 MyJob 实例中的一些结果来更新数据库。如果工作失败了，我确实需要分析失败的原因。如果是由于临时中断，我会让作业再次运行，否则我也需要取消它。

所以，我看到ThreadPoolTaskScheduler有这个受保护的方法

afterExecute(Runnable task, @Nullable Throwable ex) 

并认为它可以解决我的问题。我编写了一个类 MyJobScheduler 扩展 ThreadPoolTaskScheduler 并使用以下代码重载该方法：

@Override
protected void afterExecute(Runnable task, Throwable ex) {
    super.afterExecute(task, ex); 
    if(ex == null) {
       MyJubRunner runner = (MyJobRunner)task;
       MyJob job = runner.getJob();
       // whatever I need to do with it
    } else {
      // analyze the exception
    }
}

我在这个函数中放置了一个断点，发现它在作业执行前后被调用了两次。其次，我得到了ClassCastException，因为这个函数作为参数接收的Runnable不是我提交的任务，而是java.util.concurrent.ScheduledThreadPoolExecutor$ScheduledFutureTask的实例。我尝试在将作业提交给调度程序时收到的 MyJob 和 ScheduledFuture 实例之间进行自己的映射，但发现 afterExecute 函数收到的此参数与上面提到的 ScheduledFuture 没有任何共同之处实例。

我的应用程序可以同时处理数十个作业，我确实需要找出哪一个已完成以及如何完成。

我的问题很清楚：是否可以从 afterExecute 方法中找出哪个任务已完成。如果可能的话-如何去做。如果没有，还有什么其他方法可以监控预定作业的完成情况？如果有人可以帮助我找到它，我将不胜感激。

Answer 1

您需要将从

ScheduledFuture

返回的所有

threadPool.schedule(jobRunner, Instant.now());

实例存储在数组或列表或其他内容中，并定期检查所有这些实例是否已完成，例如与另一个作业和一个单独的线程池。

类型

ScheduledFuture

包装您的

JobRunner

实例，并提供一些方法来检查作业的状态，例如取消或完成，请参阅 https://docs.oracle.com/javase/8/docs/api/java/util/concurrent/ScheduledFuture.html