加速离散幂等任务

Question

我正在运行多个幂等任务来收集一批数据。我发现，很多时候，由于数百项任务中的几项任务，计算会显着延迟。

我想要的是一种观察这些任务的方式，并在它们明显延迟的情况下再次启动落后者。

在Java中是否有一个标准库或习惯用法来实现这一点？我目前正在使用ExecutorService/ExecutorCompletionService对来完成这项工作。

Answer 1

如果您有权访问表示此任务的Future对象，则可以根据需要检查isDone()和cancel()。您必须轮询这些未来的对象，并相应地重新提交。它还取决于适当处理InterruptExceptions的底层Runnables。

Answer 2

您可以创建一种类型的任务管理器，其中包含对每个任务的引用。此任务管理器可以负责启动每个任务并管理ExecutorService。每个任务的第一个也是最后一个操作是向管理器注册任务的开始和结束。然后，经理可以建立一个统计图，它是执行每项任务所用时间的平均值。

任务管理器周期性地扫描其正在运行的任务列表，寻找仍在运行并且与特定任务所用的平均时间有显着漂移的“异常值”。然后，它可以取消这些任务并重新启动它们。

下面是你可以做什么的一个非常粗略的轮廓。

public class Task implements Runnable {
     protected TaskManager manager_ = null;
     protected String taskClass_ = null;
     protected String taskId_ = null;

     protected Task(TaskManager manager, String taskClass) {
        manager_ = manager;
        taskClass_ = taskClass;
     }

     /*
      * Override this and perform specific task.
      */
     protected void perform() { }

     public void run() {
      try {
          manager_.taskStarted(this);
          perform();
          manager_.taskCompleted(this);
      catch(InterruptedException) {
          manager_.taskAborted(this);
      }
      finally {
      }
    }
}


public class TaskManager {
    ExecutorService service_ = null;

    public TaskManager() {
       service_ = new ExecutorService();
       // start the monitoring thread.
       service_.execute(this);
    }

    public void runTask(Task t) {
       service_.execute(t);
    }

    public void taskStarted(Task t) {

        1. Note the time that this task (with unique id) has started.
        2. Add time to a hash map.
        3. Add task to list of executing tasks.
    }

    public void taskComplete(Task t) {
        1. Find the task id in hash map
        2. note how long it took to execute.
        3. modify statistics of how long the task took against
           the task Class Id.
        4. Remove task from list of executing tasks.
    }

    public void taskAborted(Task t) {
      // just remove the task from list of running tasks 
      // without altering the statistics.
    }
    public void run() {
         1. Go though the list of executing tasks looking for 
            tasks whose current time  - start time is outside the
            time statistics for the task class id.
         2. cancel the task and start again.
    }
}