Регистрация потока в Phaser

Question

Я узнаю о Фазере. При этом я столкнулся с проблемой. Ниже приведен код, который у меня есть,

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();
    } 
}

При выполнении вышеупомянутой программы вывод:

После регистра: t3, время =22:01:26.636, зарегистрировано = 1, прибыло = 0, не получено = 1, фаза = 0.

После регистра: t2, время =22:01:26.636, зарегистрировано = 1, прибыло = 0, не получено = 1, фаза = 0.

После регистра: t1, время =22:01:26.636, зарегистрировано = 1, прибыло = 0, не получено = 1, фаза = 0.

Перед ожиданием 0:: t3, время =22:01:28.728, зарегистрировано = 1, прибыло = 0, необработано = 1, фаза = 0.

Перед ожиданием 0:: t2, время =22:01:28.728, зарегистрировано = 1, прибыло = 0, не получено = 1, фаза = 0.

Перед ожиданием 0:: t1, время =22:01:28.728, зарегистрировано = 1, прибыло = 0, не получено = 1, фаза = 0.

После продвижения 0:: t1, время =22:01:28.728, зарегистрировано = 1, прибыло = 0, не получено = 1, фаза = 3.

После продвижения 0:: t2, время =22:01:28.728, зарегистрировано = 1, прибыло = 0, не получено = 1, фаза = 3.

После продвижения 0:: t3, время =22:01:28.729, зарегистрировано = 1, прибыло = 0, не получено = 1, фаза = 3.

Перед ожиданием 1:: t2, время =22:01:30.730, зарегистрировано = 1, прибыло = 0, не получено = 1, фаза = 3.

Перед ожиданием 1:: t3, время =22:01:30.730, зарегистрировано = 1, прибыло = 0, не получено = 1, фаза = 3.

После продвижения 1:: t2, время =22:01:30.730, зарегистрировано = 1, прибыло = 0, не получено = 1, фаза =4.

После продвижения 1:: t3, время =22:01:30.732, зарегистрировано = 1, прибыло = 0, не получено = 1, фаза =5.

Перед ожиданием 1:: t1, время =22:01:30.730, зарегистрировано = 1, прибыло = 0, необработано = 1, фаза = 3.

После продвижения 1:: t1, время =22:01:30.732, зарегистрировано = 1, прибыло = 0, не получено = 1, фаза =6.

Вы можете увидеть здесь много расхождений. Потоки не продвигаются в последовательности. Кроме того, есть несколько пропущенных или / и не последовательных фаз.

Когда я переместил строку кода this.phaser.register() из конструктора в начало метода run, вывод был:

После регистра: t1, время =22:10:58.230, зарегистрировано = 3, прибыло = 0, не получено = 3, фаза = 0.

После регистра: t3, время =22:10:58.230, зарегистрировано = 3, прибыло = 0, не получено = 3, фаза = 0.

После регистра: t2, время =22:10:58.230, зарегистрировано = 3, прибыло = 0, не получено = 3, фаза = 0.

Перед ожиданием 0:: t2, время =22:11:00.314, зарегистрировано = 3, прибыло = 0, не получено = 3, фаза = 0.

Перед ожиданием 0:: t1, время =22:11:00.314, зарегистрировано = 3, прибыло = 0, не получено = 3, фаза = 0.

Перед ожиданием 0:: t3, время =22:11:00.314, зарегистрировано = 3, прибыло = 0, не получено = 3, фаза = 0.

После продвижения 0:: t2, время =22:11:00.315, зарегистрировано = 3, прибыло = 0, не получено = 3, фаза = 1.

После продвижения 0:: t3, время =22:11:00.315, зарегистрировано = 3, прибыло = 0, не получено = 3, фаза = 1.

После продвижения 0:: t1, время =22:11:00.315, зарегистрировано = 3, прибыло = 0, не получено = 3, фаза = 1.

Перед ожиданием 1:: t1, время =22:11:02.319, зарегистрировано = 3, прибыло = 0, необработано = 3, фаза = 1.

Перед ожиданием 1:: t2, время =22:11:02.319, зарегистрировано = 3, прибыло = 0, необработано = 3, фаза = 1.

Перед ожиданием 1:: t3, время =22:11:02.319, зарегистрировано = 3, прибыло = 0, необработано = 3, фаза = 1.

После продвижения 1:: t3, время = 22: 11: 02.320, зарегистрировано = 3, прибыло = 0, не получено = 3, фаза = 2.

После продвижения 1:: t2, время = 22: 11: 02.320, зарегистрировано = 3, прибыло = 0, не получено = 3, фаза = 2.

После продвижения 1:: t1, время = 22: 11: 02.321, зарегистрировано = 3, прибыло = 0, не получено = 3, фаза = 2.

Это выглядит намного лучше, чем выполнение потоков и этапов в последовательности.

Вот мои вопросы:

1) Почему существует много расхождений, когда стороны были зарегистрированы внутри конструктора Runnable?

2) Во втором результате статистические данные для прибывших и необработанных равны нулю (неверно) на каждом этапе. Итак, как получить правильные цифры для них?

Любая помощь приветствуется.

Answer 1

В первом примере "создание Phaser в конструкторе" вы регистрируете только один поток в Phaser. Вам нужно создать три задачи, зарегистрировать три потока в phaser.

Измените код, как это, и он будет работать. (не забудьте удалить начальную задачу RunnableTask = 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();
    }
  }

Ваш второй пример работает, но он не корректен. Он работает только потому, что вы спите в начале метода run, поэтому все потоки перехватывают регистрацию в phaser до того, как вы вызовете метод continue и продвигают его в phaser. Если вы удалите сон, то после вызова этой линии

t1.start();

Будет запущен метод запуска T1, а поток t1 будет зарегистрирован в фазере. Тогда возможно, что метод this.phaser.arriveAndAwaitAdvance() в t1 run будет вызван до того, как потоки t2 и t3 будут запущены и зарегистрированы в фазере, поэтому фазер не будет их ждать.

Вы должны зарегистрироваться в phaser в конструкторе задачи или в методе, который вызывается до запуска потока.