从 ScheduledExecutorService 中运行的任务本身停止周期性任务

Question

在 ScheduledExecutorService 中运行时，是否有一种好方法可以从任务本身内部停止重复任务？

可以说，我有以下任务：

Future<?> f = scheduledExecutor.scheduleAtFixedRate(new Runnable() {
    int count = 0;
    public void run() {
       System.out.println(count++);
       if (count == 10) {
           // ??? cancel self
       }
    }
}, 1, 1, TimeUnit.SECONDS);

从外部，很容易通过 f.cancel() 取消，但是如何在指定位置停止重复？ （通过 AtomicReference 传递 Future 并不安全，因为当 ScheduleAtFixedRate 返回 f 较晚并且变量也设置较晚时，存在一个潜在的窗口，并且任务本身可能已经运行，在引用中看到 null。）

Answer 1

当重复任务抛出异常或错误时，它会被放置在 Future 中，并且该任务不会再次重复。您可以选择抛出 RuntimeException 或 Error。

Answer 2

您可以使用命名类，而不是使用匿名内部类，该类可以具有您在计划任务时从 Executor 获取的 Future 对象的属性。

abstract class FutureRunnable implements Runnable {

    private Future<?> future;

    /* Getter and Setter for future */

}

当您安排任务时，您可以将 Future 传递给 Runnable。

FutureRunnable runnable = new FutureRunnable() {

    public void run() {
        if (/* abort condition */)
            getFuture().cancel(false);
    }

};
Future<?> future = executor.scheduleAtFixedRate(runnable, ...);
runnable.setFuture(future);

也许您必须确保在设置 Future 之前不会执行该任务，否则您将收到 NullPointerException。

Answer 3

对于 Runnable 来说，了解有关其运行的执行器的任何信息，或者如果达到 10 不是错误状态则抛出错误，这似乎是一个糟糕的设计。

你能在调度和执行之外循环到10吗？这可能需要使用非调度执行器，因为您需要自己手动调度它们。

Answer 4

这是另一种方式，这甚至是线程安全的；

    final Future<?>[] f = {null};
    f[0]=  scheduledExecutor.scheduleAtFixedRate(new Runnable() {
        int count = 0;
        public void run() {

            System.out.println(count++);
            if (count == 10) {
                Future<?> future;
                while(null==(future = f[0])) Thread.yield();//prevent exceptionally bad thread scheduling 
                future.cancel(false);
                return;
                //cancel self
            }
        }
    }, 1, 1, TimeUnit.SECONDS);

Answer 5

现在才看到这个...因为我想做同样的事情...这是我的解决方案，我怀疑这是线程安全的。

首先为 Future 创建一个容器：

public static class Cancel {
    private ScheduledFuture<?> future;

    public synchronized void setFuture(ScheduledFuture<?> future) {
        this.future = future;
    }

    public synchronized void stop() {
        LOG.debug("cancelling {}", future);
        future.cancel(false);
    }
}

然后是 future 代码：

    final Cancel controller = new Cancel();

    synchronized (controller) {
        ScheduledFuture<?> future = scheduler.scheduleWithFixedDelay(() -> {
            if (<CONTINUE RUNNING CONDITION) {

            } else {
                // STOP SCHEDULABLE FUTURE
                controller.stop();
            }
        }, startTime, timeBetweenVisbilityChecks);
        controller.setFuture(future);
    }
}

所以请注意，在创建 future 并在控制器上设置 future 之前，将无法调用 stop。

请记住，Runnable 是匿名内部类，它将完全在不同的线程中运行。