Future 와 Callable 그리고 스레드 풀

/**
 * A task that returns a result and may throw an exception.
 * Implementors define a single method with no arguments called
 * {@code call}.
 *
 * <p>The {@code Callable} interface is similar to {@link
 * java.lang.Runnable}, in that both are designed for classes whose
 * instances are potentially executed by another thread.  A
 * {@code Runnable}, however, does not return a result and cannot
 * throw a checked exception.
 *
 * <p>The {@link Executors} class contains utility methods to
 * convert from other common forms to {@code Callable} classes.
 *
 * @see Executor
 * @since 1.5
 * @author Doug Lea
 * @param <V> the result type of method {@code call}
 */
@FunctionalInterface
public interface Callable<V> {
    /**
     * Computes a result, or throws an exception if unable to do so.
     *
     * @return computed result
     * @throws Exception if unable to compute a result
     */
    V call() throws Exception;
}

public interface Future<V> {

    boolean cancel(boolean mayInterruptIfRunning);

    boolean isCancelled();

    boolean isDone();

    V get() throws InterruptedException, ExecutionException;

    V get(long timeout, TimeUnit unit)
        throws InterruptedException, ExecutionException, TimeoutException;
}

• get(): 작업이 완료될 때까지 기다린 후 결과를 반환하며, 만약 작업이 실패하면 예외를 발생시킨다.
• isDone(): 작업의 완료 여부를 반환한다.
• isCancelled(): 작업의 취소 여부를 반환한다.
• cancel(): 작업을 취소 시키며,그 후 isDone() 메서드는 항상 true 를 반환한다.

public interface Executor {

    /**
     * Executes the given command at some time in the future.  The command
     * may execute in a new thread, in a pooled thread, or in the calling
     * thread, at the discretion of the {@code Executor} implementation.
     *
     * @param command the runnable task
     * @throws RejectedExecutionException if this task cannot be
     * accepted for execution
     * @throws NullPointerException if command is null
     */
    void execute(Runnable command);
}

<T> Future<T> submit(Callable<T> task);
<T> Future<T> submit(Runnable task, T result);
Future<?> submit(Runnable task);

public class ExecuteServiceMain {

    public static void main(String[] args) {
        ExecutorService executorService = Executors.newFixedThreadPool(5);
        executorService.execute(
                () -> {
                    System.out.println("Executor in thread: " + Thread.currentThread().getName());
                    while (true) {
                        if (Thread.currentThread().isInterrupted()) {
                            System.out.println("Interrupted");
                            break;
                        }
                    }
                }
        );
        executorService.shutdownNow();
    }
}

// 콘솔 출력
// Executor in thread: pool-1-thread-1
// Interrupted

public interface ScheduledExecutorService extends ExecutorService {

    /**
     * Submits a one-shot task that becomes enabled after the given delay.
     */
    public ScheduledFuture<?> schedule(Runnable command,
                                       long delay, TimeUnit unit);

    /**
     * Submits a value-returning one-shot task that becomes enabled
     * after the given delay.
     */
    public <V> ScheduledFuture<V> schedule(Callable<V> callable,
                                           long delay, TimeUnit unit);

    /**
     * Submits a periodic action that becomes enabled first after the
     * given initial delay, and subsequently with the given period;
     */
    public ScheduledFuture<?> scheduleAtFixedRate(Runnable command,
                                                  long initialDelay,
                                                  long period,
                                                  TimeUnit unit);

    /**
     * Submits a periodic action that becomes enabled first after the
     * given initial delay, and subsequently with the given delay
     * between the termination of one execution and the commencement of
     * the next.
     */
    public ScheduledFuture<?> scheduleWithFixedDelay(Runnable command,
                                                     long initialDelay,
                                                     long delay,
                                                     TimeUnit unit);

}

• schedule: 정의한 작업을 일정 시간 뒤에 실행 시킨다.
• scheduleAtFixedRate: initialDelay 인자 값 만큼 지난 후 작업이 실행 되고, period 인자 값 만큼 주기적으로 실행하는데, 작업이 늦게 끝나는 경우 끝난 후 지연 없이 바로 실행된다.
  • 예시) 1초 뒤 작업이 실행되고, 1분 동안 주기적으로 실행 되어 폴링하는 구조
    • scheduleExecutorService.scheduleAtFixedRate(task, 1, 60, TimeUnit.SECONDS);
• scheduleWithFixedDelay: initialDelay 인자 값 만큼 지난 후 작업이 실행 되고, delay 인자 값 만큼 주기적으로 실행하는데, 작업이 늦게 끝나는 경우 delay 만큼 지연 후 실행된다.

    public ThreadPoolExecutor(int corePoolSize,
                              int maximumPoolSize,
                              long keepAliveTime,
                              TimeUnit unit,
                              BlockingQueue<Runnable> workQueue) {
        this(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue,
             Executors.defaultThreadFactory(), defaultHandler);
    }

    public ThreadPoolExecutor(int corePoolSize,
                              int maximumPoolSize,
                              long keepAliveTime,
                              TimeUnit unit,
                              BlockingQueue<Runnable> workQueue,
                              ThreadFactory threadFactory) {
        this(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue,
             threadFactory, defaultHandler);
    }

    public ThreadPoolExecutor(int corePoolSize,
                              int maximumPoolSize,
                              long keepAliveTime,
                              TimeUnit unit,
                              BlockingQueue<Runnable> workQueue,
                              RejectedExecutionHandler handler) {
        this(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue,
             Executors.defaultThreadFactory(), handler);
    }

    public ThreadPoolExecutor(int corePoolSize,
                              int maximumPoolSize,
                              long keepAliveTime,
                              TimeUnit unit,
                              BlockingQueue<Runnable> workQueue,
                              ThreadFactory threadFactory,
                              RejectedExecutionHandler handler) {
        if (corePoolSize < 0 ||
            maximumPoolSize <= 0 ||
            maximumPoolSize < corePoolSize ||
            keepAliveTime < 0)
            throw new IllegalArgumentException();
        if (workQueue == null || threadFactory == null || handler == null)
            throw new NullPointerException();
        this.corePoolSize = corePoolSize;
        this.maximumPoolSize = maximumPoolSize;
        this.workQueue = workQueue;
        this.keepAliveTime = unit.toNanos(keepAliveTime);
        this.threadFactory = threadFactory;
        this.handler = handler;

        String name = Objects.toIdentityString(this);
        this.container = SharedThreadContainer.create(name);
    }

• corePoolSize: 스레드 풀 고정 크기
• maximumPoolSize: 최대 확장 가능한 스레드 풀 크기
• keepAliveTime: corePoolSize 보다 더 많은 스레드가 요구 되어 maximumPoolSize 값 만큼 확장이 일어난 경우, 해당 시간보다 더 길게 사용하면 초과된 스레드가 종료되고 corePoolSize 로 다시 돌아감
  • TimeUnit: keepAliveTime 을 적용하는 시간 단위
• workQueue: 작업을 수행해야 할 스레드 관리 큐(corePoolSize 에 관리되는 스레드가 모두 작업을 수행중인 경우 해당 큐에 저장됨)
  • SynchronousQueue: 큐 크기가 0이고, 작업을 직접 스레드에 전달하며 즉시 처리 가능한 스레드가 없으면 새 스레드를 생성한다. Executors.newCachedThreadPool() 에 사용되고 빠른 응답 시간, 무제한으로 스레드를 생성 할 수 있다는 특징이 있음
  • LinkedBlockingQueue: 연결 리스트 기반의 큐이며 FIFO 순서를 보장하고, 인자 값으로 큐 크기를 지정하지 않으면 Integer.MaxValue 까지 동적으로 늘어난다. Executors** 의 기본적인 큐로 사용되고 put, take 각각의 Lock 을 사용하기 때문에 락 경합이 ArrayBlockingQueue **보다 줄어들고, 인자를 지정하지 않고 사용하는 경우 메모리 부족이 발생하지만 대체적으로 안정적인 처리량이 특징이다.
  • ArrayBlockingQueue: 고정 크기 기반의 배열 큐로 FIFO 순서를 보장한다. put, take 모두 한 개의 Lock 을 사용하여 락 경합이 발생할 수 있으며 최초 생성 시 인자 값을 통해 용량을 결정한 후 변경이 불가능 하다는 특징이 있다.
• threadFactory: 기본 값으로 defaultThreadFactory 를 사용하며, 스레드를 생성할 때 스레드의 이름을 지정하거나 우선순위를 지정할 수 있다. 기본 팩토리에서 우선순위 5로 설정하여 스레드가 생성된다.
• handler: 모든 스레드가 사용되고 있고, workQueue가 포화상태라면 새 작업을 거부하는데, 이때 다양한 정책을 설정할 수 있다.
  • AbortPolicy: 기본 전략으로 refectedExecution 예외를 발생 시킨다.
  • CallerRunsPolicy: 실행을 호출하는 스레드가 스스로 작업을 실행한다.
  • DiscardPolicy: 현재 요청한 작업이 거절되면 삭제한다.
  • DiscardOldestPolocy: 가장 오래되고 처리되지 않은 작업을 삭제한 뒤 다시 작업을 시도한다.

ExecutorService fixedThreadPool = Executors.newFixedThreadPool(5);

ExecutorService cachedThreadPool = Executors.newCachedThreadPool();

ScheduledExecutorService scheduledThreadPool = Executors.newScheduledThreadPool(5);

ExecutorService singleThread = Executors.newSingleThreadExecutor();