[Java] SynchronousQueue

SynchronousQueue, java.util.concurrent.SynchronousQueue, 직접 핸드오프 큐

정의

java.util.concurrent.SynchronousQueue<E> 는 용량 0 의 BlockingQueue. 저장 공간이 없다. put 은 매칭되는 take 가 있어야 진행하고, 반대도 마찬가지.

직접 핸드오프 (direct handoff) 패턴, producer 와 consumer 가 서로를 정확히 한 명씩 기다린다.

시각화

특수한 동작

SynchronousQueue<String> q = new SynchronousQueue<>();
q.size();           // 항상 0
q.isEmpty();        // 항상 true
q.peek();           // 항상 null
q.iterator().hasNext();   // 항상 false

q.put("x");         // take 자가 없으면 block
q.take();           // put 자가 없으면 block

q.offer("x");       // take 자가 즉시 없으면 false 반환
q.poll();           // put 자가 즉시 없으면 null 반환

사용 사례

Executors.newCachedThreadPool() 의 내부 큐

ExecutorService pool = Executors.newCachedThreadPool();

내부적으로 SynchronousQueue 를 사용. 새 작업이 오면 idle thread 가 있으면 즉시 전달, 없으면 새 thread 생성. 큐에 작업이 쌓이지 않는다.

Producer 가 Consumer 도착을 보장받아야 할 때

SynchronousQueue<Order> q = new SynchronousQueue<>();

// Producer
q.put(order);   // consumer 가 이 order 를 받는 것을 확인하고 진행

// Consumer
Order o = q.take();
process(o);

저장 공간이 없으므로 producer 입장에서 “내 데이터가 처리됐다” 는 시점이 명확. 다만 throughput 은 producer/consumer 가 항상 비슷한 속도여야만 좋다.

공정성 옵션

SynchronousQueue<String> fair = new SynchronousQueue<>(true);   // FIFO
SynchronousQueue<String> unfair = new SynchronousQueue<>();      // LIFO 기본

true 면 가장 오래 기다린 스레드부터 매칭.

참고

• BlockingQueue
• LinkedBlockingQueue
• Blocking