Spring Framework 5.0による Reactive Web Application #JavaDayTokyo

‹#›© 2016 Pivotal Software, Inc. All rights reserved. ‹#›© 2017 Pivotal Software, Inc. All rights reserved. Spring Framework 5.0 Reactive Web Application Toshiaki Maki (@making) 2016-05-17 Java Day Tokyo 2017

© 2017 Pivotal Software, Inc. All rights reserved. Reactive??? "In a nutshell reactive programming is about non-blocking, event-driven applications that scale with a small number of threads with backpressure as a key ingredient that remains to ensure producers do not overwhelm consumers" - Rossen Stoyanchev

© 2017 Pivotal Software, Inc. All rights reserved. Use Case: Serve a lot of slow clients https://speakerdeck.com/sdeleuze/developing-reactive-applications-with-reactive-streams-and-java-8 📱📱 📱📱 📱📱 📱📱 📱📱 📱📱

© 2017 Pivotal Software, Inc. All rights reserved. Use Case: Push message to client https://speakerdeck.com/sdeleuze/developing-reactive-applications-with-reactive-streams-and-java-8 💻 💬💬💬 Server-Sent Events WebSocket RabbitMQ Apache Kafka ∞

© 2017 Pivotal Software, Inc. All rights reserved. blocking ====> event-based public class IAmBlocking { public void blockingCode() { String a = callRemoteA(); String b = callRemoteB(); String c = callRemoteC(); String d = callRemoteD(); System.out.println(a+" "+b+" "+c+" "+d); } } 😅

© 2017 Pivotal Software, Inc. All rights reserved. blocking ====> event-based public class AmIReactive { public void amIReallyReactive() { callRemoteA(a -> { callRemoteB(b -> { callRemoteC(c -> { callRemoteD(d -> println(a+" "+b+" "+c+" "+d), exd -> exd.printStackTrace()) }, exc -> exc.printStackTrace()) }, exb -> exb.printStackTrace() }, exa -> exa.printStackTrace());}} 😕

© 2017 Pivotal Software, Inc. All rights reserved. public class IAmReactive { public void iAmReallyReactive() { when(callRemoteA(), callRemoteB(), callRemoteC(), callRemoteD()) .doOnError(e -> e.printStakcTrace()) .subscribe(t -> // Tuple4<A, B, C, D> println(t.t1 +" "+t.t2+" "+t.t3+" "+t.t4)); blocking ====> event-based 😀

© 2017 Pivotal Software, Inc. All rights reserved. Reactive Streams •Standard interfaces for asynchronous stream processing with non-blocking back pressure •De-facto standard for interoperability between reactive libraries •Implemented by http://www.reactive-streams.org/

© 2017 Pivotal Software, Inc. All rights reserved. Reactive Streams •Standard interfaces for asynchronous stream processing with non-blocking back pressure •De-facto standard for interoperability between reactive libraries •Implemented by http://www.reactive-streams.org/ RxJava 2 Reactor Akka Streams

‹#›© 2016 Pivotal Software, Inc. All rights reserved. Reactive Streams has 4 interfaces public interface Publisher<T> { void subscribe(Subscriber<? super T> s); } public interface Subscription { void request(long n); void cancel(); } public interface Subscriber<T> { void onSubscribe(Subscription s); void onNext(T t); void onError(Throwable t); void onComplete(); } public interface Processor<T, R> extends Publisher<T>, Subscriber<R> {}

© 2017 Pivotal Software, Inc. All rights reserved. Back-pressure •Allows to control the amount of inflight data •Regulate the transfer between •Slow publisher and fast consumer •Fast publisher and slow consumer https://speakerdeck.com/sdeleuze/developing-reactive-applications-with-reactive-streams-and-java-8

© 2017 Pivotal Software, Inc. All rights reserved. Reactor •Natively built on top of Reactive Streams with Rx API •Developed by Pivotal •Focus on Java 8 •java.util.function.* •Duration / CompletableFuture / Stream • Lightweight Rx API with 2 types: •Flux / Mono https://projectreactor.io/

© 2017 Pivotal Software, Inc. All rights reserved. Flux Flux<Integer> stream1 = Flux.just(1, 2, 3) .map(x -> x * 2) .filter(x -> x > 2); // 4, 6 Flux<String> stream2 = Flux.just("a", "b", "c"); Flux.zip(stream1, stream2)  .doOneNext(t -> println(t.t1 + ":" + t.t2)) .subscribe(); Flux.merge(stream1, stream2)  .doOneNext(x -> println(x)).subscribe();

© 2017 Pivotal Software, Inc. All rights reserved. Flux Flux<Integer> stream1 = Flux.just(1, 2, 3) .map(x -> x * 2) .filter(x -> x > 2); // 4, 6 Flux<String> stream2 = Flux.just("a", "b", "c"); Flux.zip(stream1, stream2)  .doOneNext(t -> println(t.t1 + ":" + t.t2)) .subscribe(); Flux.merge(stream1, stream2)  .doOneNext(x -> println(x)).subscribe(); 4:a 6:b

© 2017 Pivotal Software, Inc. All rights reserved. Flux Flux<Integer> stream1 = Flux.just(1, 2, 3) .map(x -> x * 2) .filter(x -> x > 2); // 4, 6 Flux<String> stream2 = Flux.just("a", "b", "c"); Flux.zip(stream1, stream2)  .doOneNext(t -> println(t.t1 + ":" + t.t2)) .subscribe(); Flux.merge(stream1, stream2)  .doOneNext(x -> println(x)).subscribe(); 4:a 6:b 4 6 a b c

© 2017 Pivotal Software, Inc. All rights reserved. String location = "Tokyo, Japan"; mainService.fetchWeather(location) .timeout(Duration.ofSeconds(2)) .doOnError(ex -> logger.error(ex.getMessage())) .onErrorResume(ex -> backupService.fetchWeather(location)) .map(w -> format("Weather in %s is %s", w.location(), w.description())) .subscribe(message -> logger.info(message));

© 2017 Pivotal Software, Inc. All rights reserved. String location = "Tokyo, Japan"; mainService.fetchWeather(location) .timeout(Duration.ofSeconds(2)) .doOnError(ex -> logger.error(ex.getMessage())) .onErrorResume(ex -> backupService.fetchWeather(location)) .map(w -> format("Weather in %s is %s", w.location(), w.description())) .subscribe(message -> logger.info(message)); Mono<Weather> fetchWeather(String city);

© 2017 Pivotal Software, Inc. All rights reserved. String location = "Tokyo, Japan"; mainService.fetchWeather(location) .timeout(Duration.ofSeconds(2)) .doOnError(ex -> logger.error(ex.getMessage())) .onErrorResume(ex -> backupService.fetchWeather(location)) .map(w -> format("Weather in %s is %s", w.location(), w.description())) .subscribe(message -> logger.info(message)); times out and emits an error after 2 sec

© 2017 Pivotal Software, Inc. All rights reserved. String location = "Tokyo, Japan"; mainService.fetchWeather(location) .timeout(Duration.ofSeconds(2)) .doOnError(ex -> logger.error(ex.getMessage())) .onErrorResume(ex -> backupService.fetchWeather(location)) .map(w -> format("Weather in %s is %s", w.location(), w.description())) .subscribe(message -> logger.info(message)); logs a message in case of errors

© 2017 Pivotal Software, Inc. All rights reserved. String location = "Tokyo, Japan"; mainService.fetchWeather(location) .timeout(Duration.ofSeconds(2)) .doOnError(ex -> logger.error(ex.getMessage())) .onErrorResume(ex -> backupService.fetchWeather(location)) .map(w -> format("Weather in %s is %s", w.location(), w.description())) .subscribe(message -> logger.info(message)); switches to a different service in case of error

© 2017 Pivotal Software, Inc. All rights reserved. String location = "Tokyo, Japan"; mainService.fetchWeather(location) .timeout(Duration.ofSeconds(2)) .doOnError(ex -> logger.error(ex.getMessage())) .onErrorResume(ex -> backupService.fetchWeather(location)) .map(w -> format("Weather in %s is %s", w.location(), w.description())) .subscribe(message -> logger.info(message)); transforms a weather instance into a String message

© 2017 Pivotal Software, Inc. All rights reserved. String location = "Tokyo, Japan"; mainService.fetchWeather(location) .timeout(Duration.ofSeconds(2)) .doOnError(ex -> logger.error(ex.getMessage())) .onErrorResume(ex -> backupService.fetchWeather(location)) .map(w -> format("Weather in %s is %s", w.location(), w.description())) .subscribe(message -> logger.info(message)); triggers the processing of the chain

© 2017 Pivotal Software, Inc. All rights reserved. Type comparison No value Single value Multiple values JDK CompletableFuture<Void> CompletableFuture<T> CompletableFuture<List<T>> Reactive Streams Publisher<Void> Publisher<T> Publisher<T> RxJava1 Completable Single<T> Observable<T> RxJava2 Completable  Maybe<T> Single<T> Maybe<T> Flowable<T> (*) Observable<T> Reactor Mono<Void> (*) Mono<T> (*) Flux<T> (*) (*) ... implements Publisher

© 2017 Pivotal Software, Inc. All rights reserved. java.util.concurrent.Flow in JDK 9 Reactive Streams JDK 9 org.reactivestreams java.util.concurrent Publisher Flow.Publisher Subscriber Flow.Subscriber Subscription Flow.Subscription Processor Flow.Processor

© 2017 Pivotal Software, Inc. All rights reserved. Reactive Streams ↔ j.u.c.Flow // Reactive Streams (Reactor) Publisher<String> pub = Flux.just("hello"); // java.util.concurrent.Flow Flow.Publisher<String> flow = JdkFlowAdapter.publisherToFlowPublisher(pub); // java.util.concurrent.Flow Flow.Publisher<String> flow = /* ... */; // Reactive Streams (Reactor) Flux<String> pub = JdkFlowAdapter.flowPublisherToFlux(flow);

© 2017 Pivotal Software, Inc. All rights reserved. Spring Framework 5.0 •Performance improvements •JDK 9 - HTTP/2 •Reactive Spring •Functional APIs •Kotlin Support https://speakerdeck.com/snicoll/spring-framework-5-dot-0-themes-and-trends

© 2017 Pivotal Software, Inc. All rights reserved. @Controller, @RequestMapping Spring MVC Servlet API Servlet Container Router functions Servlet 3.1, Netty, Undertow Spring WebFlux HTTP / Reactive Streams

© 2017 Pivotal Software, Inc. All rights reserved. Spring WebFlux @RestController  public class EchoController { @PostMapping("/echo") Flux<String> upperCase (@RequestBody Flux<String> body) {  return body.map(String::toUpperCase);  }  }

© 2017 Pivotal Software, Inc. All rights reserved. Returns Infinite-Stream @RestController  public class HelloController { @GetMapping Flux<Integer> infinite() { Stream<Integer> s = Stream.iterate(0, i -> i + 1);  return Flux.fromStream(s);  }}

© 2017 Pivotal Software, Inc. All rights reserved. Returns Infinite-Stream @RestController  public class TweetController { @GetMapping Flux<Tweet> infiniteTweet() { Stream<Tweet> s = Stream.iterate(0, i -> i + 1) .map(i -> new Tweet("hello" + i));  return Flux.fromStream(s);}}

© 2017 Pivotal Software, Inc. All rights reserved. Flux<Tweet> as a Server-Sent Events curl ... -H "Accept: text/event-stream" < HTTP/1.1 200 OK < Content-Type: text/event-stream < data: {"text":"hello0"} data: {"text":"hello1"} data: {"text":"hello2"} data: {"text":"hello3"} ...

© 2017 Pivotal Software, Inc. All rights reserved. Flux<Tweet> as a JSON Stream curl ... -H "Accept: application/stream+json" < HTTP/1.1 200 OK < Content-Type: application/stream+json < {"text":"hello0"} {"text":"hello1"} {"text":"hello2"} {"text":"hello3"} ...

© 2017 Pivotal Software, Inc. All rights reserved. WebClient •Reactive HTTP Client WebClient webClient = WebClient.create(); Mono<String> s = webClient.get() .uri("http://api.example.com") .exchange() .flatMap(res -> res.bodyToMono(String.class)); a nice RestTemplate alternative

© 2017 Pivotal Software, Inc. All rights reserved. WebClient •Reactive HTTP Client WebClient webClient = WebClient.create(); Mono<String> s = webClient.get() .uri("http://api.example.com") .retrieve() .bodyToMono(String.class);

© 2017 Pivotal Software, Inc. All rights reserved. WebClient •Reactive HTTP Client WebClient webClient = WebClient.create(); Flux<Tweet> tweets = webClient.get() .uri("http://api.example.com") .retrieve() .bodyToFlux(Tweet.class);

© 2017 Pivotal Software, Inc. All rights reserved. Router Functions RouterFunctions<ServerResponse> routes() {  return RouterFunctions.route( RequestPredicates.GET("/"), req -> ServerResponse.ok() .body(Mono.just("Hello","World"), String.class));  }

© 2017 Pivotal Software, Inc. All rights reserved. Router Functions RouterFunctions<ServerResponse> routes() {  return route( RequestPredicates.GET("/"), req -> ServerResponse.ok() .body(Mono.just("Hello","World"), String.class));  }

© 2017 Pivotal Software, Inc. All rights reserved. Router Functions RouterFunctions<ServerResponse> routes() {  return route(GET("/"), req -> ServerResponse.ok() .body(Mono.just("Hello","World"), String.class));  }

© 2017 Pivotal Software, Inc. All rights reserved. Router Functions RouterFunctions<ServerResponse> routes() { return route(POST("/echo"), req -> { Mono<String> body = req .bodyToMono(String.class) .map(String::toUpperCase); return ok().body(body, String.class)); });}

© 2017 Pivotal Software, Inc. All rights reserved. Spring MVC VS Spring WebFlux @GetMapping Flux<String> hello() {  return Flux.just("Hello") .delayElements( ofSeconds(1));  } @GetMapping String hello() { Thread.sleep(1000);  return"Hello";  } server.tomcat.max-threads=200 (default) https://gist.github.com/making/f32e81c5684a5fd810039854091dd793 Tomcat Netty

© 2017 Pivotal Software, Inc. All rights reserved. Concurrent users = 100 Transaction rate: 94.79 trans/sec Response time: 1.05 sec Live peek threads: 129 Transaction rate: 94.88 trans/sec Response time: 1.05 sec Live peek threads: 30

© 2017 Pivotal Software, Inc. All rights reserved. Spring Projects Reactor Ecosystem Spring Security Spring Data Spring Cloud Spring Integration Reactor Projects Reactor Netty Reactor Kafka Lettuce Thymeleaf

© 2017 Pivotal Software, Inc. All rights reserved. Spring Projects Reactor Ecosystem Spring Security Spring Data Spring Cloud Spring Integration Reactor Projects Reactor Netty Reactor Kafka Lettuce MongoDB Redis Cassandra Couchbase Thymeleaf

© 2017 Pivotal Software, Inc. All rights reserved. Reactive Repository public interface ReactiveCrudRepository<ID,T> { Mono<T> findById(ID id); Mono<T> findById(Mono<ID> id); Flux<T> findAll(); Mono<Long> count(); Mono<T> save(T entity); Mono<T> saveAll(Publisher<T> entityStream); Mono<Void> delete(T entity) // ... }

© 2017 Pivotal Software, Inc. All rights reserved. @Tailable for Infinite streams public interface PersonRepository extends ReactiveMongoRepository<Person,String> { @Tailable Flux<Person> findByFirstname(String firstname); }

© 2017 Pivotal Software, Inc. All rights reserved. What about JPA/JDBC? •JDBC is blocking ⌛ •JPA is blocking ⌛ 🤔 https://static.rainfocus.com/oracle/oow16/sess/1461693351182001EmRq/ppt/ CONF1578%2020160916.pdf https://www.voxxed.com/blog/2016/09/non-blocking-database-access/ •Non-Blocking JDBC in JDK 10?

© 2017 Pivotal Software, Inc. All rights reserved. Switching execution context BlockingRepository<User> repo = ...; Flux<User> users = Flux.defer(() -> Flux.fromIterable(repo.findAll())) .subscribeOn(Schedulers.elastic()); Make the subscription and request happen on a particular thread

© 2017 Pivotal Software, Inc. All rights reserved. Switching execution context Flux<User> users = ...; BlockingRepository<User> repo = ...; users.publishOn(Schedulers.elastic()) .doOneNext(u -> repo.save(u)) .then() ; Switch rest of the flux on a particular thread

© 2017 Pivotal Software, Inc. All rights reserved. Reactive Everywhere CF Java Client Firehose Nozzle Firehose Nozzle Cloud Foundry 💻 Firehose Doppler Endpoint WebSocket Rector Kafka (Publisher) Rector Kafka (Consumer) Reactor Netty Sever-Sent Event Spring WebFluxLog, Metrics

© 2017 Pivotal Software, Inc. All rights reserved. Thank you!! • Handson • https://github.com/reactor/lite-rx-api-hands-on • Slides • https://speakerdeck.com/simonbasle/reactor-3 • https://speakerdeck.com/sdeleuze/developing-reactive-applications-with- reactive-streams-and-java-8 • https://speakerdeck.com/snicoll/spring-framework-5-dot-0-themes-and- trends • https://speakerdeck.com/mp911de/reactive-spring • https://speakerdeck.com/christophstrobl/sneak-peek-on-spring-data-kay • https://speakerdeck.com/normanmaurer/netty-meetup-2017-san-francisco

Spring Framework 5.0による Reactive Web Application #JavaDayTokyo

More Related Content

What's hot

Viewers also liked

Similar to Spring Framework 5.0による Reactive Web Application #JavaDayTokyo

More from Toshiaki Maki

Recently uploaded

Spring Framework 5.0による Reactive Web Application #JavaDayTokyo