Java将作业安排在一定时间内最长时间运行

Question

我正在安排一项任务，每10秒运行一次。然而，这个任务应该有一个动态允许的执行时间。换句话说，如果允许的最大时间是5秒，并且任务运行超过5秒，则应该终止/关机。

我曾经尝试过使用@Schedule与cron时间，但无论我尝试什么，我都不能杀死一旦它运行。但是，有人建议我不要使用@Schedule，用ScheduledExecutorService创建一个正常的任务，但我不知道如何做。

我最初的方法是这样的：

    @Scheduled(cron = "0/10 * * * * ?")
    @Transactional
    public void testMethod(Integer period) {
        ThreadPoolTaskScheduler scheduler;

        scheduler.setAwaitTerminationSeconds(period);
        scheduler.setWaitForTasksToCompleteOnShutdown(false);

        importantMethod();
        
    }

我试图这样改写它：

public void testMethod(){
ScheduledExecutorService scheduler = Executors.newScheduledThreadPool(1);            
  scheduler.scheduleAtFixedRate(importantMethod(), delay,
                                  period, TimeUnit.SECONDS);
}

但是，我不知道如何将它设置为不延迟每10秒或每5分钟运行一次，并且只在超过允许的最大时间之后才关闭它。

任何帮助都将不胜感激。

Answer 1

我不确定这是否是最好的方法，但这是我想出的工作方法。

您需要三种类型的任务：

1. 任务本身，它必须定期运行--这是一个重要的任务。
2. 另一个将取消它的执行的任务，我将从现在起称之为杀死任务。
3. 任务来提交执行的重要任务--提交任务，这是ScheduledExecutorService定期运行的计划。

这个想法是，当您的重要任务开始执行时，它将安排杀死任务在5秒后运行。为了实现这一点，它需要知道它的Future实例。为此，它需要提交任务来完成提交并让它知道它的‘Future’。

这是important task

import java.time.LocalTime;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.Future;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;

public class ImportantTask implements Runnable {

    //that's just to make it easier to track what happens with each task
    private static long NEXT_ID = 1;

    //needed to schedule kill task
    private final ScheduledExecutorService executorService;
    private final long taskId;
    private Future<?> thisTaskFuture;
    //only needed to know when to cancel all tasks and end test
    private CountDownLatch countDownLatch;

    public ImportantTask(ScheduledExecutorService executorService) {
        this.executorService = executorService;
        this.taskId = NEXT_ID++;
    }

    @Override
    public void run() {
        //make half the tasks take too long
        long executionTimeSeconds = this.taskId % 2 == 0 ? 6 : 4;
        System.out.println(this.getLogMessage("started"));

        CancelFutureTask cancelFutureTask = new CancelFutureTask(this.thisTaskFuture);
        //this schedules the kill task for single execution, after 5 seconds delay
        this.executorService.schedule(cancelFutureTask, 5, TimeUnit.SECONDS);
        try {
            //simulating long execution
            Thread.sleep(executionTimeSeconds * 1_000);
        } catch (InterruptedException exc) {
            System.out.println(this.getLogMessage("interrupted"));
            //might need to check Thread.currentThread().isInterrupted(), depends on your exact case
            return;
        } finally {
            this.countDownLatch.countDown();
        }
        //the task is already finished, even if cancel is called, it has no effect
        System.out.println(this.getLogMessage("finished"));
    }

    private String getLogMessage(String taskStatus) {
        return String.format("Task with id - %d, %s at %s", this.taskId, taskStatus, LocalTime.now());
    }

    public void setThisTaskFuture(Future<?> thisTaskFuture) {
        this.thisTaskFuture = thisTaskFuture;
    }

    public void setCountDownLatch(CountDownLatch countDownLatch) {
        this.countDownLatch = countDownLatch;
    }
}

CountDownLatch的出现只是为了在执行几次之后更容易地停止main方法，您可能不需要它。注意代码中的注释。

kill task非常简单：

import java.util.concurrent.Future;

public class CancelFutureTask implements Runnable {

    private final Future<?> future;
    private final boolean mayInterrupt;

    public CancelFutureTask(Future<?> future, boolean mayInterrupt) {
        this.future = future;
        this.mayInterrupt = mayInterrupt;
    }

    public CancelFutureTask(Future<?> future) {
        this(future, true);
    }

    @Override
    public void run() {
        this.future.cancel(this.mayInterrupt);
    }
}

它只是取消了一个Future，如果未来已经完成了执行，它就不会产生任何效果。

和submit task

import java.util.concurrent.CountDownLatch;
import java.util.concurrent.Future;
import java.util.concurrent.ScheduledExecutorService;

public class SubmitNewImportantTask implements Runnable {

    private final ScheduledExecutorService executorService;
    private final CountDownLatch countDownLatch;

    public SubmitNewImportantTask(ScheduledExecutorService executorService, CountDownLatch countDownLatch) {
        this.executorService = executorService;
        this.countDownLatch = countDownLatch;
    }

    @Override
    public void run() {
        ImportantTask importantTask = new ImportantTask(this.executorService);
        //obtain Future instance
        Future<?> future = this.executorService.submit(importantTask);
        //let the task know of its Future instance
        importantTask.setThisTaskFuture(future);
        importantTask.setCountDownLatch(this.countDownLatch);
    }
}

它提交用于执行的important task，获取Future实例并在important task中设置它。

我用来测试的主要方法。

import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.ScheduledThreadPoolExecutor;
import java.util.concurrent.TimeUnit;

public class FuturesMain {

    public static void main(String[] args) throws Exception {
        CountDownLatch count = new CountDownLatch(6);
        ScheduledExecutorService executorService = new ScheduledThreadPoolExecutor(10);
        SubmitNewImportantTask submitTask = new SubmitNewImportantTask(executorService, count);
        executorService.scheduleAtFixedRate(submitTask, 0, 10, TimeUnit.SECONDS);
        count.await();
        executorService.shutdown();
        boolean allTerminated = executorService.awaitTermination(10, TimeUnit.SECONDS);
        System.out.println("All terminated - " + allTerminated);
    }
}

Answer 2

根据描述，我的假设是：

1. 任务应该每10秒运行一次。
2. 任务的最大超时时间应为5秒。
3. 最大超时应该是可配置的

application.properties中的配置

scheduler.timeout = 5000

代码

    @Value("${scheduler.timeout}")
    private Long timeout;
    
    @Scheduled(cron = "0/10 * * * * ?")
    public void taskScheduler() {
        System.out.println("Task Started: " + LocalDateTime.now());
        this.executeTask((Callable<Object>) () -> {
            TimeUnit.MILLISECONDS.sleep(3000);//Change values for testing: sleep time
            System.out.println("Task Worked: " + LocalDateTime.now());
            return true;
        }, timeout);
    }
    
    ExecutorService service = Executors.newFixedThreadPool(1);
    ScheduledExecutorService canceller = Executors.newSingleThreadScheduledExecutor();
    
    public <T> Future<T> executeTask(Callable<T> c, long timeoutMS) {
        final Future<T> future = service.submit(c);
        canceller.schedule(() -> {
            if (!future.isDone())
                System.out.println("Task Cancelled: " + LocalDateTime.now());
            future.cancel(true);
            return "done";
        }, timeoutMS, TimeUnit.MILLISECONDS);
        return future;
    }

输出

任务在3秒内完成时(睡眠时间)

Task Started: 2022-02-06T22:42:30.010704200
Task Worked: 2022-02-06T22:42:33.014775100

任务在6秒内完成时(睡眠时间)

Task Started: 2022-02-06T22:47:30.003249500
Task Cancelled: 2022-02-06T22:47:35.013195900