任务中的ScheduledExecutorService取消任务

Question

我有一个定时任务，它将进行一个http调用，然后检查返回结果。如果结果是一定的值，我会做一些事情，然后这个定时的任务可以终止，不再需要调度。我如何在我的任务中取消这个时间表？？ScheduledExecutorService是否提供类似于终止条件的参数？

        ScheduledExecutorService service = new ScheduledThreadPoolExecutor(5);

        service.scheduleAtFixedRate(() -> {
            // http request
            // check
            // result is xxx, no more schedule
            // How can I stop this task schedule within my task? ?
            
        }, 1, 60, TimeUnit.SECONDS);

Answer 1

可能起作用的最简单的事情是，当您想要停止时，从调度函数的主体中抛出一个异常。我觉得很丑。见这里的javadoc。

另一种方法是不定期调度(即不使用scheduleAtFixedRate)，而是只调度一次，并在检查调用结果后重新安排(或不重新安排)。

如果您想要近似地了解scheduleAtFixedRate所做的事情，则必须跟踪调用所用的时间。

当调用超过60秒时，此解决方案近似于scheduleAtFixedRate，并退化为scheduleAtFixedDelay：

private static final ScheduledExecutorService scheduler = Executors.newSingleThreadScheduledExecutor();

public static void main(String[] args) {
    scheduler.schedule(() -> {
        performHttpCallAndScheduleAgainIfNeeded(scheduler);
    }, 1, TimeUnit.SECONDS);
}

private static void performHttpCallAndScheduleAgainIfNeeded(ScheduledExecutorService scheduler) {
    final long startTime = System.currentTimeMillis();
    boolean callWasOk = performHttpCall();
    if (!callWasOk) {
        final long elapsed = System.currentTimeMillis() - startTime;
        final long millisToWaitForNextAttempt = Math.max(0, 60000 - elapsed);
        scheduler.schedule(() -> performHttpCallAndScheduleAgainIfNeeded(scheduler), millisToWaitForNextAttempt, TimeUnit.MILLISECONDS);
    } else {
        // the call was ok... you have nothing to do.
    }
}

private static boolean performHttpCall() {
    // do whatever you need to do here...
}

如果您仍然希望使用scheduleAtFixedRate或scheduleWithFixedDelay，则必须保持某种状态(注意并发性)，并注意这两种方法都返回了取消它们所必须使用的未来。

这里的诀窍是，您必须将scheduleAtFixRate调用的结果传递给实际的调度函数。您可以通过使用容器而不是值来解决这个鸡蛋问题:这里我使用CompletableFuture作为容器：

public static void main(String[] args) {
    final CompletableFuture<ScheduledFuture<?>> cancellablePeriodicTask = new CompletableFuture<>();
    final ScheduledFuture<?> cancellable = scheduler.scheduleAtFixedRate(() -> {
        performHttpCallAndScheduleAgainIfNeeded(cancellablePeriodicTask);
    }, 1, 60, TimeUnit.SECONDS);
    cancellablePeriodicTask.complete(cancellable);
}

private static void performHttpCallAndScheduleAgainIfNeeded(CompletableFuture<ScheduledFuture<?>> cancellable) {
    boolean callWasOk = performHttpCall();
    if (callWasOk) {
        // here you cancel
        cancellable.whenComplete((scheduledFuture, throwable) -> {
          if (throwable == null) {  
              scheduledFuture.cancel(true);
          }
        });
    }

    // call was not ok here, nothing to do since the scheduler will call this again.
}

private static boolean performHttpCall() {
    // do whatever you need to do here...
}

在这种方法中，CompletableFuture作为在线程之间传递值的机制并不是唯一起作用的东西(SynchronousQueue也在这里工作)。

Answer 2

一对遗嘱执行人服务

一种方法是使用一对执行者服务。一个是调度服务，重复运行您的任务，直到满足某些条件。当满足时，该任务的最后一次执行将向其他executor服务提交一个取消任务。然后，另一个第二个执行器服务将执行计划执行程序服务的关闭。

步骤

提供非预定的执行者服务。

提供预定的执行者服务。

将重复任务实例化为Runnable或Callable。向其构造函数传递对两个executor服务的引用。

在计划好的执行者服务上安排您的任务。

每次任务运行时，检查退出情况。

• 当这个条件是错误的，不要做更多的事。让run/call方法完成。
• 当该条件为真时，向非计划执行器服务提交一个新任务。该新任务将对计划执行程序服务的引用作为其构造函数的参数。该任务的run/call方法取消已传递的计划执行器服务。

要执行取消操作，任务调用ScheduledExecutorService#shutdown和#awaitTermination。

示例代码

package work.basil.tasking;

import java.time.Duration;
import java.time.Instant;
import java.util.Objects;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;

public class App
{
    public static void main ( String[] args )
    {
        App app = new App();
        app.demo();
    }

    private void demo ( )
    {
        ScheduledExecutorService coreScheduledExecutorService = Executors.newSingleThreadScheduledExecutor();
        ExecutorService cancellationExecutorService = Executors.newSingleThreadExecutor();
        Duration expires = Duration.ofMinutes( 2 );
        Runnable coreTask = new CoreTask( expires , coreScheduledExecutorService , cancellationExecutorService );
        coreScheduledExecutorService.scheduleAtFixedRate( Objects.requireNonNull( coreTask ) , 0 , 20 , TimeUnit.SECONDS );

        try { Thread.sleep( expires.plus( Duration.ofMinutes( 1 ) ).toMillis() ); } catch ( InterruptedException e ) { e.printStackTrace(); }
        if ( Objects.nonNull( coreScheduledExecutorService ) )
        {
            if ( ! coreScheduledExecutorService.isShutdown() )
            {
                coreScheduledExecutorService.shutdown();
                try { coreScheduledExecutorService.awaitTermination( 1 , TimeUnit.MINUTES ); } catch ( InterruptedException e ) { e.printStackTrace(); }
            }
        }
        if ( Objects.nonNull( cancellationExecutorService ) )
        {
            if ( ! cancellationExecutorService.isShutdown() )
            {
                cancellationExecutorService.shutdown();
                try { cancellationExecutorService.awaitTermination( 1 , TimeUnit.MINUTES ); } catch ( InterruptedException e ) { e.printStackTrace(); }
            }
        }
    }

    class CoreTask implements Runnable
    {
        private ScheduledExecutorService scheduledExecutorServiceRunningThisTask;
        private ExecutorService cancellationExecutorService;
        private Duration exiration;
        Instant whenCreated;

        public CoreTask ( final Duration expiration , final ScheduledExecutorService scheduledExecutorServiceRunningThisTask , final ExecutorService cancellationExecutorService )
        {
            this.exiration = Objects.requireNonNull( expiration );
            this.scheduledExecutorServiceRunningThisTask = Objects.requireNonNull( scheduledExecutorServiceRunningThisTask );
            this.cancellationExecutorService = Objects.requireNonNull( cancellationExecutorService );
            this.whenCreated = Instant.now();
        }

        @Override
        public void run ( )
        {
            Duration elapsed = Duration.between( this.whenCreated , Instant.now() );
            System.out.print( "Core task running. " + Instant.now() + " | Elapsed: " + elapsed + " | " );
            if ( elapsed.toSeconds() > this.exiration.toSeconds() )
            {
                System.out.println( "Core task is asking for cancellation. " + Instant.now() );
                this.cancellationExecutorService.submit( ( ) -> this.scheduledExecutorServiceRunningThisTask.shutdown() );
            } else
            {
                System.out.println( "Core task is completing another `run` execution. " + Instant.now() );
            }
        }
    }
}

跑的时候。

Core task running. 2021-12-05T04:20:41.659240Z | Elapsed: PT0.000857S | Core task is completing another `run` execution. 2021-12-05T04:20:41.672656Z
Core task running. 2021-12-05T04:21:01.663990Z | Elapsed: PT20.00593S | Core task is completing another `run` execution. 2021-12-05T04:21:01.664514Z
Core task running. 2021-12-05T04:21:21.659970Z | Elapsed: PT40.001914S | Core task is completing another `run` execution. 2021-12-05T04:21:21.660327Z
Core task running. 2021-12-05T04:21:41.663228Z | Elapsed: PT1M0.005188S | Core task is completing another `run` execution. 2021-12-05T04:21:41.663420Z
Core task running. 2021-12-05T04:22:01.662737Z | Elapsed: PT1M20.004684S | Core task is completing another `run` execution. 2021-12-05T04:22:01.663140Z
Core task running. 2021-12-05T04:22:21.663495Z | Elapsed: PT1M40.005431S | Core task is completing another `run` execution. 2021-12-05T04:22:21.664237Z
Core task running. 2021-12-05T04:22:41.663013Z | Elapsed: PT2M0.004967S | Core task is completing another `run` execution. 2021-12-05T04:22:41.663248Z
Core task running. 2021-12-05T04:23:01.662875Z | Elapsed: PT2M20.004835S | Core task is asking for cancellation. 2021-12-05T04:23:01.663117Z

顺便说一句，请记住，来自System.out的控制台输出不一定按时间顺序显示。当您关心序列时，研究Instant.now()值以验证执行顺序。