[reactive] 4-1. java Future/FutureTask/Callable/Runnable

callable vs runnable interface

	Runnable interface	Callable interface
java version package	java 1.0 ~ java.lang	java 1.5 ~ java.util.concurrent
return	계산 결과를 받을 수 없음	계산 결과 받을 수 있음(a generic value V)
throw	checked exception을 throw 불가	checked exception을 throw 가능
override method	run()	call()

https://www.geeksforgeeks.org/difference-between-callable-and-runnable-in-java/

Difference Between Callable and Runnable in Java - GeeksforGeeks

 

자바 비동기 맛보기

0. ExecutorService 특징

• es.execute vs es.submit 
• 시간이 오래 걸리는 작업을 할 때는 submit 추천. future object를 리턴하기 때문에 나중에 get으로 꺼낼 수 있음(get은 블로킹!)
• 그냥 스레드의 병렬 처리만 필요하면 execute

	execute()	submit()
accept	Runnable	Runnable and Callable
declared in	Executor interface	ExecutorService interface
return	void	a Future object

• es.shutdown 은 graceful shutdown이라서 할 거 다 할 때까지 기다림
• es.shutdownNow 가 강제 종료
• es.awaitTermination 은 타임아웃을 기다리는 용도로 실제 shutdown 시키지는 않음
• 따라서 shutdown 먼저 걸로 awaitTermination을 써야 함

 

1. Future class

public static void main(String[] args) throws InterruptedException, ExecutionException {
    //스래드를 새로 만들고 반납하는 것이 비용이 많이 들어가니, 풀이 나옴
    //맥시멈 제한없고 처음에 스레드 없음 요청할 때 새로 만들고 다음 요청 시 기존 만든 스레드 실행
    ExecutorService es = Executors.newCachedThreadPool();
    //es.execute -> void

    //Future 비동기 핸들러지 결과물은 아님
    Future<String> f = es.submit(() -> {
        Thread.sleep(5_000);
        log.info("helloo");
        return "hello";
    });
    //f.isDone() -> non blocking method;
    System.out.println(f.isDone());
    log.info("before");
    //f.get -> blocking method
    System.out.println(f.get()); //비동기가 완성될 때 까지 블로킹(20초 기다림)
    log.info("exit");//맨 마지막
    System.out.println(f.isDone());
}

 

2. FutureTask class

Future 자체를 object로 만들어줌

public static void main(String[] args) throws ExecutionException, InterruptedException {
    ExecutorService es = Executors.newCachedThreadPool();

    FutureTask<String> f = new FutureTask<String>(() -> {
       Thread.sleep(3_000);
       log.info("async");
       return "hello";
    }){//익명클래스
        @Override
        protected void done() { //다 하면 이거해; f.get 할 필요없음
            try {
                System.out.println(get());
            } catch (InterruptedException e) {
                e.printStackTrace();
            } catch (ExecutionException e) {
                e.printStackTrace();
            }
        }
    };

    es.execute(f);
    System.out.println(f.isDone());
    //System.out.println(f.get());
    es.shutdown(); //다 끝나면 종료
}

 

3. FutureTask 확장

public static void main(String[] args) throws ExecutionException, InterruptedException {
    ExecutorService es = Executors.newCachedThreadPool();

    CallbackFutureTask f = new CallbackFutureTask(() -> {
        Thread.sleep(3_000);
        if(1 == 1){
            throw new RuntimeException(">async error");
        }
        log.info("async");
        return "hello";
    }, s -> System.out.println("success:" + s)
    , e -> System.out.println(">error: " + e.getMessage())
    );

    es.execute(f);
    es.shutdown(); //다 끝나면 종료
}

interface SuccessCallback{
    void onSuccess(String result);
}

interface ExceptionCallback{
    void onError(Throwable t);
}

public static class CallbackFutureTask extends FutureTask<String>{
    SuccessCallback sc;
    ExceptionCallback ec;

    public CallbackFutureTask(Callable<String> callable, SuccessCallback sc, ExceptionCallback ec) {
        super(callable);
        //if(sc == null) throw null; //null point exception
        this.sc = Objects.requireNonNull(sc); //if null -> NPE
        this.ec = Objects.requireNonNull(ec);
    }

    @Override
    protected void done() {
        try {
            sc.onSuccess(get());
        } catch (InterruptedException e) { //종료 시그널
            Thread.currentThread().interrupt();
        } catch (ExecutionException e) { // 찐에러
            ec.onError(e.getCause());
        }
    }
}