Lock과 Condition (await()과 signal(),signalAll())

synchronized 블록으로 동기화를 구현하면 원하는 쓰레드를 선택해서 notify()를 할 수 없어 경쟁상태(race condition)와 기아상태(starvation)가 발생할 수 있습니다. 해당 내용은 아래 포스팅에서 확인하실 수 있습니다!

https://silver-programmer.tistory.com/entry/wait%EA%B3%BC-notify-notifyAll

wait()과 notify(), notifyAll()

 

이 때 synchronized가 아닌, 자바에서 제공하는 lock 클래스들을 이용하여 동기화하면 쓰레드 별로 구별해서 notify() 할 수 있습니다.  즉, wait() & notify() 형식이 아니라, Lock & Condition 으로 쓰레드의 종류를 구분할 수 있습니다. 이번 포스팅은 이와 관련된 내용을 준비하였습니다!

---

lock 클래스 

lock 클래스의 메서드를 사용하면 자동적으로 lock의 잠금과 해제를 관리하는 synchronized와 달리, 수동으로 lock을 잠그고 해제해야 합니다. 즉, 아래와 같이 코드가 변경됩니다.

---

/*
synchronized: 자동 lock 관리
*/
synchronized(lock){//임계 영역}

/*
lock: 수동 lock 관리
*/

lock.lock();
// 임계 영역
lock.unlock();

---

lock 클래스에는 아래와 같이 3가지 종류가 있습니다.

ReentrantLock // 재진입 가능한 lock. 가장 일반적인 배타 lock
ReentrantReadWriteLock // 읽기에는 공유적이고, 쓰기에는 배타적인 lock
StampedLock // JDK1.8부터 추가된, ReentrantReadWriteLock에 낙관적인 lock 기능 추가

 

1. ReentrantLock

이때까지 lock이라고 불려 왔던 lock입니다. 읽기, 쓰기 상관없이 오직 한 쓰레드만 lock을 획득할 수 있습니다. 

---

public void method() {
        ReentrantLock lock = new ReentrantLock();
        lock.lock();
        try{
            //임계 영역
        } finally{
            lock.unlock(); // finally 블럭내에 적음으로써 어떠한 상황에서도 lock이 풀릴 수 있도록 한다.
        }
}

---

2. ReentrantReadWriteLock

- 특정 임계 영역에 읽기 lock이 걸려있으면, 다른 쓰레드들도 읽기 lock을 중복해서 걸 수 있어 여러 쓰레드가 동시에 읽을 수 있습니다. 하지만 쓰기 lock은 걸 수 없습니다.

- 특정 임계 영역에 쓰기 lock이 걸려있으면, 읽기 lock을 걸 수 없고, 다른 쓰레드들이 중복해서 쓰기 lock 또한 걸 수 없습니다.

---

3. StampedLock

- lock을 걸거나 해지할 때 '스탬프(long 타입의 정수값)'을 사용하며, 읽기와 쓰기를 위한 lock 외에 '낙관적 읽기 lock(optimistic reading lock)'가 추가된 것입니다.

- 낙관적 읽기 lock이란, 읽기 lock이 걸려있는 상태에서 쓰기 lock이 lock을 얻으려면 읽기 lock이 풀릴 때까지 기다리는 것이 아니라, 쓰기 lock에 의해 읽기 lock을 바로 풀게하는 lock입니다. 따라서, 낙관적 읽기에 실패하면 읽기 lock을 얻어서 다시 읽어 와야 합니다. 

---

int getBalance(){
        StampedLock lock=new StampedLock();
        long stamp = lock.tryOptimisticRead(); // 낙관적 읽기 lock을 건다.
        int curBalance = this.balance;
        
        if(lock.validate(stamp)){ // 쓰기 lock에 의해 낙관적 읽기 lock이 풀렸는지 확인
            stamp = lock.readLock(); // 읽기 lock이 풀렸으면, 읽기 lock을 얻으려고 기다린다.
            
            try{
                curBalance = this.balance;
            }finally {
                lock.unlockRead(stamp); // 읽기 lock을 푼다.
            }
        }
        return curBalance; // 낙관적 읽기 lock이 풀리지 않았으면 곧바로 읽어온 값 반환
    }

---

Condition

이전 포스팅의 요리사와 손님 예제에서처럼, 구분하고 싶은 쓰레드를 각각 다른 Condition으로 만들어서 각각의 waiting pool에서 따로 기다리도록 할 수 있습니다. 아래 코드와 같이  이미 생성된 lock으로부터 newCondition()을 호출하여 생성합니다.

---

ReentrantLock lock = new ReentrantLock();

Condition forCook = lock.newCondition();
Condition forCustomer = lock.newCondition();

 

그 후, wait() & notify()/notifyAll() 대신 Condition의 await() & signal()/signalAll()을 이용하면 됩니다.

---

public class Main {

    public static void main(String[] args) throws InterruptedException {
        Table table = new Table(); // 여러 쓰레드가 공유하는 객체

        new Thread(new Cook(table), "COOK1").start();
        new Thread(new Customer(table, "donut"), "Customer1").start();
        new Thread(new Customer(table, "burger"), "Customer2").start();

        Thread.sleep(2000);
        System.exit(0);
    }

}

class Table {
    String[] dishNames = {"donut", "donut", "burger"};
    final int MAX_FOOD_ON_TABLE = 6; // table에 놓을 수 있는 음식의 최대 개수
    private ArrayList<String> dishesOnTable = new ArrayList<>(); // table 위에 있는 음식 리스트
    private ReentrantLock lock = new ReentrantLock();
    private Condition forCook = lock.newCondition();
    private Condition forCustomer = lock.newCondition();

    public void addDishToTable(String dish) { // table에 음식 추가하는 메서드
        lock.lock(); // synchronized 삭제

        try {
            while (dishesOnTable.size() >= MAX_FOOD_ON_TABLE) { // table에 음식이 가득찼으면 추가하지 않는다.
                String name = Thread.currentThread().getName();
                System.out.println(name + " is waiting."); // Cook 쓰레드가 기다릴 것이다.
                try {
                    forCook.await(); // COOK 쓰레드를 기다리게 한다. (sleep)
                    Thread.sleep(500);
                } catch (InterruptedException e) {}
            }
            dishesOnTable.add(dish); // 음식 추가
            forCustomer.signal(); // 기다리고 있는 Customer 쓰레드를 깨운다. (notify)
            System.out.printf("Table has %s. Total %d/%d dishes.\n", dishesOnTable.toString(), dishesOnTable.size(), MAX_FOOD_ON_TABLE);
        } finally {
            lock.unlock();
        }
    }

    public void removeDishFromTable(String dish) { // 해당 dish를 table에서 제거하는 메서드
        lock.lock();    //synchronized (this) {
        String name = Thread.currentThread().getName();

        try {
            while (dishesOnTable.size() == 0) {
                System.out.println(name + " is waiting.");
                try {
                    forCustomer.await(); // Customer 쓰레드를 기다리게 한다. (wait)
                    Thread.sleep(500);
                } catch (InterruptedException e) {
                }
            }

            while (true) {
                for (String s : dishesOnTable) {
                    if (dish.equals(s)) {
                        dishesOnTable.remove(s);
                        forCook.signal(); // COOK 쓰레드를 깨운다. (notify)
                        return; // 원하는 음식을 찾았으므로 먹으러 가야한다.
                    }
                }

                try { // 원하는 음식이 없는 Customer를 기다리게 한다.
                    System.out.println(name + " is waiting.");
                    forCustomer.await(); // Customer 쓰레드를 기다리게 한다. (wait)
                    Thread.sleep(500);
                } catch (InterruptedException e) {}

            } // while(true)
            // } synchronized
        } finally {
            lock.unlock();
        }
    }

    }

    public int dishNum() {
        return dishNames.length;
    }
}

class Customer implements Runnable {
    private Table table;
    private String food;

    Customer(Table table, String food) {
        this.table = table;
        this.food = food;
    }

    @Override
    public void run() {
        while (true) {
            try {
                Thread.sleep(100);
            } catch (InterruptedException e) {}
            String name = Thread.currentThread().getName();
            table.removeDishFromTable(food);
            System.out.println(name + " ate a " + food);
        }
    }
}

class Cook implements Runnable {
    private Table table;

    public Cook(Table table) {
        this.table = table;
    }

    @Override
    public void run() {
        while (true) {
            // 임의의 음식을 가지고 온 후, 요리사가 요리하여 table에 추가하는 메서드
            int idx = (int) (Math.random() * table.dishNum());
            table.addDishToTable(table.dishNames[idx]);

            try {
                Thread.sleep(10);
            } catch (InterruptedException e) {}
        }
    }
}

 

물론, 구분된 쓰레드 내부(Customer 쓰레드 별로)에서 '기아 현상'이나 '경쟁 상태'가 발생할 수 있습니다. 이 경우에는 더 세분화해서 Condition을 구성함으로써 개선해볼 수 있겠습니다. 

 

감사합니다!!

---

[참고자료]

남궁 성, [Java의 정석 3rd Edition], 도우출판, 2016