java - 在两个类之间共享线程池

Question

我有一个固定大小12的线程池。现在我有两个实现Runnable的类，每个类有20个对象。我可以提交所有任务，线程池将照常完成其工作。

我想做的是做一个分隔符。因此，如果我提交这 40 个任务，线程池将不会同时处理每个类超过 6 个任务。所以线程池的行为就像2个大小为6的较小线程池。是否可以通过java或guava的API实现？

Answer 1

不用问“为什么”——它可以使用信号量来实现，每个信号量都以 6 的计数创建，每个信号量同时将每种类型的任务数量限制为 6 个。

这是一个基本的工作示例:

public class TaskDelimitingTest {

  private enum Tasks {TASK1, TASK2};

  private static ConcurrentHashMap<Tasks, AtomicInteger> taskObserver = new ConcurrentHashMap<>();

  public static class Task implements Runnable {

    private static final Random random = new Random(System.currentTimeMillis());

    private final Semaphore sem = new Semaphore(6, true);
    private final Tasks task;

    public Task(Tasks task) {
      this.task = task;
    }

    @Override
    public void run() {
      try {
        taskObserver.get(task).incrementAndGet();
        Thread.sleep(random.nextInt(1000));
        taskObserver.get(task).decrementAndGet();
        sem.release();
      } catch (InterruptedException e) {
        e.printStackTrace();
      }
    }

    public void postToExecution(ExecutorService executor) {
      try {
        sem.acquire();
        executor.execute(this);
      } catch (InterruptedException e) {
        e.printStackTrace();
      }
    }
  }

  public static class Task1 extends Task {

    public Task1() {
      super(Tasks.TASK1);
    }

  }

  public static class Task2 extends Task {

    public Task2() {
      super(Tasks.TASK2);
    }

  }

  public static void main(String[] args) {
    ExecutorService executor = Executors.newFixedThreadPool(12);

    Thread t1 = new Thread(() -> {
      taskObserver.put(Tasks.TASK1, new AtomicInteger());
      IntStream.rangeClosed(1, 100).forEach(i -> {
        new Task1().postToExecution(executor);
        System.out.println(taskObserver);
      });
    });
    Thread t2 = new Thread(() -> {
      taskObserver.put(Tasks.TASK2, new AtomicInteger());
      IntStream.rangeClosed(1, 100).forEach(i -> {
        new Task2().postToExecution(executor);
        System.out.println(taskObserver);
      });
    });
    t1.start();
    t2.start();
  }

}

在这个例子中，我在两个单独的线程中创建了两种类型各100个任务，以便它们可以相互竞争，我还在run方法中放置了Thread.sleep，以便它们模拟不同的执行时间。

该程序的输出是 - 在“预热”阶段

    {TASK2=1, TASK1=1}
    ...
    {TASK2=2, TASK1=3}
    ...
    {TASK2=4, TASK1=3}
    ...
    {TASK2=4, TASK1=4}
    ...
    {TASK2=4, TASK1=5}
    ...

有时池会饱和，然后就只会这样:

    {TASK2=6, TASK1=6}
    ...

因此每种类型最多只能同时执行 6 个线程。