在尝试了解 Phaser 和 CyclicBarrier 之间的区别时，我遇到了一些链接 Difference between Phaser and CyclicBarrier和 https://www.infoq.com/news/2008/07/phasers/ 我读到 Phaser 与 Fork/Join 接口(interface)兼容，而 CyclicBarrier 不兼容，这里有一段代码可以证明这一点:

移相器

 public static void main(String[] args) throws InterruptedException {

        CountDownLatch countDownLatch = new CountDownLatch(1);

        Phaser phaser = new Phaser(16){
            @Override
            protected boolean onAdvance(int phase, int registeredParties) {
                return phase ==1 || super.onAdvance(phase, registeredParties);
            }
        };

        System.out.println("Available Processors: "+Runtime.getRuntime().availableProcessors());

        ExecutorService executorService = ForkJoinPool.commonPool(); // Runtime.getRuntime().availableProcessors() -1

        for (int i = 0; i < 16; i++) {
            final int count = 0;
            executorService.submit(() -> {
                while (!phaser.isTerminated()) {
                    try {
                        Thread.sleep(ThreadLocalRandom.current().nextInt(300, 2000));
                        System.out.println(Thread.currentThread().getName() + count + " ... ");
                        phaser.arriveAndAwaitAdvance();
                        System.out.println(Thread.currentThread().getName() + count + " ... continues ... ");
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                }
                countDownLatch.countDown();
            });
        }
        countDownLatch.await();
    }

循环障碍

public static void main(String[] args) throws InterruptedException {

        AtomicInteger phases = new AtomicInteger();
        CountDownLatch  countDownLatch = new CountDownLatch(1);
        CyclicBarrier cyclicBarrier = new CyclicBarrier(16, () -> phases.incrementAndGet());

        ExecutorService executorService = ForkJoinPool.commonPool();

        for (int i = 0; i < 16; i++) {
            executorService.submit(() -> {
                while (phases.get() < 1) {
                    try {
                        Thread.sleep(ThreadLocalRandom.current().nextInt(300, 2000));
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                    try {
                        System.out.println(Thread.currentThread().getName() + " Ok, I am waiting ");

                        cyclicBarrier.await();

                        System.out.println(Thread.currentThread().getName() + " continued it's way ... ");
                    } catch (BrokenBarrierException e) {
                        e.printStackTrace();
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                    countDownLatch.countDown();
                }
            });
        }
        countDownLatch.await();
    }

解释:

这两个代码运行一个 fork/join 线程池，这意味着线程是守护线程，这就是我使用 CountDownLatch 的原因。 commonPool() 方法将创建一个线程池，其中的线程等于 Runtime.getRuntime().availableProcessors()，我的是 12，因此它将创建 12 个线程。这两个示例中的 Phaser 和 CyclicBarrier 都定义了 16 个参与方，即它们需要在循环屏障中调用 16 次 await()，并在 Phaser 中调用 arriveAndAwaitAdvance()，以移动在。

在相位器的示例中，当第 12 个线程阻塞时，fork/join 将产生更多线程，它会创建更多线程，因此相位器最终将终止。然而，当第 12 个线程到达 await() 时，使用 CyclicBarrier 程序停止并且永远不会前进，它挂起。显然，因为屏障需要 16 次调用才能使线程前进，而创建的线程只有 12 次调用。线程池不会像处理 Phaser 那样创建更多线程来推进 CyclicBarrier。

问题:

fork/join 如何通过 Phaser 而不是通过 CyclicBarrier 设法创建更多线程？ 为什么方法 arriveAndAwaitAdvance() 使线程池创建新线程，以及如何，但是方法 await() 没有导致线程池创建更多线程？

最佳答案

Phaser 能够做到这一点是因为它在内部调用了 ForkJoinPool.managedBlock(ManagedBlocker)阻塞线程时。

ForkJoinPool 的这个 API 可供任何人访问，因此您可以轻松地增强您的 CyclicBarrier 版本以使用它，并消除线程饥饿。例如:

ForkJoinPool.managedBlock(new ManagedBlocker() {

    boolean isReleasable = false;

    @Override
    public boolean block() throws InterruptedException {
        try {
            cyclicBarrier.await();
        } catch (BrokenBarrierException aE) {
            throw new IllegalStateException(aE);
        }
        return isReleasable = true;
    }

    @Override
    public boolean isReleasable() {
        return isReleasable;
    }
});

关于java - Fork/Join 上下文中的 Phaser 与 CyclicBarrier，我们在Stack Overflow上找到一个类似的问题： https://stackoverflow.com/questions/57240612/