Phaser - 如何将其用作 CountDownLatch(1)？

Question

我知道我可以直接使用CountDownLatch，但是，作为练习并更好地理解Phaser，我想使用它而不是COuntDownLatch。

因此，我将创建 N 个等待者，以及一个需要翻转闩锁的线程。所有等待者，如果在翻转之前到达，都会阻塞，但是在锁存器倒计时之后，所有后续的await() 都会立即返回。

与Phaser 我不知道如何实现这一点...屏障很容易，因为我们有 N + 1 个线程，每个线程都到达并等待。然而，为了确保没有线程会在第一阶段之后等待，不知何故让我望而却步。

我能想出的唯一方法，不是很好，如下：

Phaser p = new Phaser(1);
int phase = p.getPhase();
....
// thread that awaits()
new Thread(new Runnable() {
    ....
    p.awaitAdvance(phase)
}

而另一个线程只是将移相器推进到下一个阶段。这并不理想，因此任何指针都将不胜感激。

Answer 1

TL;DR：在这种情况下，使用Phaser.arriveAndDeregister() 向移相器的等待者发出非阻塞信号，这对应于操作CountDownLatch.countDown()。

Phaser和CountDownLatch。首先要澄清的是Phaser不能通常编码CountDownLatch。与Phaser 同步的任务都必须相互等待（all-to-all 同步）。在CountDownLatch 中，有一组任务等待其他任务打开闩锁。

Phaser 和CyclicBarrier。 这两种机制都用于全对全同步。它们之间的两个区别是：1）Phaser 使用移相器的任务集可能会在移相器的生命周期中增长，而在CyclicBarrier 中，参与者的数量是固定； 2) 使用Phasers，一个任务可以通知其他成员（参与者），只要它从该移相器中注销就不会等待，而所有使用循环屏障的任务只能等待并通知。

使用 Phaser 编码 CountDownLatch。 要使用移相器对 CountDownLatch(1) 进行编码，您需要记住以下几点：

1. 聚会人数 = 服务员人数 + 1：通过new Phaser(PARTIES_COUNT) 或Phaser.register 注册的聚会人数。
2. CountDown.await() = Phaser.arriveAndAwaitAdvance()
3. CountDown.countDown() = Phaser.arriveAndDeregister()

示例。假设您希望子任务等待父任务的信号。使用CountDownLatch 你会写：

import java.util.concurrent.*;

class CountDownExample {
    public static void main(String[] args) throws Exception {
        CountDownLatch l = new CountDownLatch(1);
        new Thread(() -> {
            try {
                l.await();
                System.out.println("Child: running");
            } catch (InterruptedException e) {
                throw new RuntimeException(e);
            }
        }).start();
        System.out.println("Parent: waiting a bit.");
        Thread.sleep(100);
        l.countDown();
    }
}

使用Phaser 你会写：

import java.util.concurrent.*;

class PhaserExample {
    public static void main(String[] args) throws Exception {
        Phaser ph = new Phaser(2); // 2 parties = 1 signaler and 1 waiter
        new Thread(() -> {
            ph.arriveAndAwaitAdvance();
            System.out.println("Child: running");
        }).start();
        System.out.println("Parent: waiting a bit.");
        Thread.sleep(100);
        ph.arriveAndDeregister();
    }
}

你可能想看看this post for another example。

Answer 2

我不同意@tiago-cogumbreiro 的回答。其实Phaser可以用来模拟CountDownLatch的行为。

您可以使用 Phaser 的arrive() 方法来实现它。它不会等待其他线程到达。因此，可以实现CountDownLatch 行为。

Phaser 的arriveAndAwaitAdvance() 使调用它的线程等待，直到其他线程/方到达。因此，该方法可以用来模拟CyclicBarrier的行为。

以下是使用Phaser 模拟CountDownLatch(3) 行为的源代码。如果只使用一个WorkerThread，可以模拟CountDownLatch(1)

如果将arrive() 方法替换为arriveAndAwaitAdvance() 方法，下面的代码将模拟CyclicBarrier。

public class PhaserAsCountDownLatch {

    static Phaser phaser = new Phaser(3);

    public static void main(String[] args) {
     
        new WorkerThread2().start();
        new WorkerThread3().start();
        new WorkerThread().start();
        
        //waits till phase 0 is completed and phase is advanced to next.
        phaser.awaitAdvance(0); 
        
        System.out.println("\nFROM MAIN THREAD: PHASE 0 COMPLETED");
        System.out.println("FROM MAIN THREAD: PHASE ADVANCED TO 1");
        System.out.println("MAIN THREAD ENDS HERE\n");
    }

    static class WorkerThread extends Thread{
        @Override
        public void run() {
            for (int i = 1; i < 5; i++) {
                System.out.println("tid: " + Thread.currentThread().getId() +  ", BEFORE ARRIVING val is: " + i);
            }
        
            phaser.arrive();
        
            System.out.println("ARRIVED tid: " + Thread.currentThread().getId());
        }
    }

    static class WorkerThread2 extends Thread{
        @Override
        public void run() {
        
            //won't wait for other threads to arrive. Hence, CountDownLatch behaviour can be achieved
            phaser.arrive();
        
            System.out.println("ARRIVED tid: " + Thread.currentThread().getId());
            for (int i = 200; i < 231; i++) {
                System.out.println("tid: " + Thread.currentThread().getId() +  " AFTER ARRIVING. val is: " + i);
            }
        }
    }

    static class WorkerThread3 extends Thread{
        @Override
        public void run() {
        
            //won't wait for other threads to arrive. Hence, CountDownLatch behaviour can be achieved
            phaser.arrive();
        
            System.out.println("ARRIVED tid: " + Thread.currentThread().getId());
            for (int i = 300; i < 331; i++) {
                System.out.println("tid: " + Thread.currentThread().getId() +  " AFTER ARRIVING. val is: " + i);
            }
        }
    }
}