QUEUE와 DELAYED 프로세싱

J2SE 5.0 출시에는 새로운 top-level Queue 인터페이스가 Collections Framework에 추가되어 Map, List, Set 인터페이스와 어우러진다. 일반적으로 queue는 먼저 들어온 것이 먼저 나가는 데이터 스트럭쳐이지만 priority queue같은 몇몇 구현들에서는 queue 뒤에 엘리먼트가 첨부되지 않는다. 이 queue와 FIFO(first in, first out)구조에 관한 얘기는 은행에서의 대기선에 비유할 수 있겠다. 은행직원은 고객이 대기선에 있는지 확인한 후 대기선의 첫번째 손님을 맞이하여 손님이 원하는 거래를 처리한다. 그리고 그 손님은 이제 대기선에서 제외된다.

J2SE 5.0에는 Queue 인터페이스와 함께 몇 가지 새로운 queue 구현이 있다. DelayQueue가 그 중 하나인데, DelayQueue에서 queue 안의 아이템들은 지연시간 동안 처리되지 않는다. 이번 테크팁에서는 새로운 Queue 인터페이스와 DelayQueue 구현에 대해 알아보도록 하자.

첫번째로 Queue 인터페이스를 분석해보자. 이 인터페이스는 Collection을 확장하고 다섯개의 고유 메서드들을 추가한다.

• E element()
• boolean offer(E o)
• E peek()
• E poll()
• E remove()

J2SE 5.0이 새롭게 generics를 지원하므로 여기에서의 E는 어느 타입이든지 가능하며, Queue가 생성될 때 정의된 엘리먼트 타입으로 결정된다.

Queue의 엘리먼트를 추가하고 제거하는 데 당연히 Collection 인터페이스 메서드들을 사용할 수도 있지만, 그 메서드들은 동작할 때 부가적인 요구사항을 가지므로, Queue 인터페이스에서는 사용하지 않을 것을 권장하고 있다. 예를 들어 Collection.add 메서드로 queue에 엘리먼트를 추가하는 대신에 offer 메서드로 queue에 엘리먼트를 추가할 수 있다. 이 둘은 무슨 차이가 있을까? add 실행이 오류가 날 수 있다. 그 한 예는 queue에 사이즈 제한이 있을 때이다.(은행 대기선에 비유하자면, 10명만 대기 가능할 경우) Collection의 add 메서드를 사용하면 add는 예외를 던지면서 실패한다. 이와 비교할 때 offer 메서드는 false를 리턴하며 “실패”하게 된다. 따라서 offer 메서드를 사용하면 실제로 예외적인 상황에서만(특히 체크 안 된 런타임 예외가 던져졌을 경우) 예외처리(exception handling)을 사용하게 된다.

Queue의 다른 네가지 메서드는 두개씩 짝지어 설명할 수 있다. remove/poll, element/peek. remove와 poll 메서드는 둘 다 queue의 첫번째 엘리먼트 즉 “head”를 제거하는 데 사용된다. 빈 Collection 객체에서 호출되었을 때, remove 메서드는 예외를 던지고, poll 메서드는 단순히 null 값을 리턴한다. head 엘리먼트를 제거하는 대신 단지 그 엘리먼트를 살펴볼 수도 있다. 이 때 element와 peek 메서드가 사용된다. 여기서 element 메서드는 빈 queue에서는 예외를 던지고, peek은 null값을 리턴한다. Queue는 일반적으로 태스크를 진행하는 데 사용되므로 빈 queue를 갖는 것이 예외상황일 필요는 없다. 따라서, poll/peek 모델이 사용하기에 보다 적합할 것이다. (앞에서 본 메서드들 중 예외를 안 던지고 null 리턴하는 메서드들)

Queue는 다음과 같은 상황에서 일반적으로 사용된다.

   class Producer implements Runnable {n     private final Queue queue;n     Producer(Queue q) { queue = q; }n     public void run() {n       try {n         while(true) { queue.offer(produce()); }n       } catch (InterruptedException ex) { … handle …}n     }n     Object produce() { … }n   }nn   class Consumer implements Runnable {n     private final Queue queue;n     Consumer(Queue q) { queue = q; }n     public void run() {n       try {n         Object o;n         while((o = queue.poll()) != null) { consume(o); }n       } catch (InterruptedException ex) { … handle …}n     }n     void consume(Object x) { … }n   }n

Queue가 꽉 차 있거나(생산자의 입장에서) 비어있을 때(소비자의 입장에서) 어떤 일이 일어나는 지 궁금할 것이다. 이 시점에서 새로운 queue 인터페이스 BlockingQueue를 설명하는 것이 좋겠다. Queue를 사용하여 엘리먼트를 무작정 추가하거나(offer사용) 삭제하는(poll 사용) 대신 BlockingQueue의 put 메서드로 엘리먼트를 추가하고 take메서드로 제거할 수 있다. put과 take는 모두 이를 호출한 쓰레드가, 특정 조건 하에서 블로킹되게끔 한다. put은 queue가 꽉 차 있을 경우, take는 queue가 비어있을 경우가 블로킹 조건이다.

BlockingQueue의 일반적인 사용패턴은 다음과 같다.

   class Producer implements Runnable {n     private final BlockingQueue queue;n     Producer(BlockingQueue q) { queue = q; }n     public void run() {n       try {n         while(true) { queue.put(produce()); }n       } catch (InterruptedException ex) { … handle …}n     }n       Object produce() { … }n  }n  n   class Consumer implements Runnable {n     private final BlockingQueue queue;n     Consumer(BlockingQueue q) { queue = q; }n     public void run() {n       try {n         while(true) { consume(queue.take()); }n       } catch (InterruptedException ex) { … handle …}n     }n     void consume(Object x) { … }n   }n

각 생성된 생산자가 꽉 찬 queue에 새로운 아이템을 추가하려 하면 put메서드에서 기다리고, take 메서드에서는 꺼내갈 것이 추가될 때까지 기다린다. queue가 비어있다면 while(true) 조건문에는 아무런 변화도 일어나지 않을 것이다.

DelayQueue는 BlockingQueue 인터페이스의 구체적인 구현이다. DelayQueue 에 추가된 아이템들은 반드시 새로운 Delayed 인터페이스를 구현해야하며, 이 Delayed는 한 개의 메서드, long getDelay(TimeUnit unit)를 갖고 있다. DelayQueue는 우선순위 힙 데이터 스트럭쳐에 기반한 시간 기준 스케쥴링 Queue로 동작한다

데모를 위해, 다음의 프로그램 DelayTest는 몇 초안에 실행되는 Delayed 인터페이스를 구현한다. 알아둬야할 것은1) nanosecond는 10억분의 1초, 2) nanosecond 유니트에서 작업을 가능케하는 System의 새로운 메서드, nanoTime 이 있다는 것이다. getDelay 메서드가 nanosecond로 리턴된 횟수를 필요로 하기 때문에 nanosecond에서 작업하는 것이 중요하다.

   import java.util.Random;n   import java.util.concurrent.Delayed;n   import java.util.concurrent.DelayQueue;n   import java.util.concurrent.TimeUnit;nn   public class DelayTest {n     public static long BILLION = 1000000000;n     static class SecondsDelayed implements Delayed { n       long trigger;n       String name;n       SecondsDelayed(String name, long i) { n         this.name = name;n         trigger = System.nanoTime() + (i * BILLION);n       }n       public int compareTo(Delayed d) {n         long i = trigger;n         long j = ((SecondsDelayed)d).trigger;n         int retuValue;n         if (i < j) {n           retuValue = -1;n         } else if (i > j) {n           retuValue = 1;n         } else {n           retuValue = 0;n         }n         retu retuValue;n       }n       public boolean equals(Object other) {n         retu ((SecondsDelayed)other).trigger == trigger;n       }n       public long getDelay(TimeUnit unit) {n         long n = trigger – System.nanoTime();n         retu unit.convert(n, TimeUnit.NANOSECONDS);n       }n       public long getTriggerTime() {n         retu trigger;n       }n       public String getName() {n         retu name;n       }n       public String toString() {n         retu name + ” / ” + String.valueOf(trigger);n       }n     }n     public static void main(String args[]) n             throws InterruptedException {n       Random random = new Random();n       DelayQueue<SecondsDelayed> queue = n             new DelayQueue<SecondsDelayed>();n       for (int i=0; i < 10; i++) {n         int delay = random.nextInt(10);n         System.out.println(“Delaying: ” + n               delay + ” for loop ” + i);n         queue.add(new SecondsDelayed(“loop ” + i, delay));n       }n       long last = 0;n       for (int i=0; i < 10; i++) {n         SecondsDelayed delay = (SecondsDelayed)(queue.take());n         String name = delay.getName();n         long tt = delay.getTriggerTime();n         if (i != 0) {n           System.out.println(“Delta: ” + n                 (tt – last) / (double)BILLION);n         }n         System.out.println(name + ” / Trigger time: ” + tt);n         last = tt;n       }n     }n   }n

DelayTest 프로그램은 엘리먼트를 실행하기 전 DelayQueue 안에 그 10개의 엘리먼트를 위치시킨다.

다음은 DelayQueue를 한번 구동했을 때의 출력물이다.

   Delaying: 8 for loop 0n   Delaying: 7 for loop 1n   Delaying: 2 for loop 2n   Delaying: 4 for loop 3n   Delaying: 0 for loop 4n   Delaying: 9 for loop 5n   Delaying: 3 for loop 6n   Delaying: 4 for loop 7n   Delaying: 6 for loop 8n   Delaying: 2 for loop 9n   loop 4 / Trigger time: 1883173869520000n   Delta: 1.9995545n   loop 2 / Trigger time: 1883175869074500n   Delta: 0.0012475n   loop 9 / Trigger time: 1883175870322000n   Delta: 0.9995177n   loop 6 / Trigger time: 1883176869839700n   Delta: 0.9995187n   loop 3 / Trigger time: 1883177869358400n   Delta: 6.408E-4n   loop 7 / Trigger time: 1883177869999200n   Delta: 2.0001667n   loop 8 / Trigger time: 1883179870165900n   Delta: 0.9986953n   loop 1 / Trigger time: 1883180868861200n   Delta: 0.9995595n   loop 0 / Trigger time: 1883181868420700n   Delta: 1.001262n   loop 5 / Trigger time: 1883182869682700n

이 출력물은 loop4를 위한 아이템이 time 0, 즉 지연 없이 시작하기로 설정되어있다는 것을 가리키고 있으며,

   Delaying: 0 for loop 4n

따라서 첫번째로 다음과 같이 구동한다.

   loop 4 / Trigger time: 1883173869520000n

Loop 2에서는 2초간의 지연이 있으며,

   Delaying: 2 for loop 2n

따라서 다음과 같이 나타난다.

   loop 2 / Trigger time: 1883175869074500n

여기서의 delta는 1.9995545로 약 2초이다.

   Delta: 1.9995545n

다른 loop를 위한 비슷한 delta들도 존재한다.

좀 더 실제적인 예를 위해 DelayQueue에서 pull된 것을 단지 출력하는 대신, queue의 아이템들을 Runnable 구현시키고, 그 아이템들의 run 메서드를 호출할 수 있다.

Queue, DelayQueue 와 J2SE 5.0에서 바뀐 다른 Collections Framework에 대한 좀 더 많은 정보를 원한다면 Collection Framework Enhancements를 참조하기 바란다.

@http://kr.sun.com/developers/techtips/c2004_1019.html