토비의 봄 TV 9회 스프링 리액티브 프로그래밍 (5) 비동기 RestTemplate과 비동기 MVC/Servlet

(시청일 : 20171119)

[스프링 리액티브 웹 개발 5부. 비동기 RestTemplate과 비동기 MVC의 결합]

- https://www.youtube.com/watch?v=ExUfZkh7Puk

- CyclicBarrier : 자바에서 사용하는 simple 동기화 기법

<Callable과 Runnable 의  차이점>

	Callable	Runnable
리턴값 존재 여부	ㅇ	X
Exception 던지도록 선언되어있는지 여부	ㅇ	X

- ListenableFuture : 비동기 작업 결과를 가져올 수 있는 future 타입인데, 성공/실패 시의 콜백을 등록할 수 있다.

- DeferredResult : 컨트롤러의 리턴을 별도의 작업에서 할 수 있게 해준다. 하지만 이스레드는 스프링에의해 관리 되는 것이 아니다.

 DeferredResult클래스는 어떤 요청에 대한 응답을 이벤트를 Queue에 저장하고 있다가 DeferredResult.setResult() 메소드가 호출되면 DispatcherSerlvet으로 응답을 보낸다.  즉 서버가 Push하는 기술들을 구현할 수 있게 해준다.

출처: http://highlucksw.tistory.com/37

<자바의 쓰레드풀의 동작원리>

큐를 먼저 채우다가 큐가 다 차면, maxPoolSize까지 쓰레드를 더 추가로 늘렸다가 쓰레드가 다 차면, 오류 발생

■ Tobytv009Application.java

package toby.live.asyncrest;

import io.netty.channel.nio.NioEventLoopGroup;

import org.springframework.beans.factory.annotation.Autowired;

import org.springframework.boot.SpringApplication;

import org.springframework.boot.autoconfigure.AutoConfigureOrder;

import org.springframework.boot.autoconfigure.SpringBootApplication;

import org.springframework.context.annotation.Bean;

import org.springframework.http.ResponseEntity;

import org.springframework.http.client.Netty4ClientHttpRequestFactory;

import org.springframework.scheduling.annotation.Async;

import org.springframework.scheduling.annotation.AsyncResult;

import org.springframework.scheduling.annotation.EnableAsync;

import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;

import org.springframework.stereotype.Service;

import org.springframework.util.concurrent.ListenableFuture;

import org.springframework.web.bind.annotation.GetMapping;

import org.springframework.web.bind.annotation.RestController;

import org.springframework.web.client.AsyncRestTemplate;

import org.springframework.web.client.RestTemplate;

import org.springframework.web.context.request.async.DeferredResult;

@SpringBootApplication

@EnableAsync

public class Tobytv009Application {

    @RestController

    public static class MyController {

        //RestTemplate rt = new RestTemplate(); // 블로킹 방식  => 각각 2초 작업시간이 걸리는 100개의 API 호출 시에 하나의 쓰레드로 처리하기 때문에 약 200초 걸림

        //AsyncRestTemplate rt = new AsyncRestTemplate(); // 논블로킹 방식 => 각각 2초 작업시간이 걸리는 100개의 API 호출 시에 100개의 쓰레드로 처리하기 때문에 약 2초 걸림

        AsyncRestTemplate rt = new AsyncRestTemplate(new Netty4ClientHttpRequestFactory(new NioEventLoopGroup(1))); // 논블로킹 방식 => 각각 2초 작업시간이 걸리는 100개의 API 호출 시에 1개의 쓰레드로 처리하기 때문에 약 2초 걸림

//        @GetMapping("/rest")

//        public String rest(int idx) {

//            String res = rt.getForObject("http://localhost:8081/service?req={req}";, String.class, "hello" + idx);

//

//            return res;  // html/text

//        }

//        @GetMapping("/rest")

//        public ListenableFuture<ResponseEntity<String>> rest(int idx) {

//            return rt.getForEntity("http://localhost:8081/service?req={req}";, String.class, "hello" + idx);

//        }

        @Autowired MyService myService;

        @GetMapping("/rest")

        public DeferredResult<String> rest(int idx) {   // 결과를 가공하여 리턴 또는 다른 API와 의존적인 관계로 만드는 법

            DeferredResult<String> dr = new DeferredResult<>();

            ListenableFuture<ResponseEntity<String>> f1 = rt.getForEntity("http://localhost:8081/service1?req={req}";, String.class, "hello" + idx);

            f1.addCallback(s1->{

                ListenableFuture<ResponseEntity<String>> f2 = rt.getForEntity("http://localhost:8081/service2?req={req}";, String.class, s1.getBody());

                f2.addCallback(s2->{

                     //dr.setResult(s2.getBody());

                    ListenableFuture<String> f3 =  myService.work(s2.getBody());

                    f3.addCallback(s3->{

                        dr.setErrorResult(s3);

                    },e3->{

                        dr.setErrorResult(e3.getMessage());

                    });

                },e2->{

                    dr.setErrorResult(e2.getMessage());

                });

            }, e1->{

                dr.setErrorResult(e1.getMessage());

            });

            return dr;

        }

    }

    @Service

    public static class MyService {

        @Async

        public ListenableFuture<String> work(String req) {

            return new AsyncResult<>(req + "/asyncwork");

        }

    }

    @Bean

    ThreadPoolTaskExecutor myThreadPool() {  //

        ThreadPoolTaskExecutor te = new ThreadPoolTaskExecutor();

        te.setCorePoolSize(1);

        te.setMaxPoolSize(1);

        te.initialize();

        return te;

    }

    public static void main(String[] args) {

        SpringApplication.run(Tobytv009Application.class, args);

    }

}

■ RemoteService.java

package toby.live.asyncrest;

import org.springframework.boot.SpringApplication;

import org.springframework.boot.autoconfigure.SpringBootApplication;

import org.springframework.web.bind.annotation.GetMapping;

import org.springframework.web.bind.annotation.RestController;

@SpringBootApplication

public class RemoteService {

    @RestController

    public static class MyController {

        @GetMapping("/service1")

        public String service1(String req) throws InterruptedException {

            Thread.sleep(2000);

            return req + "/service1";  // html/text

        }

        @GetMapping("/service2")

        public String service2(String req) throws InterruptedException {

            Thread.sleep(2000);

            return req + "/service2";  // html/text

        }

    }

    public static void main(String[] args) {

        System.setProperty("SERVER.PORT","8081");

        System.setProperty("server.tomcat.max-threads","1000");

        SpringApplication.run(RemoteService.class, args);

    }

}

■ LoadTest.java

package toby.live.asyncrest;

import lombok.extern.slf4j.Slf4j;

import org.springframework.util.StopWatch;

import org.springframework.web.client.RestTemplate;

import java.util.concurrent.*;

import java.util.concurrent.atomic.AtomicInteger;

@Slf4j

public class LoadTest {

    static AtomicInteger counter = new AtomicInteger(0);

    public static void main(String[] args) throws InterruptedException, BrokenBarrierException {

        ExecutorService es = Executors.newFixedThreadPool(100);

        RestTemplate rt = new RestTemplate();

        String url = "http://localhost:8080/rest?idx={idx}";;

        CyclicBarrier barrier = new CyclicBarrier(101);

        StopWatch main = new StopWatch();

        main.start();

        for (int i=0; i<100; i++){

            es.submit(()->{

                int idx = counter.addAndGet(1);

                barrier.await();

                log.info("Thread {}", idx);

                StopWatch sw = new StopWatch();

                sw.start();

                String res = rt.getForObject(url,String.class, idx);

                sw.stop();

                log.info("Elapsed: {} {} / {}", idx, sw.getTotalTimeSeconds(), res);

                return null;

            });

        }

        barrier.await();

        es.shutdown();

        es.awaitTermination(100, TimeUnit.SECONDS);

        main.stop();

        log.info("Total: {}", main.getTotalTimeSeconds());

    }

}