Заметка про wait notify примитивы в Java

Давайте попробуем решить следующую задачу. Есть класс Container:

class Container {
    public String value;

    public void put(String value) {
        assert(this.value == null);
        this.value = value;
    }

    public String get() {
        assert(this.value != null);
        String value = this.value;
        this.value = null;
        return value;
    }
}

Мы хотим класть в него значения и забирать методами put() и get() Мы считаем ошибкой попытку положить в полный контейнер и забрать из пустого. Создадим и запустим два потока, один кладет текущее время в контейнер, а другой читает что есть в контейнере:

    public static void main(String[] args) {
        Container container = new Container();

        Thread producer = new Thread(() -> {
            while (true) {
                container.put(String.valueOf(System.currentTimeMillis()));
            }
        });


        Thread consumer = new Thread(() -> {
            while (true) {
                System.out.println(container.get());
            }
        });
        producer.start();
        consumer.start();
    }

Это предсказуемо падает с ошибкой:

1524892150711
Exception in thread "Thread-0" Exception in thread "Thread-1" java.lang.AssertionError
	at ru.yamakarov.Container.put(WaitNotifyExample.java:31)
	at ru.yamakarov.WaitNotifyExample.lambda$main$0(WaitNotifyExample.java:11)
	at java.lang.Thread.run(Thread.java:748)
java.lang.AssertionError
	at ru.yamakarov.Container.get(WaitNotifyExample.java:36)
	at ru.yamakarov.WaitNotifyExample.lambda$main$1(WaitNotifyExample.java:18)
	at java.lang.Thread.run(Thread.java:748)

Два потока никак не синхронизованы. Попробуем это сделать вот так:

Thread producer = new Thread(() -> {
    synchronized (container) {
        container.put(String.valueOf(System.currentTimeMillis()));
        container.notify();
    }
    while (true) {
        synchronized (container) {
            try {
                container.wait();
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            System.out.println("producer");
            container.put(String.valueOf(System.currentTimeMillis()));
            container.notify();
        }
    }
});


Thread consumer = new Thread(() -> {
    while (true) {
        synchronized (container) {
            try {
                container.wait();
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            System.out.println("consumer");
            System.out.println(container.get());
            container.notify();
        }
    }
});

И этот код какое-то время работает.

...
1524893196499
producer
consumer
1524893196499
producer
consumer
1524893196499
producer
consumer
1524893196499
producer
consumer
1524893196499
producer

Однако заканчивается все тем, что либо consumer пропускает notify либо producer. Судя по документации может быть спонтанное пропысание на wait:

      As in the one argument version, interrupts and spurious wakeups are
      possible...

Тогда должен вылететь ошибка на assert. Но у меня такого не происходило. Получается представленное мной решение, одно из самых плохих. Он некоторое время работает и в неизвестный момент ломается. Хуже оно было бы, если ломалось не так быстро, скажем только при spurious wakeups. Но я не знаю как этого добиться. Вот другое решение с использованием флагa:

class ContainerState {
    public boolean produced;
}

Код запуска:

ContainerState containerState = new ContainerState();

Thread producer = new Thread(() -> {
    while (true) {
        if (!containerState.produced) {
            container.put(String.valueOf(System.currentTimeMillis()));
            containerState.produced = true;
        }
    }
});


Thread consumer = new Thread(() -> {
    while (true) {
        if (containerState.produced) {
            System.out.println(container.get());
            containerState.produced = false;
        }
    }
});

Неожиданным в этом решении оказалось то, что в какой то момент вывод времени прекращается. Я не понял, что останавливается, потому что добавление System.out.pritln() убирает этот эффект. А дебажить многопоточный код, вообще ад. У меня подозрение в том, что один из тредов выбирает все ресурсы и второй перестает выполняться полностью. Таким образом флаг produced не меняется и ничего не происходит. Thread.yield() помогает на моей машине, но может не помочь на вашей судя по документации.

И начинает вылетать забавное исключение, после некоторого времени выполнения. По моей гипотезе, это разогрев JVM:

...
1524895907057
Exception in thread "Thread-1" java.lang.AssertionError
	at ru.yamakarov.Container.get(WaitNotifyExample.java:48)
	at ru.yamakarov.WaitNotifyExample.lambda$main$1(WaitNotifyExample.java:26)
	at java.lang.Thread.run(Thread.java:748)

Такого же не может быть. Ведь мы сначала кладем значение а потом поднимаем флаг:

if (!containerState.produced) {
    container.put(String.valueOf(System.currentTimeMillis()));
    containerState.produced = true;
}

И вот может быть, так как оптимизитор может произвести перестановку этих операций с целью увеличения производительности. Можно полечить сделав produced volatile:

public volatile boolean produced;

Вот тут я нашел объяснение этому. Там что-то с кэшами и реодерингом. Так работает. Но что если producer будет работать гораздо медленнее consumer? consumer будет бессмысленно жечь электричество, проверяя, а не появились ли данные. Тут поможет wait() в while цикле:

Thread producer = new Thread(() -> {
    while (true) {
        System.out.println("producer");
        synchronized (containerState) {
            try {
                while(containerState.produced) {
                    containerState.wait();
                }
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
        if (!containerState.produced) {
            container.put(String.valueOf(System.currentTimeMillis()));
            containerState.produced = true;
        }
        synchronized (containerState) {
            containerState.notify();
        }
    }
});


Thread consumer = new Thread(() -> {
    while (true) {
        System.out.println("consumer");
        synchronized (containerState) {
            try {
                while(!containerState.produced) {
                    containerState.wait();
                }
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
        if (containerState.produced) {
            System.out.println(container.get());
            containerState.produced = false;
        }
        synchronized (containerState) {
            containerState.notify();
        }
    }
});

И теперь хороший стабильный вывод:

...
1524898482460
consumer
producer
1524898482460
consumer
producer
1524898482460
consumer
producer
1524898482460
consumer
producer
1524898482460
consumer
...

Я написал вот такой Producer-Consumer на wait и notify. Рекомендую попробовать самим написать нечто подобное, но в проде лучше используйте решение на очереди. Будете лучше спать по ночам без spurious wakeups.