Java Executor(Thread 생성, 관리)

프로젝트 중 Java API Method의 프로세스 유효 시간을 적용해야 할 일이 생긴적이 있었다. 해서 공부를 해봤는데...

 

Java에서는 Executor, ExecutorService 객체를 통해 Thread를 생성/관리를 할 수 있었다.

 

종류는 크게 3가지가 존재한다.

CachedThreadPool
 - Thread Caching
 - 60초동안 작업 없으면 Thread Pool에서 제거

FixedThreadPool
 - 고정된 Thread 개수를 가진다. - fixedThreadPool을 생성 할 때 해당 머신의 CPU 코어수를 기준으로 생성 시 더 좋은 performance를 얻을 수 있다고 한다.

SingleThreadExecutor
 - 한 개의 Thread로 작업을 처리

 

package timeoutTest;

import java.util.concurrent.*;

public class TimeoutTest1 {
    // fixedThreadPool 생성 갯수
    public static int CONF_THREADCNT = 2;

    // time out cofig value
    public static Long CONF_TIMEOUT = 5000L;

    // 테스트용 Thread Sleep time
    public static Long CONF_PROCESSSLEEPTIME = 1000L;

    // time out Messagae
    public static String TIMEOUTMSG = "time over...";

    public static void main(String[] args) throws Exception {
        /*
            CachedThreadPool
             - Thread Caching
             - 60초동안 작업 없으면 Thread Pool에서 제거

            FixedThreadPool
             - 고정된 Thread 개수를 가진다.
             - fixedThreadPool을 생성 할 때 해당 머신의 CPU 코어수를 기준으로 생성 시 더 좋은 performance를 얻을 수 있음.

            SingleThreadExecutor
             - 한 개의 Thread로 작업을 처리
        */
        TimeoutTest1 t = new TimeoutTest1();

        ExecutorService cachedThreadPool = t.getCachedThreadPool();
        ExecutorService fixedThreadPool = t.getFixedThreadPool();
        ExecutorService singleThreadPool = t.getSingleThreadPool();

        String result;
        int processResult;

        t.doTest1(cachedThreadPool);
        // 하위 문자열 출력을 통해 doTest1 method는 submit method의 process 종료를 기다리지 않음을 알 수 있다.
        System.out.println("기다리지 않는다.");

        result = t.doTest2(cachedThreadPool);
        // 하위 result 변수값 출력을 통해 doTest2 method는 submit method process 종료를 기다리는것을 알 수 있다.
        System.out.println(result);

        try {
            t.doTest3(cachedThreadPool);
        } catch (Exception e) {
            /*
                doTest3 process 중 timeout 발생 시 throws Exception으로 인해 해당 catch문으로 빠지게 된다.
                일반적으로 생각했을 경우 Exception이 발생하면 해당 process는 종료되어야 하나 실행시켜보면 종료되지 않고 Runnable에 정의된 process가 계속 실행됨을 알 수 있다. (doTest4 참조)
            */
            System.out.println(TIMEOUTMSG);
        }

        try {
            t.doTest4(cachedThreadPool);
        } catch (InterruptedException ie) {
            System.out.println("여기로 오나?");
        } catch (Exception e) {
            // time out 발생 시 process를 종료시키기 위한 방법

            /*
                shutdown();
                 - 현재 process중인 task는 계속 진행 후 종료하며 새로운 task는 수락하지 않는다.
                 - 혼동 주의) cachedThreadPool의 task를 저지 할 뿐, 실행시켜보면 doTest4 method의 호출은 막을 수 없다.
                             shutdown, shutdownNow는 blocking을 발생시키지 않는다. 즉, 이후 정의된 프로세스는 Callable(혹은 Runnable)프로세스의 상태와 상관 없이 진행된다.
                 - shutdown method 사용 중 현재 task 종료를 기다리기 위한(혹은 다른 이유로) blocking이 필요한 상황이라면 awaitTermination(long timeout, TimeUnit unit) 사용한다.

                * awaitTermination
                 - 현재 task가 timeout 내에 프로세스 되었다면 true, 아니면 false를 return한다.
            */
            cachedThreadPool.shutdown();

            // 아래 출력문을 통해 blocking이 발생되지 않음을 확인할 수 있음.
            System.out.println(TIMEOUTMSG);

            // shutdown 이후 재호출 시 RejectedExecutionException이 발생한다.
            t.doTest4(cachedThreadPool);

            /*
                shutdownNow();
                 - 현재 thread를 바로 중단시킨다. current Thread interrupt와 동반되어야 하며 doTest4 참고
            */
        }
    }

    public void doTest1(ExecutorService cachedThreadPool) throws Exception {
        System.out.println("doTest1 started");

        cachedThreadPool.submit(() -> {
            // Callable 객체를 인스턴스 하여 submit의 parameter에 제공되어야 하며 이와 같이 Lambda Expression으로 간략하게 표현 가능하다.
            // submit method를 호출하면 doTest1 method는 submit method의 process 종료를 기다리지 않고 비동기로 진행된다.

            int cnt = 1;

            try {
                while (cnt <= 6) {
                    System.out.println(cnt + " Seconds...");

                    processSleep(CONF_PROCESSSLEEPTIME);

                    cnt++;
                }
            } catch (Exception e) {
                e.printStackTrace();

                return "process failed";
            }

            return "process success";
        });

        System.out.println("doTest1 ended");
    }

    public String doTest2(ExecutorService cachedThreadPool) throws Exception {
        System.out.println("doTest2 started");

        String result;

        result = cachedThreadPool.submit(() -> {
            int cnt = 1;

            try {
                while (cnt <= 6) {
                    System.out.println(cnt + " Seconds...");

                    processSleep(CONF_PROCESSSLEEPTIME);

                    cnt++;
                }

                return "process success";
            } catch (Exception e) {
                e.printStackTrace();

                return "process failed";
            }
        }).get();
        //    ↖ submit method는 Future Class를 반환하며 이를 통해 get method를 호출 할 수 있다.
        //      get method를 호출하게 되면 doTest2 method는 submit method process 종료를 기다린다.
        //      parameter를 제공하여 해당 process의 time out을 설정 할 수 있다. (doTest3 참조)

        System.out.println("doTest2 ended");
        return result;
    }

    public void doTest3(ExecutorService cachedThreadPool) throws Exception {
        System.out.println("doTest3 started");

        cachedThreadPool.submit(() -> {
            int cnt = 1;

            try {
                while (cnt <= 6) {
                    System.out.println(cnt + " Seconds...");

                    processSleep(CONF_PROCESSSLEEPTIME);

                    cnt++;
                }
            } catch (Exception e) {
                e.printStackTrace();
            }
        }).get(CONF_TIMEOUT, TimeUnit.MILLISECONDS);
        // submit method의 time out은 5초로 설정되어 있는 상태이며 Runnable에 정의된 process는 총 6초 정도가 걸리는 상황.

        System.out.println("doTest3 ended");
    }

    public void doTest4(ExecutorService cachedThreadPool) throws Exception {
        System.out.println("doTest4 started");

        try {
            cachedThreadPool.submit(() -> {
                Thread t = Thread.currentThread();

                int cnt = 1;

                try {
                    while (!t.isInterrupted()) {
                        if (cnt > 6) {
                            break;
                        }

                        System.out.println(cnt + " Seconds...");

                        processSleep(CONF_PROCESSSLEEPTIME);

                        cnt++;
                    }
                } catch (InterruptedException ie) {
                    System.out.println("current Thread is interrupted");

                    t.interrupt();
                } catch (Exception e) {
                    System.out.println("ee");
                    e.printStackTrace();
                }
            }).get(CONF_TIMEOUT, TimeUnit.MILLISECONDS);
        } catch (TimeoutException te) {
            System.out.println("te");

            /*
                shutdownNow();
                 - shutdown과는 다르게 현재 수행중인 task
                 - Executors로 생성한 thread는 none-daemon thread이므로 이를 수동으로 종료 시켜야 한다.
                 - Thread interrupt를 이용한 강제 종료가 구현되지 않으면 현 task에 대한 강제 종료를 보장 할 수 없다.
            */
            cachedThreadPool.shutdownNow();
            Thread.currentThread().interrupt();
        } catch (InterruptedException ie) {
            System.out.println("ie");
        } catch (Exception e) {
            System.out.println("e");
        }

        System.out.println("doTest4 ended");
    }

    public ExecutorService getCachedThreadPool() throws Exception {
        return Executors.newCachedThreadPool();
    }

    public ExecutorService getFixedThreadPool() throws Exception {
        return Executors.newFixedThreadPool(CONF_THREADCNT);
    }

    public ExecutorService getSingleThreadPool() throws Exception {
        return Executors.newSingleThreadExecutor();
    }

    public void processSleep(Long time) throws Exception {
        Thread.sleep(time);
    }
}