代码之家 › 专栏 › 技术社区 › voo

使用ExecutorService将任务确定性分配给线程

executorservice concurrency multithreading java

voo · 技术社区 · 12 年前

给定具有固定线程池的Executor服务,是否可以保证将任务确定性地分配给线程?更准确地说,假设只有两个线程,即池-线程-0和池-线程-1,并且有两个任务要执行。我希望实现的是,前一个线程始终执行第一个线程,而后一个线程处理剩余的线程。

下面是一个示例:

public static void main(String[] args) throws InterruptedException, ExecutionException {
    ExecutorService executorService = newFixedThreadPool(2,
            new ThreadFactoryBuilder().setNameFormat("pool-thread-%d").build());

    for (int i = 0; i < 5; i++) {
        List<Callable<Integer>> callables = ImmutableList.of(createCallable(1), createCallable(2));
        executorService.invokeAll(callables);

    }
}

public static Callable<Integer> createCallable(final int task) {
    return new Callable<Integer>() {
        @Override
        public Integer call() throws Exception {
            currentThread().sleep(1000);
            System.out.println(Thread.currentThread().getName() + " executes task num: " + task);
            return task;
        }
    };
}

我的机器的示例输出:

pool-thread-0 executes task num: 1
pool-thread-1 executes task num: 2

pool-thread-0 executes task num: 2
pool-thread-1 executes task num: 1

pool-thread-0 executes task num: 2
pool-thread-1 executes task num: 1

pool-thread-0 executes task num: 2
pool-thread-1 executes task num: 1

pool-thread-0 executes task num: 1
pool-thread-1 executes task num: 2

简而言之,我希望确保pool-theread-0始终执行第一个任务。任何帮助都将不胜感激!

1 回复 | 直到 12 年前

GPI 12 年前

ExecutorService的设计目的不是为其Callable/Runable提供“线程相关性”。有人可能会说“这就是重点”,API是用来让程序员处理工作描述(Callable)的,而不是线程处理。

您的设计,即“每个线程都有一个随机数据生成器”,不适合ExecutorService,原因如下:

您无法控制将创建(或销毁)什么线程以及何时(如果一个线程崩溃了怎么办?池将重新创建它,但它将获得什么随机生成器?)。因此,我们无法推断出一种可靠的方式来表示“这个线程”具有“这个生成器”,更不用说“第二个”线程具有“这个生成程序”,因为可能甚至没有第二个线程(如果每个任务的处理速度都快到比您调度它们的速度还要快呢?)。
您无法控制何时执行哪些任务。好使用Executors.newFixedThreadPool,您可以按照提交的顺序进行调度,但据您所知,OS调度程序可能会将所有优先级赋予线程1,线程1将完成所有工作,而线程2将完全不做任何事情(两者之间的比例可以是任意的)。
将“数据生成器”传递给线程的唯一方法是重写执行器服务的ThreadFactory。否则,您将无法访问线程实例(在运行时从可调用对象自身显现)。因此,要将特定的生成器与特定的线程相关联,您必须知道当前正在创建的线程号,如果您正在计算线程数,这很容易,但如果您试图知道该线程的Callable是什么(请参见第2点),则很难。

因此,我强烈建议您定义一些其他方式来将工作单元与数据生成器关联起来,因为“线程实例”通常是不可靠的,至少不能通过Executor Service实现。例如,当你说

我需要提供它们处理的线程和数据的组合是可重复的。

我知道,您将总是发送一定数量的Callable,并且需要每个Callable处理特定生成器发出的特定数据集。假设我们有给定数量的任务和3个生成器,任务(N)将使用生成器 N%3 .

为了使结果可重复,您还需要使用同一生成器的任务不并发执行(您希望通过线程关联实现什么?)。

有一定数量的模式可以实现这一点。

1是:重构为生产者/消费者(反过来做)

在你的执行者服务中做3个任务,每个任务都听一个 BlockingQueue (其私有等待列表),并拥有自己的私有生成器。这些是消费者。
让你的主线程成为生产者:当它创建工作单元时(过去是什么 Callable 在您的原始设计中)编号N,将其分派到等待队列编号N%3。就是这样:每个消费将按您希望的顺序依次接收自己的数据进行计算。你已经达到了“亲和力”。

2是:自己完成任务分派任务。(用粗糙的方式)

首先,重构可调用项,使其具有到需要使用的生成器的链接。然后,在主线程上,构建每个生成器要运行的任务列表。
从主线程调度每个生成器的第一个任务。
在每个可调用的末尾,使可调用调度列表中的下一个工作单元。
但是,如果您从可调用项中分派可调用项,请不要等待结果,因为这将阻止可调用项完成,从而阻止新分派的执行。这是一个僵局。

3是:与2相同,效率较低,但风险较小

而不是从内部可调用程序中调度可调用程序,而是只从主线程中调度,等待未来。

通过这两种方式中的任何一种,都不能保证哪些任务将首先完成或最后完成,但可以保证您调度的工作单元与您控制的数据生成器可预测地关联,并且它们将按照您调度的顺序执行。希望这已经足够了。