[아이템 81] wait() notify() 의 예시

[csbsjy]

p.436, wait() 이후 또 검사하지 않으면 발생할 수 있는 문제 --> 동시에 두 스레드가 내부블록에 접근할 수 있는 예시를 짜자

[pci2676]

package Chap11_Concurrency.item81.waitnotify;

import org.junit.jupiter.api.Test;

import java.util.LinkedList;
import java.util.NoSuchElementException;
import java.util.Queue;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;

import static org.assertj.core.api.Assertions.assertThat;
import static org.assertj.core.api.Java6Assertions.assertThatThrownBy;

class CustomQueueTest {

    //  consumer1 in sync block
    //  consumer1 waiting
    //  consumer2 in sync block
    //  consumer2 waiting
    //  provider in sync block
    //  provider add item
    //  provider notifyAll
    //  consumer2 wake up
    //  consumer2 remove item
    //  consumer1 wake up
    //  consumer1 remove item
    // java.util.NoSuchElementException
    @Test
    void notSafe() throws InterruptedException {
        Queue<String> queue = new LinkedList<>();

        CountDownLatch consumer1CountDownLatch = new CountDownLatch(1);
        CountDownLatch consumer2CountDownLatch = new CountDownLatch(1);
        CountDownLatch endCountDown = new CountDownLatch(1);

        Thread consumer1 = new Thread(
                () -> {
                    try {
                        synchronized (queue) {
                            System.out.println("consumer1 in sync block ");

                            if (queue.isEmpty()) {
                                System.out.println("consumer1 waiting ");
                                consumer1CountDownLatch.countDown();
                                queue.wait();
                                System.out.println("consumer1 wake up");
                            }
                            System.out.println("consumer1 remove item");

                            assertThatThrownBy(queue::remove)
                                    .isInstanceOf(NoSuchElementException.class);
                            endCountDown.countDown();
                        }
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                }
        );
        Thread consumer2 = new Thread(
                () -> {
                    try {
                        synchronized (queue) {
                            System.out.println("consumer2 in sync block ");
                            if (queue.isEmpty()) {
                                System.out.println("consumer2 waiting ");
                                consumer2CountDownLatch.countDown();
                                queue.wait();
                                System.out.println("consumer2 wake up");
                            }
                            System.out.println("consumer2 remove item");
                            queue.remove();
                        }
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                }
        );

        Thread provider = new Thread(
                () -> {
                    try {
                        synchronized (queue) {
                            System.out.println("provider in sync block ");
                            System.out.println("provider add item ");
                            queue.add("SOME THING");
                            System.out.println("provider notifyAll");
                            queue.notifyAll();
                        }
                    } catch (Exception e) {
                        e.printStackTrace();
                    }
                }
        );

        consumer1.start();
        consumer1CountDownLatch.await();

        consumer2.start();
        consumer2CountDownLatch.await();

        provider.start();
        endCountDown.await();

        assertThat(endCountDown.getCount()).isEqualTo(0);
    }

// 아래는 wait에서 동시에 스레드가 깨어났을때 안전하게 다시 wait으로 돌아가는 케이스

    @Test
    void safe() throws InterruptedException {
        Queue<String> queue = new LinkedList<>();

        CountDownLatch consumer1CountDownLatch = new CountDownLatch(1);
        CountDownLatch consumer2CountDownLatch = new CountDownLatch(1);
        CountDownLatch endCountDown = new CountDownLatch(1);

        Thread consumer1 = new Thread(
                () -> {
                    try {
                        synchronized (queue) {
                            System.out.println("consumer1 in sync block ");

                            while (queue.isEmpty()) {
                                System.out.println("consumer1 waiting ");
                                consumer1CountDownLatch.countDown();
                                queue.wait();
                                System.out.println("consumer1 wake up");
                            }
                            System.out.println("consumer1 remove item");

                            queue.remove();
                            endCountDown.countDown();
                        }
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                }
        );
        Thread consumer2 = new Thread(
                () -> {
                    try {
                        synchronized (queue) {
                            System.out.println("consumer2 in sync block ");
                            while (queue.isEmpty()) {
                                System.out.println("consumer2 waiting ");
                                consumer2CountDownLatch.countDown();
                                queue.wait();
                                System.out.println("consumer2 wake up");
                            }
                            System.out.println("consumer2 remove item");
                            queue.remove();
                        }
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                }
        );

        Thread provider = new Thread(
                () -> {
                    try {
                        synchronized (queue) {
                            System.out.println("provider in sync block ");
                            System.out.println("provider add item ");
                            queue.add("SOME THING");
                            System.out.println("provider notifyAll");
                            queue.notifyAll();
                        }
                    } catch (Exception e) {
                        e.printStackTrace();
                    }
                }
        );

        consumer1.start();
        consumer1CountDownLatch.await();

        consumer2.start();
        consumer2CountDownLatch.await();

        provider.start();
        endCountDown.await(3, TimeUnit.MILLISECONDS);

        assertThat(endCountDown.getCount()).isEqualTo(1);
    }
}

[pci2676]

notifyAll을 실행한다 하더라도 wait이 동시에 전부 일어나서 각각 스레드의 sync 블럭이 동시에 실행되는 것이 아니라 동기적으로 wait이 하나씩 깨어나기 때문에 while을 사용해서 검사하면 방어할 수 있다.