0

我有一个处理较大作业的 JobService。作业被动态细分为多个任务，任务也可能产生子任务等，因此无法预测一个作业的任务总数。每个任务自己排队运行ExecutorService.submit(...) 问题是我似乎必须为每个作业创建一个单独的 ExecutorService ，因为判断“作业队列”何时完成的唯一方法是使用ExecutorService.awaitTermination(...). 不过，这似乎效率低下，因为我无法在作业和它们的 ExecutorService 之间共享单个线程池。

我正在寻找一些替代方案，我正在考虑为每项工作使用 AtomicInteger。提交新任务时递增，任务完成时递减。但是我必须轮询它什么时候为零，这看起来很混乱，还有一些异常处理混乱。

似乎必须有更好的解决方案？

4

2 回答 2

4

Submit 返回一个 Future 对象，该对象可用于等待任务完成。您可以跟踪这些并添加一个递归阻塞直到所有子任务完成的方法。这样，您可以在任何需要的地方重用执行器。

import java.util.ArrayList;
import java.util.Collection;
import java.util.List;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;
import java.util.concurrent.atomic.AtomicBoolean;

public class JobExecutor {
    ExecutorService executorService = Executors.newFixedThreadPool(1);

    private class Task implements Runnable {
        private final String name;
        private final Task[] subtasks;
        private final ExecutorService executorService;
        private volatile boolean started = false;
        private Future<?> taskFuture;

        // Separate list from subtasks because this is what you'll probably
        // actually use as you may not be passing subtasks as constructor args
        private final List<Task> subtasksToWaitOn = new ArrayList<Task>();

        public Task(String name, ExecutorService executorService,
                Task... subtasks) {
            this.name = name;
            this.executorService = executorService;
            this.subtasks = subtasks;
        }

        public synchronized void start() {
            if (!started) {
                started = true;
                taskFuture = executorService.submit(this);
            }
        }

        public synchronized void blockTillDone() {
            if (started) {
                try {
                    taskFuture.get();
                } catch (InterruptedException e) {
                    // TODO Handle
                } catch (ExecutionException e) {
                    // TODO Handle
                }
                for (Task subtaskToWaitOn : subtasksToWaitOn) {
                    subtaskToWaitOn.blockTillDone();
                }
            } else {
                // TODO throw exception
            }
        }

        @Override
        public void run() {
            for (Task subtask : subtasks) {
                subtask.start();
                subtasksToWaitOn.add(subtask);
            }
            System.out.println("My name is: " + name);
        }
    }

    void testSubmit() {
        Task subsubTask1 = new Task("Subsubtask1", executorService);
        Task subtask1 = new Task("Subtask1", executorService, subsubTask1);
        Task subtask2 = new Task("Subtask2", executorService);
        Task subtask3 = new Task("Subtask3", executorService);
        Task job = new Task("Job", executorService, subtask1, subtask2,
                subtask3);
        job.start();
        job.blockTillDone();
        System.out.println("Job done!");
    }

    public static void main(String[] args) {
        new JobExecutor().testSubmit();
    }
}

打印出来：

My name is: Job
My name is: Subtask1
My name is: Subtask2
My name is: Subtask3
My name is: Subsubtask1
Job done!
于 2012-08-09T03:02:00.053 回答
1

如果您使用的是 java7（或带有 backport 库http://www.cs.washington.edu/homes/djg/teachingMaterials/grossmanSPAC_forkJoinFramework.html的 java6 ），您可能需要考虑使用 Fork-Join 池进行此类事物：

class MainTask extends RecursiveTask<Long> {

    @Override
    protected Long compute() {
        SubTask subtask0 = new SubTask(0L);
        SubTask subtask1 = new SubTask(1L);
        SubTask subtask2 = new SubTask(2L);
        SubTask subtask3 = new SubTask(3L);
        SubTask subtask4 = new SubTask(4L);
        SubTask subtask5 = new SubTask(5L);

        subtask1.fork();
        subtask2.fork();
        subtask3.fork();
        subtask4.fork();
        subtask5.fork();

        return subtask0.compute() +
                subtask1.join() +
                subtask2.join() +
                subtask3.join() +
                subtask4.join() +
                subtask5.join();
    }

}

class SubTask extends RecursiveTask<Long> {
    private Long rawResult = null;

    private Long expected = null;

    public SubTask(long expected) {
        this.expected = expected;
    }

    @Override
    protected Long compute() {
        return expected;
    }
}

public static void main( String[] args )
{
    ForkJoinPool forkJoinPool = new ForkJoinPool();
    Long result = forkJoinPool.invoke(new MainTask());
    System.out.println(result);
}

显然，这具有硬编码的子任务，但是没有理由不能将参数传递给主任务，并使用它来生成子任务。子任务本身不必都是同一类型，但它们都应该扩展 RecursiveTask。实际上，如果一个任务生成子任务（如上面的 MainTask），至少其中一个子任务应该直接在其上调用“计算”​​（而不是分叉和连接），以便当前线程可以执行一些计算，并让其他线程完成其余的工作。

于 2012-08-09T04:17:57.890 回答