java - Executor

Java Concurrency: Executor와 Callable/Future

Java 2008/02/05 05:47

자바 5부터 새롭게 추가된 Concurrency API 중에서 Executor, 리턴이 가능한 Callable 및 Future에 대해서 살펴본다.

Executor를 이용한 쓰레드 관리

웹 서버와 같이 동시에 다수의 요청을 처리해야 하는 어플리케이션을 개발해야 할 경우 코드는 다음과 같은 형태를 띌 것이다.

while(true) {
    request = acceptRequest();
    Runnable requestHandler = new RequestHandler(request);
    new Thread(requestHandler).start();
}

위 코드가 논리적으로 문제점은 없지만, 다음과 같은 성능상의 문제점을 안고 있다.

• 소규모의 많은 요청이 들어올 경우 쓰레드 생성 및 종료에 따른 오버헤드가 발생한다.
• 생성되는 쓰레드 개수에 제한이 없기 때문에 OutOfMemoryError가 발생할 수 있다.
• 많은 수의 쓰레드가 실행될 경우, 쓰레드 스케줄링에 따른 오버헤드가 발생한다.

이런 문제점 때문에 동시에 다수의 요청을 처리해야 하는 어플리케이션에서는 쓰레드 풀을 사용하여 동시에 실행될 수 있는 쓰레드의 개수를 제한하는 것이 일반적이다.

자바5부터 새롭게 추가된 Concurrency API는 작업을 실행하기 위한 Executor 인터페이스를 제공하고 있으며, 쓰레드 풀, 큐 등 다양한 Executor 구현체를 제공하고 있다. 따라서, 앞서 Thread를 직접 사용할 때의 문제점을 해소할 수 있게 되었다.

Executor를 이용한 쓰레드 관리

Executor와 관련된 주요 API는 다음과 같다.

위 클래스 다이어그램에서 각 인터페이스는 다음과 같은 기능을 정의한다.

• Executor 인터페이스:
  제공된 작업(Runnable 구현체)을 실행하는 객체가 구현해야 할 인터페이스. 이 인터페이스는 작업을 제공하는 코드와 작업을 실행하는 메커니즘의 사이의 커플링을 제거해준다.
• ExecutorService 인터페이스:
  Executor의 라이프사이클을 관리할 수 있는 기능을 정의하고 있다. Runnable 뿐만 아니라 Callable을 작업으로 사용할 수 있는 메소드가 추가로 제공된다.
• ScheduledExecutorService:
  지정한 스케쥴에 따라 작업을 수행할 수 있는 기능이 추가되었다.

Executor 인터페이스

java.util.concurrent.Executor 인터페이스는 작업을 실행하는 클래스가 구현해야 할 인터페이스로서, 다음과 같이 정의되어 있다.

public interface Executor {
    void execute(Runnable command);
}

Executor 인터페이스의 구현체는 execute() 메소드로 전달받은 작업(Runnable 인스턴스)을 알맞게 실행하게 된다. 예를 들어, 쓰레드 풀을 구현한 Executor 구현체는 전달받은 작업을 큐에 넣은 뒤 가용한 쓰레드가 존재할 경우, 해당 쓰레드에 작업을 실행하도록 구현될 것이다.

아래 코드는 앞서 Thread를 사용했던 코드를 Executor를 사용하는 코드로 변환한 것이다.

Executor executor = …; // Executor 구현체 생성
while(true) {
    request = acceptRequest();
    Runnable requestHandler = new RequestHandler(request);
    executor.execute(requestHandler);
}

위 코드는 작업을 생성만 할 뿐 실제로 작업을 실행하지는 않는다. 단지, Executor.execute() 메소드에 생성한 작업을 전달할 뿐이다. 작업을 실제로 실행하는 책임은 Executor에 있으며, 이제 작업을 생성하는 코드에서는 작업이 어떻게 실행되는 지의 여부는 알 필요가 없다. 즉, Executor를 사용함으로써 작업을 생성하는 코드와 작업을 실행하는 메커니즘 사이의 커플링을 없앤 것이다.

ExecutorService 인터페이스

ExecutorService 인터페이스는 다음과 같이 두 가지 종류의 메소드가 추가되었다.

• Executor의 라이프 사이클을 관리
• Callable을 작업으로 사용하기 위한 메소드

먼저, 라이프 사이클과 관련된 메소드는 다음과 같다.

• void shutdown():
  셧다운 한다. 이미 Executor에 제공된 작업은 실행되지만, 새로운 작업은 수용하지 않는다.
• List<Runnable> shutdownNow():
  현재 실행중인 모든 작업을 중지시키고, 대기중인 작업을 멈추고, 현재 실행되기 위해 대기중인 작업 목록을 리턴한다.
• boolean isShutdown():
  Executor가 셧다운 되었는 지의 여부를 확인한다.
• boolean isTerminated():
  셧다운 실행 후 모든 작업이 종료되었는 지의 여부를 확인한다.
• boolean awaitTermination(long timeout, TimeUnit unit):
  셧다운을 실행한 뒤, 지정한 시간 동안 모든 작업이 종료될 때 까지 대기한다. 지정한 시간 이내에서 실행중인 모든 작업이 종료되면 true를 리턴하고, 여전히 실행중인 작업이 남아 있다면 false를 리턴한다.

예를 들어, 웹 서버를 종료하게 되면 더 이상 클라이언트의 요청을 받아서는 안 되고, 기존에 처리중이던 요청은 지정한 시간내에서 처리해야 할 것이다. 이런 경우 ExecutorService를 사용하면 다음과 같이 종료 과정을 코딩할 수 있다.

executor.shutdown();
try {
    if (!executor.awaitTermination(5, TimeUnit.SECONDS)) {
        System.out.println("아직 처리중인 작업 존재");
        System.out.println("작업 강제 종료 실행");
        executor.shutdownNow();
        if (!executor.awaitTermination(5, TimeUnit.SECONDS)) {
            System.out.println("여전히 종료하지 않은 작업 존재");
        }
    }
} catch (InterruptedException e1) {
    executor.shutdownNow();
    Thread.currentThread().interrupt();
}
System.out.println("서버 셧다운 완료");

ExecutorService 인터페이스는 작업 수행과 관련해서 추가적으로 메소드를 제공하고 있으며, 이들 메소드는 다음과 같다.

• <T> Future<T> submit(Callable<T> task)
  결과값을 리턴하는 작업을 추가한다.
• Future<?> submit(Runnable task)
  결과값이 없는 작업을 추가한다.
• <T> Future<T> submit(Runnable task, T result)
  새로운 작업을 추가한다. result는 작업이 성공적으로 수행될 때 사용될 리턴 값을 의미한다.
• <T> List<Future<T>> invokeAll(Collection<? extends Callable<T>> tasks)
  주어진 작업을 모두 실행한다. 각 실행 결과값을 구할 수 있는 Future의 List를 리턴한다.
• <T> List<Future<T>> invokeAll(Collection<? extends Callable<T>> tasks, long timeout, TimeUnit unit)
  앞서 invokeAll()과 동일하다. 지정한 시간 동안 완료되지 못한 작업은 취소되는 차이점이 있다.
• <T> T invokeAny(Collection<? extends Callable<T>> tasks)
  작업울 수행하고, 작업 결과 중 성공적으로 완료된 것의 결과를 리턴한다. 정상적으로 수행된 결과가 발생하거나 예외가 발생하는 경우 나머지 완료되지 않은 작업은 취소된다.
• <T> T invokeAny(Collection<? extends Callable<T>> tasks, long timeout, TimeUnit unit)
  invokeAny()와 동일하다. 지정한 시간 동안만 대기한다는 차이점이 있다.

Callable은 Runnable과 비슷한데, 차이점이 있다면 결과 값을 리턴할 수 있다는 점이다. Future는 Callable의 결과값을 구하는 데 사용된다. Callable과 Future에 대해서는 뒤에서 자세히 살펴보도록 하자.

ScheduledThreadPoolExecutor 클래스

ScheduledThreadPoolExecutor 클래스는 ScheduledExecutorService 인터페이스을 구현한 클래스로서 스케줄링 기능을 제공한다. ScheduledThreadPoolExecutor 클래스는 또한 쓰레드 풀을 구현하고 있는 ThreadPoolExecutor 클래스를 상속받고 있기 때문에 쓰레드 풀 기능도 함께 제공한다.

ScheduledThreadPoolExecutor 클래스에 대한 자세한 내용은 아래 사이트를 참고하기 바란다.

• http://java.sun.com/javase/6/docs/api/java/util/concurrent/ScheduledThreadPoolExecutor.html

Executors 유틸리티 클래스

java.util.concurrent.Executors 클래스는 자바 5가 기본적으로 제공하는 Executor 구현체를 구할 수 있는 메소드를 제공하는 유틸리티이다. 예를 들어, 쓰레드 풀을 구현한 Executor 구현체를 구하고 싶다면 다음과 같은 코드를 사용하면 된다.

Executor executor = Executors.newFixedThreadPool(THREADCOUNT);
while(true) {
    request = acceptRequest();
    Runnable requestHandler = new RequestHandler(request);
    executor.execute(requestHandler);
}

Executors 클래스는 newFixedThreadPool()과 더불어 다음과 같은 메소드를 이용하여 Executor 인스턴스를 구할 수 있도록 하고 있다.

• ExecutorService newFixedThreadPool(int nThreads)
  최대 지정한 개수 만큼의 쓰레드를 가질 수 있는 쓰레드 풀을 생성한다. 실제 생성되는 객체는 ThreadPoolExecutor 객체이다.
• ScheduledExecutorService newScheduledThreadPool(int corePoolSize)
  지정한 개수만큼 쓰레드가 유지되는 스케줄 가능한 쓰레드 풀을 생성한다. 실제 생성되는 객체는 ScheduledThreadPoolExecutor 객체이다.
• ExecutorService newSingleThreadExecutor()
  하나의 쓰레드만 사용하는 ExecutorService를 생성한다.
• ScheduledExecutorService newSingleThreadScheduledExecutor()
  하나의 쓰레드만 사용하는 ScheduledExecutorService를 생성한다.
• ExecutorService newCachedThreadPool()
  필요할 때 마다 쓰레드를 생성하는 쓰레드 풀을 생성한다. 이미 생성된 쓰레드의 경우 재사용된다. 실제 생성되는 객체는 ThreadPoolExecutor 객체이다.

Callable과 Future

Runnable 인터페이스는 다음과 같이 정의되어 있다.

public interface Runnable {
    public void run();
}

위 코드에서 볼 수 있듯이 Runnable.run() 메소드는 결과 값을 리턴하지 않기 때문에, run() 메소드의 실행 결과를 구하기 위해서는 공용 메모리나 파이프와 같은 것들을 사용해서 결과 값을 받아야만 했다. 이런 Runnable 인터페이스의 단점을 없애기 위해 추가된 것이 바로 Callable 인터페이스이다.

java.util.concurrent.Callable 인터페이스는 다음과 같이 정의되어 있다.

public Interface Callable<V> {
    V call() throws Exception
}

Callable 인터페이스의 call() 메소드는 결과 값을 리턴하도록 되어 있다. 또한, 자바 5부터 추가된 generic을 사용하여 어떤 타입이든 리턴 값으로 사용할 수 있도록 하였다. 예를 들어, 아래 코드는 Callable 인터페이스를 구현한 간단한 클래스를 보여주고 있다.

public class CallableImpl implements Callable<Integer> {
    public Integer call() throws Exception {
        // 작업 처리
        return result;
    }
}

ExecutorService.submit() 메소드를 사용하면 CallableImpl 클래스를 작업으로 사용할 수 있다. 아래 코드는 ExecutorService.submit() 메소드의 실행 예이다.

ExecutorService executor = Executors.newFixedThreadPool(THREADCOUNT);
…
Future<Integer> future = executor.submit(new CallableImpl());
Integer result = future.get();

ExecutorService.submit() 메소드는 전달받은 Callable 객체를 내부 메커니즘에 따라 지정한 때에 실행한다. 예를 들어, 위 경우 CallableImpl 객체는 큐에 저장되었다가 가용한 쓰레드가 생길 때 CallblaImpl.call() 메소드가 실행될 것이다.

Callable.call() 메소드가 ExecutorService.submit() 메소드에 전달될 때 곧 바로 실행되는 것이 아니기 때문에 리턴값을 바로 구할 수 없다. 이 리턴값은 미래의 어느 시점에 구할 수 있는데, 이렇게 미래에 실행되는 Callable의 수행 결과 값을 구할 때 사용되는 것이 Future이다. Future.get() 메소드는 Callable.call() 메소드의 실행이 완료될 때 까지 블록킹되며, Callable.call() 메소드의 실행이 완료되면 그 결과값을 리턴한다.

Future 인터페이스는 get() 메소드 외에도 다음과 같은 메소드를 제공하고 있다.

• V get()
  Callable 등 작업의 실행이 완료될 때 까지 블록킹 되며, 완료되면 그 결과값을 리턴한다.
• V get(long timeout, TimeUnit unit)
  지정한 시간 동안 작업의 실행 결과를 기다린다. 지정한 시간 내에 수행이 완료되면 그 결과값을 리턴한다. 대기 시간이 초과되면 TimeoutException을 발생시킨다.
• boolean cancel(boolean mayInterruptIfRunning)
  작업을 취소한다.
• boolean isCancelled()
  작업이 정상적으로 완료되기 이전에 취소되었을 경우 true를 리턴한다.
• boolean isDone()
  작업이 완료되었다면 true를 리턴한다.

관련링크:

• java.util.concurrent 패키지
• Concurrenty Lesson

출처 - http://javacan.tistory.com/entry/134