토비의 봄 TV 12회 스프링 리액티브 프로그래밍 (8) WebFlux

(시청일 : 20170114)

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

- 책 추천 : RXJava를 활용한 리액티브 프로그래밍

■ 프로젝트 환경

- 스프링부트 버전 : SpringBoot 2.0 M1

- 스프링 프레임워크 버전 : SpringFramework 5.0 RC1

- Core : Lombok

- Web : Reactive Web  

           (Web과 Reactive Web는 서로 배타적이라 한 프로젝트에 동시에 적용될 수 없다. 두가지 모두 선택하게 되면 Web이 우선시된다)

■ 예제 1

/* 

*

* LoadTest.java 

*

*/

package toby.live;

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());

}

}

/*

* 

* Toby012Application.java

* (원래는 Netty에서 돌아가야하지만, 한 프로젝트에서 Tomcat과 Netty를 동시에 돌릴 수 없어서 Tomcat에서 실행하였음.

*/

package toby.live;

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

import org.springframework.boot.SpringApplication;

import org.springframework.boot.autoconfigure.SpringBootApplication;

import org.springframework.scheduling.annotation.Async;

import org.springframework.scheduling.annotation.EnableAsync;

import org.springframework.stereotype.Service;

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

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

import org.springframework.web.reactive.function.client.WebClient;

import reactor.core.publisher.Mono;

@SpringBootApplication

@RestController

@EnableAsync

public class Toby012Application {

    static final String URL1 = "http://localhost:8081/service?req={req}";;

    static final String URL2 = "http://localhost:8081/service2?req={req}";;

    @Autowired

    MyService myService;

    WebClient client = WebClient.create();

    @GetMapping("/rest")

    public Mono<String> rest(int idx) {

        return client.get().uri(URL1, idx).exchange() // Mono<ClientResponse>

                .flatMap(c ->  c.bodyToMono(String.class)) // Mono<String>

                .flatMap((String res1) -> client.get().uri(URL2, res1).exchange()) // Mono<ClientResponse>

                .flatMap(c -> c.bodyToMono((String.class)));  // Mono<String>

    }

    public static void main(String[] args) {

        System.setProperty("reactor.ipc.netty.workerCount", "2");

        System.setProperty("reactor.ipc.netty.pool.maxConnections", "2000");

        SpringApplication.run(Toby012Application.class, args);

    }

    @Service

    public static class MyService {

        public String work(String req) {

            return req + "/asyncwork";

        }

    }

}

/*

* 

* RemoteService.java

* (Tomcat에서 실행)

*/

package toby.live;

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("/service")

        public String service(String req) throws InterruptedException {

            Thread.sleep(1000);

            //throw new RuntimeException();

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

        }

        @GetMapping("/service2")

        public String service2(String req) throws InterruptedException {

            Thread.sleep(1000);

            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);

    }

}

 

■ 예제 2

/* 

*

* LoadTest.java 

*

*/

package toby.live;

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());

}

}

/*

* 

* Toby012Application.java

* (원래는 Netty에서 돌아가야하지만, 한 프로젝트에서 Tomcat과 Netty를 동시에 돌릴 수 없어서 Tomcat에서 실행하였음.

*/

package toby.live;

import lombok.extern.slf4j.Slf4j;

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

import org.springframework.boot.SpringApplication;

import org.springframework.boot.autoconfigure.SpringBootApplication;

import org.springframework.scheduling.annotation.Async;

import org.springframework.scheduling.annotation.EnableAsync;

import org.springframework.stereotype.Service;

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

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

import org.springframework.web.reactive.function.client.WebClient;

import reactor.core.publisher.Mono;

import java.util.concurrent.CompletableFuture;

@SpringBootApplication

@RestController

@Slf4j

@EnableAsync

public class Toby012Application {

    static final String URL1 = "http://localhost:8081/service?req={req}";;

    static final String URL2 = "http://localhost:8081/service2?req={req}";;

    @Autowired

    MyService myService;

    WebClient client = WebClient.create();

    @GetMapping("/rest")

    public Mono<String> rest(int idx) {

        return client.get().uri(URL1, idx).exchange() // Mono<ClientResponse>

                .flatMap(c ->  c.bodyToMono(String.class)) // Mono<String>

                //.doOnNext(c -> log.info(c))

                .flatMap(res1 -> client.get().uri(URL2, res1).exchange()) // Mono<ClientResponse>

                .flatMap(c -> c.bodyToMono((String.class))) // Mono<String>

                .doOnNext(c -> log.info(c))

                .flatMap(res2 -> Mono.fromCompletionStage(myService.work(res2)))      // CompletableFuture<String> -> Mono<String>

                .doOnNext(c -> log.info(c));

    }

    public static void main(String[] args) {

        System.setProperty("reactor.ipc.netty.workerCount", "2");

        System.setProperty("reactor.ipc.netty.pool.maxConnections", "2000");

        SpringApplication.run(Toby012Application.class, args);

    }

    @Service

    public static class MyService {

        @Async

        public CompletableFuture<String> work(String req) {

            return CompletableFuture.completedFuture(req + "/asyncwork");

        }

    }

}

/*

* 

* RemoteService.java

* (Tomcat에서 실행)

*/

package toby.live;

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("/service")

        public String service(String req) throws InterruptedException {

            Thread.sleep(1000);

            //throw new RuntimeException();

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

        }

        @GetMapping("/service2")

        public String service2(String req) throws InterruptedException {

            Thread.sleep(1000);

            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);

    }

}

■ application.properties

server.tomcat.max-threads=1