有没有办法将 ScheduledExecutorService 与 ExecutorCompletionService 一起使用？

Question

我正在尝试同时使用 ExecutorCompletionService 和 ScheduledExecutorService。

我需要做的是安排不同的活动，每个活动都有“执行前的延迟”，然后根据上次运行的结果“重新安排它们”（不同的延迟）。”

我遇到的问题是我不能使用“延迟”的 ExcecutorCompletionService 提交

我尝试了以下方法，但它永远阻塞...

显然我遗漏了 Java 语言中的一个基本问题。

有没有办法将任务安排到 ScheduledExecutorService 以便 CompletionService “知道”？

public class Bar {

    private ScheduledExecutorService scheduledExecutor;
    private Future<Status> action1Future;
    private Future<Status> action2Future;
    private ExecutorCompletionService<Status> pool;
    private long delay1 = 10;
    private long delay2 = 20;
    private long delay3 = 30;

    public void start() {
        scheduledExecutor = Executors.newScheduledThreadPool(3);

        Action1 a1 = new ActionOne(); // Action1 implements Callable<Status>
        Action2 a2 = new ActionTwo(); // Action2 implements Callable<Status>

        pool = new ExecutorCompletionService<Status>(scheduledExecutor);
        action1Future = scheduledExecutor.schedule(a1, delay1, TimeUnit.SECONDS);
        action2Future = scheduledExecutor.schedule(a2, delay1, TimeUnit.SECONDS);
        monitorAndRestart();
    }

    private void monitorAndRestart() {
         boolean isDone=false;
        do {
        try {
            // THIS IS WHERE IT BLOCKS.
            Future<Status> processedItem = pool.get();
            if (processedItem == action1Future) {
                if (processedItem.get() == Status.GOOD) {
                    action1Future = scheduledExecutor.schedule(new ActionOne(), delay1, TimeUnit.SECONDS);
                } else {
                    action1Future = scheduledExecutor.schedule(new ActionOne(), delay2, TimeUnit.SECONDS);
                }
            } else if (processedItem == action2Future) {
                if (processedItem.get() == Status.GOOD) {
                    action1Future = scheduledExecutor.schedule(new ActionOne(), delay2, TimeUnit.SECONDS);
                } else {
                    action1Future = scheduledExecutor.schedule(new ActionOne(), delay3, TimeUnit.SECONDS);
                }
            }
        } catch (InterruptedException e) {
            isDone = true;
            // handle this.. shudown whatever
        }
        catch (ExecutionException e) {
            // handle this
        }
      } while (isDone == false);
    }

    public static void main(String[] args) {
        Bar myRunner = new Bar();
        myRunner.start();
    }
}

如果我把“延迟在可调用”中创建通过 new ActionOne(delay);并使用 CompletionService.submit(..) 它可以工作。

actionFuture1 = pool.submit(new ActionOne(delay1));
/////


public class ActionOne implements Callable<Status>(
   private final delay;
   public ActionOne(long dl) {
       delay=dl;
   }
   Status call() {
       try {
         Thread.sleep(delay * 1000); // seconds
          return doSomething()
       } catch (...) { //thread.sleep execptions}
   }
 }

所以我猜最后一个问题是：ScheduledExecutorService 有什么比 Thread.sleep(delay) 方式更好的方法吗？

Answer 1

我们在我们的一个应用程序中进行了这种重新安排，我们决定像 Ed Thomas 建议的那样包装任务。 （我们还通过传递一个返回任务下一个执行时间的迭代器使重新调度变得超级灵活——允许我们使用许多不同的调度策略）。

另一个选择是继承 ScheduledThreadPoolExecutor 并覆盖 afterExecute。这最终可能比包装任务更干净。

Answer 2

我认为 CompletionService 不会起作用，尤其是当您尝试重新安排它时。你可以使用 Guava 库并切换到 ListenableFutures 吗？这样自己构建起来会更容易一些。

对于不同的方法，请考虑让期货/任务自行重新安排。您甚至可以从原始任务的包装器中处理此问题（从您的代码中很难判断最终目标是什么）。

很难确切地说出你需要什么模式，但也许你可以玩这样的东西：

public static void scheduleRepeatedly(final Callable<Status> callable, final int someDelay, final ScheduledExecutorService executor) {
    executor.schedule(new Runnable() {
        @Override
        public void run() {
            Status status = Status.BAD;
            try {
                status = callable.call();
            } catch (Exception e) {
                // whatever logic here to determine if callable should be rescheduled and what the new delay is
                if (status == Status.BAD) {
                    scheduleRepeatedly(callable, someDelay * 2, executor);
                }
            }
        }
    }, someDelay, TimeUnit.SECONDS);
}