생산자 소비자 문제, Lock Condition, 생산자 소비자 대기 공간 분리, 스레드의 대기, 중간 정리, BlockingQueue

출처 : 인프런 - 김영한의 실전 자바 - 고급1편 (유료) / 김영한님  
유료 강의이므로 정리에 초점을 두고 코드는 일부만 인용

---

1. Lock Condition - 예제4

1) Lock Condition

(1) 해결 방안

• 생산자 소비자 문제를 해결하기 위해 wait()와 notify()를 사용했지만 생산자가 생산자를 깨우고 소비자가 소비자를 깨우는 비효율 문제가 있었음
• 핵심은 생산자 스레드는 데이터를 생성하고 대기중인 소비자 스레드에게 알려주어야 하고 소비자 스레드는 데이터를 소비하고 대기중인 생산자 스레드에게 알려주면 됨
• 즉, 생산자 스레드가 대기하는 대기 집합과 소비자 스레드가 대기하는 대기 집합을 둘로 나누면 생산자 스레드가 데이터를 생산하면 소비자 스레드가 대기하는 대기 집합에만 알려주고 소비자 스레드가 데이터를 소비하면 생산자 스레드가 대기하는 대기 집합에만 알려주면 해결됨
• 대기 집합을 분리하는 방법은 앞서 학습한 Lock, ReentrantLock을 사용하면 됨

(2) BoundedQueueV4

• synchronized, wait(), notify()를 통해 작성한 코드를 Lock 인터페이스와 ReentrantLock 구현체를 사용하도록 변경
• Condition condition = lock.newCondition()
  • Condition은 ReentrantLock을 사용하는 스레드가 대기하는 스레드 대기 공간임
  • lock.newCondition() 메서드를 호출하면 Lock(ReentrantLock)의 스레드 대기 공간이 만들어짐
  • Object.wait()에서 사용한 스레드 대기 공간은 모든 객체 인스턴스가 내부에 기본으로 가지고 있는 반면 Lock(ReentrantLock)을 사용하는 경우 스레드 대기 공간을 직접 만들어서 사용해야 함
• condition.await(): wait()와 유사한 기능으로 지정한 condition에 현재 스레드를 대기 상태로 보관함, 이때 ReentrantLock에서 획득한 락을 반납하고 대기 상태로 condition에 보관됨
• condition.signal(): notify()와 유사한 기능으로 지정한 condition에서 대기 중인 스레드를 하나 깨운 후 깨어난 스레드는 condition에서 빠져나옴

package thread.bounded;

public class BoundedQueueV4 implements BoundedQueue {

    private final Lock lock = new ReentrantLock();
    private final Condition condition = lock.newCondition();    // 스레드가 기다리는 대기 집합

    private final Queue<String> queue = new ArrayDeque<>();
    private final int max;

    public BoundedQueueV4(int max) {
        this.max = max;
    }

    @Override
    public void put(String data) {
        lock.lock();
        try {
            while (queue.size() == max) {
                log("[put] 큐가 가득 참, 생산자 대기");
                try {
                    condition.await();  // wait() 대신 await()를 사용
                    log("[put] 생산자 깨어남");
                } catch (InterruptedException e) {
                    throw new RuntimeException(e);
                }
            }

            queue.offer(data);
            log("[put] 생산자 데이터 저장, signal() 호출");
            condition.signal(); // notify() 대신 signal()을 사용

        } finally {
            lock.unlock();
        }
    }

    @Override
    public String take() {
        lock.lock();
        try {
            while (queue.isEmpty()) {
                log("[take] 큐에 데이터가 없음, 소비자 대기");
                try {
                    condition.await();  // wait() 대신 await()를 사용
                    log("[take] 소비자 깨어남");
                } catch (InterruptedException e) {
                    throw new RuntimeException(e);
                }
            }
            String data = queue.poll();
            log("[take] 소비자 데이터 획득, signal() 호출");
            condition.signal(); // notify() 대신 signal()을 사용
            return data;
        } finally {
            lock.unlock();
        }
    }

    @Override
    public String toString() {
        return queue.toString();
    }
}

 

(3) 설명

• 이 그림에서 lock은 synchronized에서 사용하는 객체 내부에 있는 모니터 락이 아니라 ReentrantLock 락을 뜻하며 ReentrantLock은 내부에 락과 락 획득을 대기하는 스레드를 관리하는 대기 큐가 있음
• 그림에서의 스레드 대기 공간은 synchronized에서 사용하는 스레드 대기 공간이 아니라 lock.newCondition()으로 반환된 condition이 스레드 대기 공간임
• 여기까지 보면 생산자, 소비자용 스레드 대기 공간을 따로 분리하지 않았기 때문에 기존의 방식인 synchronized, wait(), notify()를 사용한 이전 코드와 동일하며 구현만 ReentranLock으로 변경하였으므로 BoundedMain에서 BoundedQueueV4를 사용하도록 변경하고 실행해도 V3와 동일한 결과가 출력됨

---

2. 생산자 소비자 대기 공간 분리 - 예제5

1) 코드

(1) BoundedQueueV5

• lock.newCondition()을 두 번 호출해서 ReentrantLock을 사용하는 스레드 대기 공간을 2개 생성하여 Condition을 분리
• put(data) - 생산자 스레드가 호출
  • 큐가 가득 찬 경우: producerCond.await()를 호출해서 생산자 스레드를 생산자 전용 스레드 대기 공간에 보관
  • 데이터를 저장한 경우: 생산자가 데이터를 생산하면 큐에 데이터가 추가되고 소비자를 깨워야 하므로 consumerCond.signal()을 호출해서 소비자 전용 스레드 대기 공간에 신호를 보내서 대기중인 소비자 스레드를 하나 깨움
• take() - 소비자 스레드가 호출
  • 큐가 빈 경우: consumerCond.await()를 호출해서 소비자 스레드를 소비자 전용 스레드 대기 공간에 보관
  • 데이터를 소비한 경우: 소비자가 데이터를 소비한 경우 큐에 여유공간이 생기면 생산자를 깨워야하므로 producerCond.signal()를 호출하여 생산자 전용 스레드 대기 공간에 신호를 보내서 대기 중인 생산자 스레드를 하나 깨움
• 핵심은 스레드 대기 공간을 분리하여 생산자는 소비자를 깨우고, 소비자는 생산자를 깨운다는 점임

package thread.bounded;

public class BoundedQueueV5 implements BoundedQueue {

    private final Lock lock = new ReentrantLock();
    private final Condition producerCond = lock.newCondition();    // 생산자 스레드 대기 집합
    private final Condition consumerCond = lock.newCondition();    // 소비자 스레드 대기 집합

    private final Queue<String> queue = new ArrayDeque<>();
    private final int max;

    public BoundedQueueV5(int max) {
        this.max = max;
    }

    @Override
    public void put(String data) {
        lock.lock();
        try {
            while (queue.size() == max) {
                log("[put] 큐가 가득 참, 생산자 대기");
                try {
                    producerCond.await();   // 생산자 대기
                    log("[put] 생산자 깨어남");
                } catch (InterruptedException e) {
                    throw new RuntimeException(e);
                }
            }
            queue.offer(data);
            log("[put] 생산자 데이터 저장, signal() 호출");
            consumerCond.signal();  // 소비자 깨우기

        } finally {
            lock.unlock();
        }
    }

    @Override
    public String take() {
        lock.lock();
        try {
            while (queue.isEmpty()) {
                log("[take] 큐에 데이터가 없음, 소비자 대기");
                try {
                    consumerCond.await();   // 소비자 대기
                    log("[take] 소비자 깨어남");
                } catch (InterruptedException e) {
                    throw new RuntimeException(e);
                }
            }
            String data = queue.poll();
            log("[take] 소비자 데이터 획득, signal() 호출");
            producerCond.signal();  // 소비자 깨우기
            return data;
        } finally {
            lock.unlock();
        }
    }

    @Override
    public String toString() {
        return queue.toString();
    }
}

 

(2) BoundedMain

• 수정한 BoundedQueueV5를 사용하도록 실행해 보면 생산자 먼저 실행했을 경우에는 비효율이 원래 발생하지 않았기 때문에 기존과 동일함
• 그러나 소비자 먼저 실행했을 경우에 실행 결과를 보면 소비자가 소비자를 깨우는 로그가 사라지고 소비자가 생산자를 깨움으로써 비효율 로직이 없어진 것을 확인할 수 있음

/* 실행 결과
12:12:20.440 [     main] == [소비자 먼저 실행] 시작, BoundedQueueV5 ==

12:12:20.441 [     main] 소비자 시작
12:12:20.444 [consumer1] [소비 시도]     ? <- []
12:12:20.444 [consumer1] [take] 큐에 데이터가 없음, 소비자 대기
12:12:20.544 [consumer2] [소비 시도]     ? <- []
12:12:20.544 [consumer2] [take] 큐에 데이터가 없음, 소비자 대기
12:12:20.649 [consumer3] [소비 시도]     ? <- []
12:12:20.650 [consumer3] [take] 큐에 데이터가 없음, 소비자 대기

12:12:20.754 [     main] 현재 상태 출력, 큐 데이터: []
12:12:20.755 [     main] consumer1: WAITING
12:12:20.755 [     main] consumer2: WAITING
12:12:20.755 [     main] consumer3: WAITING

12:12:20.755 [     main] 생산자 시작
12:12:20.756 [producer1] [생산 시도] data1 -> []
12:12:20.756 [producer1] [put] 생산자 데이터 저장, signal() 호출
12:12:20.756 [consumer1] [take] 소비자 깨어남
12:12:20.756 [producer1] [생산 완료] data1 -> [data1]
12:12:20.756 [consumer1] [take] 소비자 데이터 획득, signal() 호출
12:12:20.757 [consumer1] [소비 완료] data1 <- []
12:12:20.859 [producer2] [생산 시도] data2 -> []
12:12:20.859 [producer2] [put] 생산자 데이터 저장, signal() 호출
12:12:20.859 [producer2] [생산 완료] data2 -> [data2]
12:12:20.859 [consumer2] [take] 소비자 깨어남
12:12:20.860 [consumer2] [take] 소비자 데이터 획득, signal() 호출
12:12:20.860 [consumer2] [소비 완료] data2 <- []
12:12:20.964 [producer3] [생산 시도] data3 -> []
12:12:20.965 [producer3] [put] 생산자 데이터 저장, signal() 호출
12:12:20.965 [producer3] [생산 완료] data3 -> [data3]
12:12:20.965 [consumer3] [take] 소비자 깨어남
12:12:20.965 [consumer3] [take] 소비자 데이터 획득, signal() 호출
12:12:20.965 [consumer3] [소비 완료] data3 <- []

12:12:21.067 [     main] 현재 상태 출력, 큐 데이터: []
12:12:21.067 [     main] consumer1: TERMINATED
12:12:21.068 [     main] consumer2: TERMINATED
12:12:21.068 [     main] consumer3: TERMINATED
12:12:21.068 [     main] producer1: TERMINATED
12:12:21.068 [     main] producer2: TERMINATED
12:12:21.069 [     main] producer3: TERMINATED
12:12:21.069 [     main] == [소비자 먼저 실행] 종료, BoundedQueueV5 ==
*/

2) 분석 - 쉽게 설명하기 위의 실행결과보다 단순한 예시로 분석

(1) 생산자 실행

• c1, c2, c3는 소비자 스레드 전용 대기 공간(consumerCond)에 대기 중이고 p1, p2, p3는 생산자 전용 스레드 대기 공간(producerCond)에 대기 중이며 큐에 데이터가 비어있고 생산자인 p0 스레드가 실행 예정이라고 가정함
• 생산자 스레드인 p0 ReentrantLock의 락을 획득하고 큐에 데이터를 보관했기 때문에 소비자 스레드가 가져갈 데이터가 추가되었으므로 소비자 대기공간에 signal()을 통해 알려줌
• 소비자 스레드 중에 하나가 깨어나고 깨어난 소비자 스레드는 락 획득까지 잠시 대기하다가 이후에 p0이 반납한 ReentrantLock을 획득하고 이후 로직을 처리함

(2) 소비자 실행

• 반대로 큐에 데이터가 가득한 상태로 소비자 스레드인 c0가 실행된다고 가정
• c0 스레드는 ReentrantLock의 락을 획득하고 큐에 있는 데이터를 획득하면 큐에 데이터를 생산할 수 있는 빈 공간이 생겼으므로 생산자 대기 공간에 signal()을 통해 알려줌
• 생산자 스레드 중에 하나가 깨어나면 c0가 반납한 락을 획득하고 데이터를 저장한 다음 완료됨

(3) Object.notify() vs Condition.signal()

• notify()
  • 대기 중인 스레드 중 임의의 하나를 선택해서 깨우는데 스레드가 깨어나는 순서가 정의되어 있지 않고 JVM 구현에 다름
  • 보통은 먼저 들어온 스레드가 먼저 수행되지만 구현에 따라 다를 수 있음
  • synchronized 블록 내에서 모니터 락을 가지고 있는 스레드가 호출해야 함
• signal()
  • 대기 중인 스레드 중 하나를 깨우며 보통 Condition의 구현은 Queue 구조를 사용하기 때문에 FIFO 순서로 깨우지만 자바 버전과 구현에 따라 달라질 수 있음
  • ReentrantLock을 가지고 있는 스레드가 호출해야 함

---

3. 스레드의 대기

1) 대기 상태 정리

(1) synchronized 대기 상태

• 대기1: synchronized를 시작할 때 락이 없으면 BLOCKED 상태로 락 획득 대기하고 다른 스레드가 synchronized를 빠져나갈 때 대기가 풀리며 락 획득 시도
• 대기2: wait()를 호출했을 때 WAITING 상태로 스레드 대기 집합에서 대기하고 다른 스레드가 notify()를 호출 했을 때 빠져나감
• 소비자 스레드 c1, c2, c3가 동시에 실행되고 c1이 먼저 락을 획득하고 c2, c3가 락 획득을 위해 대기하며 BLOCKED 상태가 된다고 가정했을 때 잘 생각해 보면 락을 기다리는 c2, c3도 어딘가에서 관리되어야 락이 반환되었을 때 자바가 c2, c3 중에 하나를 선택해서 락을 제공할 수 있음
• 여태 그림에서는 그냥 스레드가 BLOCKED 상태로 변경만 되는 것처럼 설명했지만 사실 BLOCKED 상태의 스레드도 자바 내부에서 따로 관리됨

(2-1) 락 대기 집합 설명 - 소비자 스레드 실행

• 조금 더 자세히 그린 그림을 보면 자바 내부의 락 대기 집합이라는 곳에서 락을 기다리는 BLOCKED 상태의 스레드들을 관리함
• 락 대기 집합은 자바 내부에 구현되어 있기 때문에 모니터 락과 같이 개발자가 확인하기는 어려움
• 락 대기 집합을 한참 뒤에 설명한 이유는 스레드를 최대한 쉽고 단순하게 설명하기 위해 BLOCKED 상태에서 사용하는 락 대기 집합을 설명하지 않았으며 사실 락 대기 집합에 대한 내용을 몰라도 스레드를 이해하는 데에는 문제가 없음
• 다만 스레드가 모니터 락을 기대리는 상태와 wait()를 통한 스레드 대기 집합에서 대기하는 상태를 헷갈릴 수 있기 때문에 이 부분을 명확하게 하기 위해 설명함
• 즉 c1, c2, c3 동시에 생성되고 c1이 락을 획득하면 c2, c3는 락 대기 집합에서 BLOCKED상태고 대기하게 되고 c1은 큐의 데이터를 얻기 위해 확인하지만 데이터가 없기 때문에 락을 반납하고 WAITING 상태로 스레드 대기 집합에서 대기함
• 이후에 락 대기 집합에 있는 c2가 락을 획득하고 임계 영역을 수행하고 마찬가지로 큐에 데이터가 없으므로 스레드 대기 집합에서 대기하게 되고 c3로 동일한 로직을 수행하게 됨

(2-2) 락 대기 집합 설명 - 생산자 스레드 실행

• 이후 p1이 락을 획득하고 데이터를 저장한 다음 스레드 대기 집합에 알리면 스레드 대기 집합에 있는 스레드 중 하나가 깨어나서 스레드 대기 집합을 빠져나감
• 여기서는 c1이 깨어났다고 가정하면 c1은 락을 얻어서 락 대기 집합까지 빠져나가야 임계 영역을 수행할 수 있으므로 락을 얻을 때까지 락 대기 집합에서 BLOCKED 상태로 기다림
• p1이 락을 반납하면 c1이 반납된 락을 획득하고 그제야 임계 영역을 수행함
• 개념상 락 대기 집합이 1차 개기소이고 스레드 대기 집합이 2차 대기소이며 임계 영역을 안전하게 지키기 위한 2중 대기 장치가 마련되어 있고 이를 모두 탈출해야 임계 영역을 수행할 수 있음

(3) 정리

• 자바의 모든 객체 인스턴스는 멀티스레드와 임계 영역을 다루기 위해 내부에 3가지 기본 요소를 가짐
  1. 모니터락
  2. 락 대기 집합(모니터 락 대기 집합)
  3. 스레드 대기 집합
• 여기서 락 대기 집합이 1차 대기소이고 스레드 대기 집합이 2차 대기소라고 생각하면 되며 2차 대기소에 들어간 스레드는 2차, 1차 대기소를 모두 빠져나와야 임계 영역을 수행할 수 있음
• 이 3가지 요소는 서로 맞물려 돌아감
  
  • synchronized를 사용한 임계 영역에 들어가려면 모니터 락이 필요함
  • 모니터 락이 없으면 락 대기 집합에 들어가서 BLOCKED 상태로 락을 기다림
  • 모니터 락을 반납하면 락 대기 집합에 있는 스레드 중 하나가 락을 획득하고 BLOCKED -> RUNNABLE 상태가 됨
  • wait()를 호출해서 스레드 대기 집합에 들어가기 위해서는 모니터 락이 필요함
  • 스레드 대기 집합에 들어가면 모니터 락을 반납함
  • 스레드가 notify()를 호출하면 스레드 대기 집합에 있는 스레드 중 하나라 스레드 대기 집합을 빠져오고 모니터 락 획득을 시도하며 락을 얻으면 임계 영역을 수행하고 얻지 못하면 락 대기 집합에 들어가서 BLOCKED 상태로 락을 기다림

(4) ReentrantLock 대기

• synchronized와 마찬가지로 Lock(ReentrantLock)도 2가지 단계의 대기 상태가 존재하며 둘 다 같은 개념을 구현한 것이기 때문에 비슷함
• 대기1: ReentrantLock 락 획득 대기
  • 락이 없으면 ReentrantLock의 대기 큐에서 관리하고 WAITING 상태로 락 획득 대기
  • lock.lock()을 호출 했을 때 락이 없으면 대기하고 다른 스레드가 lock.unlock()을 호출 했을 때 대기가 풀리며 락 획득을 시도하고 락을 획득하면 대기 큐를 빠져나감
• 대기2: await() 대기
  • condition.await()를 호출했을 때, condition 객체의 스레드 대기 공간에서 관리하고 WAITING 상태로 대기함
  • 다른 스레드가 condition.signal()을 호출 했을 때 condition 객체의 스레드 대기 공간에서 빠져나감

(5) 2단계 대기소

• 깨어난 스레드는 바로 실행되는 것이 아니라 synchronized와 마찬가지로 ReentrantLock도 대기소가 2단계로 되어있음
• 임계 영역 안에서는 항상 락이 있는 하나의 스레드만 실행될 수 있으므로 2차 대기소인 condition 객체의 스레드 대기 공간을 빠져나온다고 바로 실행되는 것이 아님
• 깨어난 스레드가 ReentrantLock의 락을 획득하지 못하면 WAITING 상태로 락을 획득할 때까지 ReentrantLock의 대기 큐에서 대기하게 되고 락을 획득해야 RUNNABLE상태가 되면서 그다음 코드를 실행할 수 있음

---

4. 중간 정리 - 생산자 소비자 문제

1) V1 ~ V5 버전 중간 정리

(1) BoundedQueueV1

• 스레드를 제어할 수 없는 단순한 큐 자료구조로 버퍼가 가득 차거나 버퍼에 데이터가 없는 한정된 상태에서 문제가 발생했음

(2) BoundedQueueV2

• V1에서 발생한 문제를 해결하려고 sleep()을 도입했으나 임계 영역 안에서 락을 들고 대기하여 멈춰버리는 더 심각한 문제가 발생함

(3) BoundedQueueV3

• synchronized와 함께 사용할 수 있는 wait(), notify(), notifyAll()을 사용해서 문제를 해결하였음
• 그러나 스레드가 대기하는 대기 집합이 하나이기 때문에 원하는 스레드를 선택해서 깨울 수 없는 문제가 있어서 일부 상황에서 비효율이 발생함

(4) BoundedQueueV4

• V3의 한계를 극복해 보기 위해 Lock, ReentrantLock을 사용하도록 코드를 변경하였음
• 다음으로 넘어가기 위한 중간 단계의 코드로 실행 결과는 V3와 같고 구현방식을 변경함

(5) BoundedQueueV5

• Lock(ReentrantLock)은 Condition이라는 스레드 대기 공간을 제공하며 원하는 만큼 따로 만들 수 있음
• 생산자 스레드를 위한 전용 대기 공간과, 소비자 스레드를 위한 전용 대기 공간을 각각 만들 고 생산자가 데이터를 생산하고 나면 consumerCond.signal() 메서드를 통해 소비자 전용 대기 공간에 알리고, 소비자 데이터가 소비하고 나면 productCond.signal()을 통해 생산자 전용 대기 공간에 알릴 수 있게 됨
• 이렇게 대기 공간을 나누어서 앞서 V3에서 발생한 비효율 문제를 깔끔하게 해결하였음
• V5 버전은 단순한 큐의 기능을 넘어서 스레드를 효과적으로 제어하여 생산자 소비자 문제(한정된 버퍼 문제)를 매우 효율적으로 해결할 수 있는 자료 구조이므로 멀티스레드 상황에서 이런 문제가 나타난다면 V5 버전을 사용하면 됨
• 사실 V5 버전도 이미 어딘가에 만들어져 있음

(6) BlockingQueue

• BoundedQueue를 스레드 관점에서 보면 큐가 특정 조건이 만족될 때까지 스레드의 작업을 차단함
• 데이터 추가 차단: 큐가 가득 차면 데이터 추가 작업을 시도하는 스레드는 공간이 생길 때까지 차단됨
• 데이터 획득 차단: 큐가 비어 있으면 획득 작업을 시도하는 스레드는 큐에 데이터가 들어올 때까지 차단됨
• 자바는 생산자 소비자 문제, 또는 한정된 버퍼라고 불리는 문제를 해결하기 위해 java.util.concurrent.BlockingQueue라는 인터페이스와 구현체들을 제공함
• 처음부터 BlockingQueu를 사용하면 되는데 이렇게 직접 구현해 본 이유는 생산자 소비자 문제는 멀티스레드의 기본기를 배울 수 있는 가장 좋은 예시이기 때문임
• 왜 다른 스레드가 BLOCKED 상태에서 깨어날 수 없는지, synchronized, wait(), notify()가 왜 필요하고 한계점은 무엇인지, ReentrantLock을 왜 만들었고 Condition은 왜 필요한지, 생산자 와 소비자를 왜 분리해야 하는지 등등 다양한 문제를 코드로 만들어가며 해결하는 과정을 통해 멀티스레드의 기본기를 학습한 것임
• 이렇게 멀티스레드의 기본기를 잘 쌓아두면 실무에서 복잡한 멀티스레드 상황을 만나도 잘 헤쳐나갈 수 있음

---

5. BlockingQueue - 예제6

1) BlockingQueue 적용

(1) java.util.concurrent.BlockingQueue

package java.util.concurrent;

public interface BlockingQueue<E> extends Queue<E> {
    boolean add(E e);
    boolean offer(E e);
    void put(E e) throws InterruptedException;
    boolean offer(E e, long timeout, TimeUnit unit) throws InterruptedException;
    
    E take() throws InterruptedException;
    E poll(long timeout, TimeUnit unit) throws InterruptedException;
    boolean remove(Object o);
    //...
}

• 자바는 생산자 소비자 문제, 또는 한정된 버퍼라고 불리는 문제를 해결하기 위해 java.util.concurrent.BlockingQueue라는 인터페이스와 구현체들을 제공하며 이름 그대로 스레드를 차단할 수 있는 큐임
• 인터페이스이며 다양한 기능을 제공하지만 주요 메서드만 간략히 정리함
• 데이터 추가 메서드: add(), offer(), put(), offer(타임아웃)
• 데이터 획득 메서드: take(), poll(타임아웃), remove(...)
• Queue를 상속받았기 때문에 큐의 기능들도 사용할 수 있음
• 대표적인 구현체로는 배열 기반으로 구현되어 있고 버퍼의 크기가 고정되어 있는 ArrayBlockingQueue와 링크 기반으로 구현되고 버퍼의 크기를 고정하거나 무한하게 사용할 수 있는 LinkedBlockingQueue가 있음

** 참고

• Deque용 동시성 자료인 BlockingDeque도 있으며 동시성 자료 구조들은 뒤에서 다시 설명함

(2) BoundedQueueV6_1

• BlockingQueue를 사용하도록 코드를 수정
• BlockingQueue.put(data)와 BlockingQueue.take()는 앞서 만들어본 BoundedQueueV5 버전의 put(), take() 메서드와 똑같은 기능을 제공하며, InterruptedException예외를 발생시킴

package thread.bounded;

public class BoundedQueueV6_1 implements BoundedQueue {

    private BlockingQueue<String> queue;

    public BoundedQueueV6_1(int max) {
        queue = new ArrayBlockingQueue<>(max);
    }

    @Override
    public void put(String data) {
        try {
            queue.put(data);
        } catch (InterruptedException e) {
            throw new RuntimeException(e);
        }
    }

    @Override
    public String take() {
        try {
            return queue.take();
        } catch (InterruptedException e) {
            throw new RuntimeException(e);
        }
    }

    @Override
    public String toString() {
        return queue.toString();
    }
}

 

(3) ArrayBlockingQueue.put() 구현 확인

• 주요 코드만 가져와서 확인해 보면 구조가 BoundedQueueV5와 비슷하게 구현되어 있는 것을 확인할 수 있음
• ReentrantLock을 사용하고 생산자 스레드 전용 대기실과 소비자 스레드 전용 대기실도 존재하며 버퍼가 가득 차면 await()로 생산자 스레드가 대기하고, 생산자 스레드가 생산을 완료하면 소비자 스레드에 signal()로 신호를 전달함
• 차이가 있다면 lock() 대신에 인터럽트가 걸릴 수 있도록 lockInterruptibly()를 사용한 점과 내부 자료 구조의 차이 정도임

public class ArrayBlockingQueue<E> {

    final Object[] items;
    int count;
    ReentrantLock lock;
    Condition notEmpty; //소비자 스레드가 대기하는 condition
    Condition notFull; //생산자 스레드가 대기하는 condition
    
    public void put(E e) throws InterruptedException {
        Objects.requireNonNull(e);
        final ReentrantLock lock = this.lock;
        lock.lockInterruptibly();
        try {
            while (count == items.length)
                notFull.await();
            enqueue(e);
        } finally {
            lock.unlock();
        }
    }
    
   private void enqueue(E e) {
        final Object[] items = this.items;
        items[putIndex] = e;
        if (++putIndex == items.length) putIndex = 0;
        count++;
        notEmpty.signal();
    }
}

 

(4) BoundedMain

• BoundedQueueV6_1을 사용하도록 변경하고 실행해 보면 내부에서 모든 로그를 출력하지 않기 때문에 출력되는 로그의 양 차이가 있지만 실행 결과는 V5 버전과 동일하게 동작함

/* 생산자 먼저 실행 결과
15:47:58.093 [     main] == [생산자 먼저 실행] 시작, BoundedQueueV6_1 ==

15:47:58.095 [     main] 생산자 시작
15:47:58.099 [producer1] [생산 시도] data1 -> []
15:47:58.100 [producer1] [생산 완료] data1 -> [data1]
15:47:58.202 [producer2] [생산 시도] data2 -> [data1]
15:47:58.202 [producer2] [생산 완료] data2 -> [data1, data2]
15:47:58.307 [producer3] [생산 시도] data3 -> [data1, data2]

15:47:58.412 [     main] 현재 상태 출력, 큐 데이터: [data1, data2]
15:47:58.412 [     main] producer1: TERMINATED
15:47:58.412 [     main] producer2: TERMINATED
15:47:58.412 [     main] producer3: WAITING

15:47:58.413 [     main] 소비자 시작
15:47:58.413 [consumer1] [소비 시도]     ? <- [data1, data2]
15:47:58.413 [producer3] [생산 완료] data3 -> [data2, data3]
15:47:58.413 [consumer1] [소비 완료] data1 <- [data2]
15:47:58.514 [consumer2] [소비 시도]     ? <- [data2, data3]
15:47:58.515 [consumer2] [소비 완료] data2 <- [data3]
15:47:58.620 [consumer3] [소비 시도]     ? <- [data3]
15:47:58.620 [consumer3] [소비 완료] data3 <- []

15:47:58.725 [     main] 현재 상태 출력, 큐 데이터: []
15:47:58.725 [     main] producer1: TERMINATED
15:47:58.726 [     main] producer2: TERMINATED
15:47:58.726 [     main] producer3: TERMINATED
15:47:58.728 [     main] consumer1: TERMINATED
15:47:58.728 [     main] consumer2: TERMINATED
15:47:58.728 [     main] consumer3: TERMINATED
15:47:58.729 [     main] == [생산자 먼저 실행] 종료, BoundedQueueV6_1 ==
*/

/* 소비자 먼저 실행 결과
15:49:14.809 [     main] == [소비자 먼저 실행] 시작, BoundedQueueV6_1 ==

15:49:14.811 [     main] 소비자 시작
15:49:14.813 [consumer1] [소비 시도]     ? <- []
15:49:14.918 [consumer2] [소비 시도]     ? <- []
15:49:15.023 [consumer3] [소비 시도]     ? <- []

15:49:15.128 [     main] 현재 상태 출력, 큐 데이터: []
15:49:15.128 [     main] consumer1: WAITING
15:49:15.129 [     main] consumer2: WAITING
15:49:15.129 [     main] consumer3: WAITING

15:49:15.129 [     main] 생산자 시작
15:49:15.130 [producer1] [생산 시도] data1 -> []
15:49:15.130 [producer1] [생산 완료] data1 -> [data1]
15:49:15.130 [consumer1] [소비 완료] data1 <- []
15:49:15.235 [producer2] [생산 시도] data2 -> []
15:49:15.235 [producer2] [생산 완료] data2 -> [data2]
15:49:15.235 [consumer2] [소비 완료] data2 <- []
15:49:15.336 [producer3] [생산 시도] data3 -> []
15:49:15.336 [producer3] [생산 완료] data3 -> [data3]
15:49:15.336 [consumer3] [소비 완료] data3 <- []

15:49:15.441 [     main] 현재 상태 출력, 큐 데이터: []
15:49:15.441 [     main] consumer1: TERMINATED
15:49:15.441 [     main] consumer2: TERMINATED
15:49:15.441 [     main] consumer3: TERMINATED
15:49:15.442 [     main] producer1: TERMINATED
15:49:15.442 [     main] producer2: TERMINATED
15:49:15.442 [     main] producer3: TERMINATED
15:49:15.442 [     main] == [소비자 먼저 실행] 종료, BoundedQueueV6_1 ==
*/

2) 기능 설명

(1) 기능 설명

• 실무에서 멀티스레드를 사용할 때는 응답성이 중요함
• 예를 들어 대기 상태에 있어도 고객이 중지 요청을 하거나 너무 오래 대기한 경우 포기하고 빠져나갈 수 있는 방법이 필요함
• 생산자가 무언가 데이터를 생산하는데 버퍼가 빠지지 않아서 너무 오래 대기해야 한다면 무한정 기다리는 것보다는 작업을 포기하고 고객에게 메시지를 전달하는 것이 더 나은 선택일 수 있음

(2) 예시 상황

• 서버에 상품을 주문하는 고객을 생산자로보고 고객이 상품을 주문하면 고객의 요청을 생산자 스레드가 받아서 중간에 있는 큐에 넣어주며 소비자 스레드는 큐에서 주문 요청을 꺼내서 주문을 처리하는 스레드라고 가정
• 큐의 한계가 1000개이고 소비자 스레드는 한 번에 10개 정도의 주문만 처리할 수 있다고 가정하면 선착순 이벤트와 같은 이슈로 갑자기 주문이 폭주를 하게 되면 생산자 스레드는 순간적으로 1000개가 넘는 주문이 큐에 담지만 소비자 스레드는 10개밖에 처리를 못하므로 소비가 생산을 따라가지 못하고 큐가 가득 차게 됨
• 이런 상황이 되면 수많은 생산자 스레드는 큐 앞에 대기하게 되고 결국 고객이 응답을 받지 못하고 무한 대기 하게 됨
• 고객 입장에서는 무작정 무한 대기하고 결과도 알 수 없는 상황이 지속되기보다는 너무 오래 대기한 경우에 데이터 추가를 포기하고 고객에게 관련된 메시지를 보여주는 것이 더 나은 선택일 것임

(3) 큐가 가득 찼을 때 선택할 수 있는 상황

• 예외를 던지고 예외를 받아서 처리
• 대기하지 않고 즉시 false 반환
• 대기
• 특정 시간만큼만 대기

(4) BlockingQueue의 다양한 기능 - 공식 API 문서

Operation	Throws Exception	Special Value	Blocks	Times Out
Insert(추가)	add(e)	offer(e)	put(e)	offer(e, time, unit)
Remove(제거)	remove()	poll()	take()	poll(time, unit)
Examine(관찰)	element()	peek()	not applicable	not applicable

• 이런 문제를 해결하기 위해 BlockingQueue는 각 상황에 맞는 다양한 메서드를 제공하며 대기, 시간 대기 메서드는 인터럽트를 제공함
• Throws Exception - 대기 시 예외
  • add(e): 지정된 요소를 큐에 추가하며 큐가 가득 차면 IllegalStateException 예외를 던짐
  • remove(): 큐에서 요소를 제거하면 반환하며 큐가 비어있으면 NoSuchElementException 예외를 던짐
  • element(): 큐의 머리 요소를 반환하지만 요소를 큐에서 제거하지 않고 큐가 비어있으면 NoSuchElementException 예외를 던짐
• Special Value - 대기 시 즉시 반환
  • offer(e): 지정된 요소를 큐에 추가하려고 시도하면 큐가 가득 차면 false를 반환
  • poll(): 큐에서 요소를 제거하고 반환하며 큐가 비어 있으면 null을 반환
  • peek(): 큐의 머리 요소를 반환하지만 요소를 큐에서 제거하지 않고 큐가 비어 있으면 null을 반환
• Blocks - 대기
  • put(e): 지정된 요소를 큐에 추가할 때까지 대기함, 큐가 가득 차면 공간이 생길 때까지 대기함
  • take(): 큐에서 요소를 제거하고 반환함, 큐가 비어 있으면 요소가 준비될 때까지 대기함
• Thimes Out - 시간 대기
  • odff(e, time, unit): 지정된 요소를 큐에 추가하려고 시도하며 지정된 시간 동안 큐가 비워지리를 기다리다가 시간이 초과되면 false를 반환함
  • poll(time, unit): 큐에서 요소를 제거하고 반환하며 큐에 요소가 없다면 지정된 시간 동안 요소가 준비되기를 기다리다가 시간이 초과되면 null을 반환함

3) 기능 확인

(1-1) BoundedQueueV6_2 - 즉시 반환

• BlockingQueue의 offer(data), poll()을 사용하여 스레드를 대기하지 않고 즉시 반환
• put()에서는 offer(data)를 사용하여 데이터 추가가 성공하면 true를 반환하고 버퍼가 가득 차면 즉시 false를 반환함
• take()에서는 poll()을 사용하여 버퍼에 데이터가 없으면 즉시 null을 반환함

package thread.bounded;

public class BoundedQueueV6_2 implements BoundedQueue {

    // ... 코드 동일 생략

    @Override
    public void put(String data) {
        boolean result = queue.offer(data);
        log("저장 시도 결과 = " + result);
    }

    @Override
    public String take() {
        return queue.poll();
    }
    
    // ... 코드 동일 생략
}

 

(1-2) BoundedMain

• BoundedQueueV6_2를 사용하도록 변경하고 코드를 실행해 보면 BoundedQueueV1에서 실행했던 결과와 동일한 결과를 얻는 것을 확인할 수 있음
• 생산자 먼저 실행을 하면 data3이 그냥 버려지고 마지막 소비자 스레드는 null을 반환받게 되고, 소비자 먼저 실행하면 소비자가 모두 null을 반환받고 종료하며 데이터 큐에 data1, data2가 쌓이고 data3은 버려짐

(2-1) BlockingQueue6_3 - 시간 대기

• put()에 offer(data, 시간)을 사용하고 take()에는 poll(시간)을 사용하여 특정 시간만큼 대기하도록 설정
• offer()는 확인을 목적으로 1 나노초만 대기하도록 설정하였고 대기 시간이 지나면 false를 반환함
• poll()은 2초 대기하도록 설정하였고 대기 시간이 지나면 null을 반환함

package thread.bounded;

public class BoundedQueueV6_3 implements BoundedQueue {

    // ... 코드 동일 생략

    @Override
    public void put(String data) {
        try {
            boolean result = queue.offer(data, 1, TimeUnit.NANOSECONDS);
            log("저장 시도 결과 = " + result);
        } catch (InterruptedException e) {
            throw new RuntimeException(e);
        }
    }
    
    @Override
    public String take() {
        try {
            return queue.poll(2, TimeUnit.SECONDS);
        } catch (InterruptedException e) {
            throw new RuntimeException(e);
        }
    }
    // ... 코드 동일 생략
}

(2-2) BoundedMain

• BoundedQueueV6_3를 사용하도록 변경하고 코드를 실행
• 소비자 먼저 실행하도록 하면 생성된 소비자 스레드들은 큐에 데이터가 없으므로 2초간 대기하게 되고 그 이후에 생산자 스레드가 실행되어 데이터가 생성되자마자 소비가 이루어져 깔끔하게 정상 처리가 됨
• 생산자 먼저 실행하도록 하면 생산자 스레드는 큐가 가득 차면 1 나노초만 기다리게 세팅했으므로 data1, data2는 큐에 데이터가 저장되지만 data3은 저장 시도 결과가 false로 나오며 실패하게 되고 소비자 스레드는 data1, data2는 정상적으로 반환하지만 마지막 소비자 스레드는 데이터가 없으므로 2초간 기다리다가 결국 null을 반환함

/* 소비자 스레드 먼저 시작
16:40:54.080 [     main] == [소비자 먼저 실행] 시작, BoundedQueueV6_3 ==

16:40:54.081 [     main] 소비자 시작
16:40:54.084 [consumer1] [소비 시도]     ? <- []
16:40:54.189 [consumer2] [소비 시도]     ? <- []
16:40:54.294 [consumer3] [소비 시도]     ? <- []

16:40:54.399 [     main] 현재 상태 출력, 큐 데이터: []
16:40:54.399 [     main] consumer1: TIMED_WAITING
16:40:54.400 [     main] consumer2: TIMED_WAITING
16:40:54.400 [     main] consumer3: TIMED_WAITING

16:40:54.400 [     main] 생산자 시작
16:40:54.400 [producer1] [생산 시도] data1 -> []
16:40:54.401 [consumer1] [소비 완료] data1 <- []
16:40:54.401 [producer1] 저장 시도 결과 = true
16:40:54.401 [producer1] [생산 완료] data1 -> []
16:40:54.502 [producer2] [생산 시도] data2 -> []
16:40:54.503 [producer2] 저장 시도 결과 = true
16:40:54.503 [consumer2] [소비 완료] data2 <- []
16:40:54.503 [producer2] [생산 완료] data2 -> []
16:40:54.607 [producer3] [생산 시도] data3 -> []
16:40:54.608 [producer3] 저장 시도 결과 = true
16:40:54.608 [consumer3] [소비 완료] data3 <- []
16:40:54.608 [producer3] [생산 완료] data3 -> []

16:40:54.713 [     main] 현재 상태 출력, 큐 데이터: []
16:40:54.713 [     main] consumer1: TERMINATED
16:40:54.713 [     main] consumer2: TERMINATED
16:40:54.713 [     main] consumer3: TERMINATED
16:40:54.714 [     main] producer1: TERMINATED
16:40:54.714 [     main] producer2: TERMINATED
16:40:54.714 [     main] producer3: TERMINATED
16:40:54.715 [     main] == [소비자 먼저 실행] 종료, BoundedQueueV6_3 ==
*/

/* 생산자 스레드 먼저 시작
16:39:07.583 [     main] == [생산자 먼저 실행] 시작, BoundedQueueV6_3 ==

16:39:07.584 [     main] 생산자 시작
16:39:07.589 [producer1] [생산 시도] data1 -> []
16:39:07.590 [producer1] 저장 시도 결과 = true
16:39:07.590 [producer1] [생산 완료] data1 -> [data1]
16:39:07.692 [producer2] [생산 시도] data2 -> [data1]
16:39:07.692 [producer2] 저장 시도 결과 = true
16:39:07.692 [producer2] [생산 완료] data2 -> [data1, data2]
16:39:07.796 [producer3] [생산 시도] data3 -> [data1, data2]
16:39:07.796 [producer3] 저장 시도 결과 = false
16:39:07.797 [producer3] [생산 완료] data3 -> [data1, data2]

16:39:07.898 [     main] 현재 상태 출력, 큐 데이터: [data1, data2]
16:39:07.898 [     main] producer1: TERMINATED
16:39:07.899 [     main] producer2: TERMINATED
16:39:07.899 [     main] producer3: TERMINATED

16:39:07.899 [     main] 소비자 시작
16:39:07.899 [consumer1] [소비 시도]     ? <- [data1, data2]
16:39:07.900 [consumer1] [소비 완료] data1 <- [data2]
16:39:08.004 [consumer2] [소비 시도]     ? <- [data2]
16:39:08.005 [consumer2] [소비 완료] data2 <- []
16:39:08.108 [consumer3] [소비 시도]     ? <- []

16:39:08.214 [     main] 현재 상태 출력, 큐 데이터: []
16:39:08.214 [     main] producer1: TERMINATED
16:39:08.214 [     main] producer2: TERMINATED
16:39:08.214 [     main] producer3: TERMINATED
16:39:08.214 [     main] consumer1: TERMINATED
16:39:08.215 [     main] consumer2: TERMINATED
16:39:08.215 [     main] consumer3: TIMED_WAITING
16:39:08.215 [     main] == [생산자 먼저 실행] 종료, BoundedQueueV6_3 ==
16:39:10.114 [consumer3] [소비 완료] null <- []
*/

 

(3-1) BlockingQueueV6_4 - 예외

• add(data)와 remove()를 사용하도록 변경
• add(data)는 버퍼가 가득차면 예외가 즉시 발생하고 remove()는 버퍼에 데이터가 없으면 즉시 예외가 발생함

package thread.bounded;

import static util.MyLogger.log;

public class BoundedQueueV6_4 implements BoundedQueue {

    // ... 기존 코드 동일 생략
    
    @Override
    public void put(String data) {
        queue.add(data);
    }

    @Override
    public String take() {
        return queue.remove();
    }

    // ... 기존 코드 동일 생략
    
}

 

(3-2) BoundedMain

• BoundedQueueV6_4를 사용하도록 변경하고 코드를 실행해 보면 생산자 스레드가 큐에 데이터가 가득 차거나, 소비자 스레드가 빈 스레드에서 데이터를 꺼낼 때 즉시 예외가 터지는 것을 확인할 수 있음

/* 생산자 먼저 실행 결과
16:45:34.213 [     main] == [생산자 먼저 실행] 시작, BoundedQueueV6_4 ==

16:45:34.214 [     main] 생산자 시작
16:45:34.219 [producer1] [생산 시도] data1 -> []
16:45:34.220 [producer1] [생산 완료] data1 -> [data1]
16:45:34.322 [producer2] [생산 시도] data2 -> [data1]
16:45:34.322 [producer2] [생산 완료] data2 -> [data1, data2]
16:45:34.425 [producer3] [생산 시도] data3 -> [data1, data2]
Exception in thread "producer3" java.lang.IllegalStateException: Queue full
	at java.base/java.util.AbstractQueue.add(AbstractQueue.java:98)
	at java.base/java.util.concurrent.ArrayBlockingQueue.add(ArrayBlockingQueue.java:329)
	at thread.bounded.BoundedQueueV6_4.put(BoundedQueueV6_4.java:19)
	at thread.bounded.ProducerTask.run(ProducerTask.java:18)
	at java.base/java.lang.Thread.run(Thread.java:1583)

16:45:34.530 [     main] 현재 상태 출력, 큐 데이터: [data1, data2]
16:45:34.531 [     main] producer1: TERMINATED
16:45:34.531 [     main] producer2: TERMINATED
16:45:34.531 [     main] producer3: TERMINATED

16:45:34.531 [     main] 소비자 시작
16:45:34.532 [consumer1] [소비 시도]     ? <- [data1, data2]
16:45:34.532 [consumer1] [소비 완료] data1 <- [data2]
16:45:34.637 [consumer2] [소비 시도]     ? <- [data2]
16:45:34.637 [consumer2] [소비 완료] data2 <- []
16:45:34.742 [consumer3] [소비 시도]     ? <- []
Exception in thread "consumer3" java.util.NoSuchElementException
	at java.base/java.util.AbstractQueue.remove(AbstractQueue.java:117)
	at thread.bounded.BoundedQueueV6_4.take(BoundedQueueV6_4.java:24)
	at thread.bounded.ConsumerTask.run(ConsumerTask.java:16)
	at java.base/java.lang.Thread.run(Thread.java:1583)

16:45:34.846 [     main] 현재 상태 출력, 큐 데이터: []
16:45:34.846 [     main] producer1: TERMINATED
16:45:34.846 [     main] producer2: TERMINATED
16:45:34.847 [     main] producer3: TERMINATED
16:45:34.847 [     main] consumer1: TERMINATED
16:45:34.847 [     main] consumer2: TERMINATED
16:45:34.847 [     main] consumer3: TERMINATED
16:45:34.847 [     main] == [생산자 먼저 실행] 종료, BoundedQueueV6_4 ==
*/

 

(4) BoundedQueue 제거

• 사실 여기서 BoundedQueue는 단순히 위임만 하기 때문에 앞서 우리가 만든 BoundedQueue 인터페이스를 제거하고 BlockingQueue를 직접 사용해도 됨
• 대신 BoundedQueue를 사용하는 모든 코드를 BlckingQueue를 사용하도록 변경하면 됨
• BoundedMain, ConsumerTask, ProducerTask 등의 코드를 변경하면 됨

package thread.bounded;

public class ConsumerTask2 implements Runnable {

    private BlockingQueue<String> queue;

    public ConsumerTask2(BlockingQueue<String> queue) {
        this.queue = queue;
    }

    @Override
    public void run() {
        log("[소비 시도]     ? <- " + queue);
        String data = null;
        try {
            data = queue.take();
        } catch (InterruptedException e) {
            throw new RuntimeException(e);
        }
        log("[소비 완료] " + data + " <- " + queue);


    }
}

package thread.bounded;

public class ProducerTask2 implements Runnable {

    private BlockingQueue<String> queue;
    private String request;

    public ProducerTask2(BlockingQueue<String> queue, String request) {
        this.queue = queue;
        this.request = request;
    }

    @Override
    public void run() {
        log("[생산 시도] " + request + " -> " + queue);
        try {
            queue.put(request);
        } catch (InterruptedException e) {
            throw new RuntimeException(e);
        }
        log("[생산 완료] " + request + " -> " + queue);
    }
}

package thread.bounded;

public class BlockingMain {
    public static void main(String[] args) {

        BlockingQueue<String> queue = new ArrayBlockingQueue<>(2);

        // 2. 생산자, 소비자 실행 순서 선택, 반드시 하나만 선택
//        producerFirst(queue);   // 생산자 먼저 실행
        consumerFirst(queue);   // 소비자 먼저 실행
    }

    private static void consumerFirst(BlockingQueue<String> queue) {
        // ... 기존 코드 동일 생략
    }

    private static void producerFirst(BlockingQueue<String> queue) {
        // ... 기존 코드 동일 생략
    }

    private static void startProducer(BlockingQueue<String> queue, List<Thread> threads) {
        // ... 기존 코드 동일 생략
    }

    private static void startConsumer(BlockingQueue<String> queue, List<Thread> threads) {
        // ... 기존 코드 동일 생략
    }

    private static void printAllState(BlockingQueue<String> queue, List<Thread> threads) {
        // ... 기존 코드 동일 생략
    }
}

 

(5) Doug Lea

• 자바 1.5에 추가된 java.util.concurrent 패키지가 제공하는 Lock, ReentrantLock, Condition, BlockingQueue 등은 매우 잘 만들어진 라이브러리인데 코드들을 열어보면 코드 작성자에 Doug Lea라는 이름이 적혀 있음
• Doug Lea는 컴퓨터 과학 교수로 동시성 프로그래밍, 멀티스레딩, 병렬 컴퓨팅, 알고리즘 및 데이터 구조 등의 분야에서 많은 업적을 만들었음
• 특히 자바 커뮤니티 프로세스(JCP)의 일원으로 활동하면서 JSR-166이라는 자바 java.util.concurrent 패키지의 주요 설계 및 구현을 주도했음(혼자 만든 것은 아니고 대표자 이름이 적힘)
• java.util.concurrent 패키지가 제공하는 동시성 라이브러리는 매우 견고하고 높은 성능을 낼 수 있도록 최적화되어있으며 다양한 동시성 시나리오를 대응할 수 있으며 개발자가 쉽고 편리하게 복잡한 동시성 문제를 다룰 수 있게 해 줌
• 그는 동시성 프로그래밍의 복잡한 개념들을 실용적이고 효율적인 구현으로 변환하는데 큰 역할을 했음
• 결론적으로 Doug Lea의 java.util.concurrent 패키지에 대한 기여는 자바의 동시성 프로그래밍을 크게 발전시키고 이는 현대 자바 프로그래밍의 핵심적인 부분이 됨
• 대용량 트래픽 대응하려면 스레드를 잘 다뤄야 하는데 자바는 이렇게 잘 만들어진 라이브러리를 제공하고 있음
• 수많은 자바 동시성 라이브러리들 뿐만 아니라 Queue, Deque 같은 자료 구조의 코드를 열어보면 Doug Lea의 이름을 발견할 수 있음