将线程注册到 Phaser

Question

我正在学习 Phaser。在这样做的过程中，我遇到了一个问题。下面是我的代码，

public class RunnableTask implements Runnable {

    private Phaser phaser;

    public RunnableTask(Phaser phaser) {
        this.phaser = phaser;
        this.phaser.register();  // Question
    }

    @Override
    public void run() {
        // this.phaser.register();  // Question
        print("After register");
        for (int i = 0; i < 2; i++) {
            sleep();
            print("Before await" + i + ":");
            this.phaser.arriveAndAwaitAdvance();
            print("After advance" + i + ":");
        }
    }

    private void sleep() {
        try {
            TimeUnit.SECONDS.sleep(2);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }

    private void print(String msg) {

        System.out.println(String.format("%s: %s, time=%s, registered=%s, arrived=%s, unarrived=%s, phase=%s.", msg,
                Thread.currentThread().getName(), LocalTime.now(), this.phaser.getRegisteredParties(),
                this.phaser.getArrivedParties(), this.phaser.getUnarrivedParties(), this.phaser.getPhase()));
    }

 }

上述示例测试

public class TestPhaser {

    public static void main(String[] args) {

        Phaser phaser = new Phaser();

        RunnableTask task = new RunnableTask(phaser);

        Thread t1 = new Thread(task, "t1");
        Thread t2 = new Thread(task, "t2");
        Thread t3 = new Thread(task, "t3");

        t1.start();
        t2.start();
        t3.start();
    } 
}

执行上述程序后，输出为：

After register: t3, time=22:01:26.636, registered=1, arrived=0, unarrived=1, phase=0.

After register: t2, time=22:01:26.636, registered=1, arrived=0, unarrived=1, phase=0.

After register: t1, time=22:01:26.636, registered=1, arrived=0, unarrived=1, phase=0.

Before await 0:: t3, time=22:01:28.728, registered=1, arrived=0, unarrived=1, phase=0.

Before await 0:: t2, time=22:01:28.728, registered=1, arrived=0, unarrived=1, phase=0.

Before await 0:: t1, time=22:01:28.728, registered=1, arrived=0, unarrived=1, phase=0.

After advance 0:: t1, time=22:01:28.728, registered=1, arrived=0, unarrived=1, phase=3.

After advance 0:: t2, time=22:01:28.728, registered=1, arrived=0, unarrived=1, phase=3.

After advance 0:: t3, time=22:01:28.729, registered=1, arrived=0, unarrived=1, phase=3.

Before await 1:: t2, time=22:01:30.730, registered=1, arrived=0, unarrived=1, phase=3.

Before await 1:: t3, time=22:01:30.730, registered=1, arrived=0, unarrived=1, phase=3.

After advance 1:: t2, time=22:01:30.730, registered=1, arrived=0, unarrived=1, phase=4.

After advance 1:: t3, time=22:01:30.732, registered=1, arrived=0, unarrived=1, phase=5.

Before await 1:: t1, time=22:01:30.730, registered=1, arrived=0, unarrived=1, phase=3.

After advance 1:: t1, time=22:01:30.732, registered=1, arrived=0, unarrived=1, phase=6.

你可以看到这里有很多差异。线程未按顺序推进。此外，有几个阶段缺失 or/and 不在一个序列中。

当我将代码行 this.phaser.register() 从构造函数移动到运行方法的开头时，输出为：

After register: t1, time=22:10:58.230, registered=3, arrived=0, unarrived=3, phase=0.

After register: t3, time=22:10:58.230, registered=3, arrived=0, unarrived=3, phase=0.

After register: t2, time=22:10:58.230, registered=3, arrived=0, unarrived=3, phase=0.

Before await 0:: t2, time=22:11:00.314, registered=3, arrived=0, unarrived=3, phase=0.

Before await 0:: t1, time=22:11:00.314, registered=3, arrived=0, unarrived=3, phase=0.

Before await 0:: t3, time=22:11:00.314, registered=3, arrived=0, unarrived=3, phase=0.

After advance 0:: t2, time=22:11:00.315, registered=3, arrived=0, unarrived=3, phase=1.

After advance 0:: t3, time=22:11:00.315, registered=3, arrived=0, unarrived=3, phase=1.

After advance 0:: t1, time=22:11:00.315, registered=3, arrived=0, unarrived=3, phase=1.

Before await 1:: t1, time=22:11:02.319, registered=3, arrived=0, unarrived=3, phase=1.

Before await 1:: t2, time=22:11:02.319, registered=3, arrived=0, unarrived=3, phase=1.

Before await 1:: t3, time=22:11:02.319, registered=3, arrived=0, unarrived=3, phase=1.

After advance 1:: t3, time=22:11:02.320, registered=3, arrived=0, unarrived=3, phase=2.

After advance 1:: t2, time=22:11:02.320, registered=3, arrived=0, unarrived=3, phase=2.

After advance 1:: t1, time=22:11:02.321, registered=3, arrived=0, unarrived=3, phase=2.

线程执行和阶段按顺序排列看起来好多了。

这是我的问题：

1)为什么在Runnable的构造函数中注册party时会出现很多不一致的情况？

2) 在第二个结果中，到达和未到达的统计数据在每个阶段都为零（不正确）。那么，如何为他们获取正确的数字呢？

感谢任何帮助。

Answer 1

在第一个示例 "creation phaser in constructor" 中，您只向 Phaser 注册了一个线程。您必须创建三个任务，才能在 Phaser 中注册三个线程。

像这样更改代码，它将起作用。（不要忘记从您的代码中删除 initial RunnableTask task = new RunnableTask(phaser);）

    Thread t1 = new Thread(new RunnableTask(phaser), "t1");
    Thread t2 = new Thread(new RunnableTask(phaser), "t2");
    Thread t3 = new Thread(new RunnableTask(phaser), "t3");

在第二个例子中，你在所有线程中等待了 2 秒，这是准确的，所有线程几乎同时到达并等待，像这样改变你的睡眠方法来引入一些不同的线程等待查看一些到达和未到达的话题

  private void sleep() {
    try {
      Random r = new Random();    
      TimeUnit.SECONDS.sleep(r.nextInt(5));
    } catch(InterruptedException e) {
      e.printStackTrace();
    }
  }

您的第二个示例有效，但 correct.It 无效仅因为您在运行方法开始时处于休眠状态，因此所有线程在您调用 arrive 和 advance 方法之前赶上并在相位器中注册在移相器上。如果您要删除 sleep 然后在调用此行后

t1.start();

T1 运行方法将是运行并且 t1 线程将在移相器中注册。然后可能会在线程 t2 和 t3 启动并在相位器中注册之前调用 t1 运行方法中的 this.phaser.arriveAndAwaitAdvance()，因此相位器不会等着他们。

您应该在任务的构造函数中或在启动线程之前调用的方法中注册到相位器。

将线程注册到 Phaser

Registering thread to Phaser

java

multithreading

concurrency

java.util.concurrent

phaser