java - 根据 Callable 的某些属性选择 ThreadPool 线程

Question

我有一个可调用任务，我希望将其提交给执行者:

public static final class PersonalTask implements Callable<Object> {

  private final String name;
  private final int sleep;

  public PersonalTask(String name, int sleep) {
    this.name = name;
    this.sleep = sleep;
  }

  @Override
  public Object call() throws Exception {
    System.out.format("My name is %s and I'm sleeping for %d seconds%n", name, sleep);
    Thread.sleep(sleep * 1000);
    return null;
  }

}

每个任务都包含请求执行任务的人的姓名，以及一些 sleep 时间。此 sleep 持续时间是实际用例的代理，它调用了一些昂贵的操作。

为了完成这些任务，我使用了一个具有5 个线程 的固定线程池:

private static final ExecutorService executor = Executors.newFixedThreadPool(5);

为了说明我的问题，我想将以下内容提交给线程池:

public static void main(String[] args) throws Exception {
  List<PersonalTask> tasks = Arrays.asList(new PersonalTask[] { 
      new PersonalTask("Bob", 10), new PersonalTask("Bob", 10), 
      new PersonalTask("Bob", 10), new PersonalTask("Bob", 10),
      new PersonalTask("Bob", 10), new PersonalTask("Bob", 10),
      new PersonalTask("Eric", 1), new PersonalTask("Janice", 2) });

  executor.invokeAll(tasks);
}

输出结果是:

My name is Bob and I'm sleeping for 10 seconds
My name is Bob and I'm sleeping for 10 seconds
My name is Bob and I'm sleeping for 10 seconds
My name is Bob and I'm sleeping for 10 seconds
My name is Bob and I'm sleeping for 10 seconds

*** PAUSE FOR 10 SECONDS ***

My name is Bob and I'm sleeping for 10 seconds
My name is Eric and I'm sleeping for 1 seconds
My name is Janice and I'm sleeping for 2 seconds

这是因为要 hibernate 10 秒的 Bob 任务使可用的 5 个线程饱和，而其余任务(特别是属于 Eric 和 Janice 的任务)必须等待这些任务完成。

这不公平! Bob 提交的大量/缓慢的作业使可用线程饱和，并且 饿 Eric 和 Jane。

我想为 ExecutorService 提供一种机制来区分要求它安排的任务，这样我可能会想出一个更公平的解决方案。

我想暂时保持这个非常简单。 Bob 的所有任务都应该由池中的同一个线程处理。为了简单起见，我想采用 PersonalTask​​.name.hashCode() % threadPoolSize 并使用它来选择要使用的线程。

这意味着 Bob 只能使用可用的 5 个线程之一。这将使剩下的 4 个线程可以自由处理其他人的请求。

我意识到这并不完美，因为具有相同 hash % size 值的其他人仍然会落后于 Bob。事实上，他们现在不得不等待很长时间，很多，因为在该线程的队列中有 6 * 10 秒 作业在他们前面。

我可以在 Java 中使用哪些模式来完成此任务？

Answer 1

使用您自己的 BlockingQueue 实现(如 PriorityBlockingQueue )和手动构建的 ThreadPoolExecutor接受您的 BlockingQueue 作为参数。在您的 BlockingQueue 实现中，您可以保留一组所有现有人员(在任务中)并遍历它并返回该人员的下一个可用任务。