> For the complete documentation index, see [llms.txt](https://taejun.gitbook.io/tech/llms.txt). Markdown versions of documentation pages are available by appending `.md` to page URLs; this page is available as [Markdown](https://taejun.gitbook.io/tech/2-srping/webflux.md).
# WebFlux
## 스프링 코어 for Reactive
스프링 5.0 에서부터 리액티브 코드를 위한 여러가지 클래스들이 수정, 추가되었다.
### ReactiveAdapter, ReactiveAdapterRegistry
`RxJava, Reactor` 에서 사용하는 발행자 클래스를 `Publihser` 로 변환해주는 `Adapter` 가 `springframework.core` 에 추가되어 사용 가능해졌다.
아래처럼 `ReactiveAdapter` 를 상속받아 RxJava Maybe 와 Publisher 간의 변환 작업을 해주는 Adapter 를 작성해서 사용하거나
```
@Component
public class MaybeReactiveAdapter extends ReactiveAdapter {
public MaybeReactiveAdapter() {
/**
* Descriptor for a reactive type that can produce 0..1 values.
* @param type the reactive type
* @param emptySupplier a supplier of an empty-value instance of the reactive type
*/
super(ReactiveTypeDescriptor.singleOptionalValue(Maybe.class, Maybe::empty),
maybe -> ((Maybe) maybe).toFlowable(), // Maybe->Publisher
publisher -> Flowable.fromPublisher(publisher).singleElement()); // Publisher->Maybe
}
}
```
`ReactiveAdapterRegistry` 를 사용해 싱글턴 `Instance` 변수에 `Adapter` 용 코드를 작성해 필요할때 마다 꺼내어 쓸 수 있다.
```
@PostConstruct
public void init() {
ReactiveAdapterRegistry
.getSharedInstance()
.registerReactiveType(ReactiveTypeDescriptor.singleOptionalValue(Maybe.class, Maybe::empty),
maybe -> ((Maybe) maybe).toFlowable(), // Maybe->Publisher
publisher -> Flowable.fromPublisher(publisher).singleElement()); // Publisher->Maybe
}
...
ReactiveAdapter adapter = ReactiveAdapterRegistry
.getSharedInstance()
.getAdapter(Maybe.class);
...
```
### 리액티브 I/O, 코덱
`springframework.core.io` 에 저장된 `DataBuffer`, `DataBufferUtils` 를 사용하면 I/O 작업이 필요한 파일, 네트워크 자원으로 부터 리액티브 스트림 형태로 작업을 처리할 수 있다. `jav.nio.ByteBuffer` 클래스의 형변환 기능을 추가하여 보다 쉽게 사용 가능하다.
```
Flux reactiveHamlet = DataBufferUtils.read(
new DefaultResourceLoader().getResource("hamlet.txt"),
new DefaultDataBufferFactory(),
1024);
```
`springframework.core.codec` 에 정의된 인터페이스 `Encoder`, `Decoder` 를 사용하면 `Non Blocking` 방식으로 직렬화 데이터를 자바객체, 자바객체를 직렬화 데이터로 변환 가능하다.
## WebFlux
`Sprinb Boot 2` 에 리액티브 웹서버를 위한 `WebFlux` 모델을 사용할 수 있도록 `spring-boot-starter-webflux` 라는 새로운 패키지를 추가할 수 있게 되었다.
해당 모듈은 `Reactive Stream Adapter` 위에 구축된며 `Servlet 3.1+ 지원서버(Tomcat, Jetty 등)`, `Netty`, `Undertow` 서버엔진에서 모두 지원한다.
> 위의 엔진들은 `java 8` 에 추가된 `java NIO` 로 구현되어 Http 요청을 논블럭킹으로 처리한다.
일반적은 `WebMVC` 모듈도 `Spring 5.0` 에 이르러 `spring-boot-starter-web` `Servlet 3.1` 을지원하면서 일부분은 리액티브 스트림을 지원하게 되었다.
`ResponseBodyEmitterReturnValueHandler` 클래스가 업그레이드 되면서 `ReactiveTypeHandler` 필드를 사용해 `WebMVC` 의 인프라 구조를 크게 해치지 않고 `컨트롤러 메서드`가 반환하는 `Flux, Mono, Flowable` 등의 `Publisher`(리액티브 스트림)을 처리한다.
물론 서블릿 API 를 사용하기에 `블록킹/스레드풀` 방식을 사용한다.
### WebFlux 개요
기존 `WebMVC` 모델 구조는 아래와 같다.
> `ViewResolver` 는 `Rest` 방식에선 생략된다.
각종 서블릿 컨테이너(`Tomcat, JBoss` 등) 요청을 서블릿 클래스를 상속한 `DispatcherServlet` 이 스프링 부트 컨트롤러 매핑에 따라 요청을 분배한다.
그림처럼 기존 `WebMVC` 방식은 `동기/블로킹` 방식으로 동작한다.
스프링 부트에서 리액티브 방식을 사용하려면 `Servlet 3.1+(Tomcat, Jetty 등)`, `Netty`, `Undertow` 와 같은 서버를 사용해 리액티브하게 구조가 변경되어야 하는데
다행이도 스프링 프로젝트팀이 동일한 어노테이션 기반 프로그래밍 모델을 사용하면서 `비동기/논블록킹` 으로 동작하도록 이미 개발해두었다.
#### WebFlux with Flux
대략적으로 `WebFlux` 에서 Http Request, Response 어떻게 리액티브로 구현했는지 아래 인터페이스로 확인할 수 있다.
```
interface ServerHttpRequest { // Http Request 를 객체로 표현
Flux getBody();
...
}
interface ServerHttpResponse { // Http Response 를 객체로 표현
Mono writeWith(Publihser body);
...
}
interface ServerWebExchange { // HTTP Request, Response 컨테이너 역할
ServerHttpRequest getRequest();
ServerHttpResponse getResponse();
Mono getSession();
...
}
```
서블릿의 `ServletRequest`, `ServletResponse` 와 연관지어서 새로운 Http 요청, 반환을 객체로 표현할 수 있는 인터페이스들이 정의되어있고
`DataBuffer` 를 사용해 리액티브 타입과의 결합도를 낮춘다.
```
interface WebHandler {
Mono handle(ServerWebExchange exchanage);
}
interface WebFilterChain {
Mono filter(ServerWebExchange exchanage);
}
interface WebFilter {
Mono filter(ServerWebExchange exchanage, WebFilterChain chain);
}
```
`WebHandler` 는 `DispatcherServlet` 역할, 실행결과를 받지 못할 수 있음으로 `handle` 메서드는 `Mono` 가 반환된다.
> 클라이언트를 위한 Http 반환은 `exchange` 안의 `ServerHttpResponse` 에
`WebFilter` 는 서블릿의 요청, 반환 필터처럼 리액티브에서도 비지니스 로직에 집중할 수 있도록 필터기능이 제공된다.
```
public interface HttpHandler {
Mono handle(ServerHttpRequest request, ServerHttpResponse response);
}
```
마지막으로 `WebHandler` 로부터 전달받은 `exchange` 객체를 `url` 매핑할 수 있도록 `HttpHandler` 의 `handle` 로 전달된다.
### WebFlux - Functional Reactive Web Server
`Vert.x` 나 `Ratpack` 과 같은 프레임워크의 인기비결은 스프링의 복잡한 MVC 설정으로 라우팅 설정과 로직이 없이\
간결한 설정으로 라우팅 로직을 작성할 수 있는 API 들이 잘 정의되어 있기 때문이다.
스프링 프레임워크도 위 프레임워크처럼 간결하게 라우팅 로직을 처리할 수 있는 API 를 개발하였다.
`spring-boot-starter-webflux` 모듈의 `org.springframework.web.reactive.function.server` 패키지에 정의된 `RouterFunction` 클래스 사용하여 라우팅 로직 정의가 가능하다.
```
import static org.springframework.web.reactive.function.server.RequestPredicates.*;
import static org.springframework.web.reactive.function.server.RouterFunctions.*;
import static org.springframework.http.MediaType.APPLICATION_JSON;
@SpringBootApplication
public class DemoApplication {
final ServerRedirectHandler serverRedirectHandler = new ServerRedirectHandler();
public static void main(String[] args) {
SpringApplication.run(DemoApplication.class, args);
}
@Bean
public RouterFunction routes(OrderHandler orderHandler) {
return nest(path("/orders"),
nest(accept(APPLICATION_JSON),
route(GET("/{id}"), orderHandler::get)
.andRoute(method(HttpMethod.GET), orderHandler::list))
.andNest(contentType(APPLICATION_JSON),
route(POST("/"), orderHandler::create))
.andNest((serverRequest) -> serverRequest.cookies().containsKey("Redirect-Traffic"),
route(all(), serverRedirectHandler))
);
}
}
```
`RouterFunction` 을 스프링 `Bean` 으로 등록하고 `/orders` 에 대한 라우팅 로직 설정, 기본적인 `Path, Http method, cookie` 포함여부 등 여러 로직처리 가능하다.
함수형으로 웹서버를 구축할경우 `Netty` 서버를 같이 사용하면 별도의 스프링 설정없이 서버 실행이 가능하다.
이런점때문에 단순 테스트의 경우 `@SpringBootApplication` 을 사용하지 않고 단순 `Netty` 서버를 사용해 빠르게 서버를 실행하고 구현한 메서드들을 테스트 할 수 있다.
만약 패스워드 암호화 및 복호화 테스트를 한다면 서버기능을 하는 객체 외에 추가적으로 필요한 객체는 해시 기능이 있는 `spring-boot-starter-security` 의 `PasswordEncoder` 뿐이다.
> `PasswordEncoder` 만 사용한다면 `spring-security-core` 만 의존성 처리해도 된다.
별도의 스프링 관련 어노테이션, Bean 등록과정 없이 `RouterFunction, Netty, PasswordEncoder` 3개 객체만 잘 정의해서 아래처럼 서버 실행이 가능하다.
```
import static org.springframework.web.reactive.function.server.RequestPredicates.*;
import static org.springframework.web.reactive.function.server.RouterFunctions.*;
public static void main(String... args) {
long start = System.currentTimeMillis();
BCryptPasswordEncoder passwordEncoder = new BCryptPasswordEncoder(18); // encoder 생성
HttpHandler httpHandler = RouterFunctions.toHttpHandler( // RouterFunction -> HttpHandler 변경
route(POST("/password"), request -> request // RouterFunction 으로 라우터 로직 생성
.bodyToMono(PasswordDTO.class)
.map(p -> passwordEncoder.matches(p.getRaw(), p.getSecured()))
.flatMap(isMatched -> isMatched
? ServerResponse.ok().build()
: ServerResponse.status(HttpStatus.EXPECTATION_FAILED).build())
)
);
ReactorHttpHandlerAdapter reactorHttpHandler = new ReactorHttpHandlerAdapter(httpHandler);
// HandlerAdapter 에 HttpHandler 삽입, BiFunction 를 구현한 클래스임
DisposableServer server = HttpServer.create() // Netty Server
.host("localhost").port(8080)
.handle(reactorHttpHandler) // BiFunction> 요구함
.bindNow(); // 서버 엔진 시작
LOGGER.debug("Started in " + (System.currentTimeMillis() - start) + " ms"); // Started in 703 ms
server.onDispose().block(); // main 스레드 차단
}
```
서버가 0.7 초만에 실행된다.\
스프링 컨테이너, 의존성 주입, 어노테이션 처리를 하지 않음으로 속도가 굉장히 빠르며\
간단한 테스트진행은 위와같은 방식으로 진행하면 편하다.
### WebFlux - Annotated Controller
`RouterFunctions`를 사용해도 되지만 `WebMVC` 모델에서 사용하는 `@RestController, @RequestMapping` 등의 어노테이션을 `WebFlux` 에서도 사용할 수 있다.
```
@RestController
@RequestMapping("/member")
@RequiredArgsConstructor
public class MemberController {
private final MemberRepository memberRepository;
@GetMapping
public Flux getAll() {
return memberRepository.findAll();
}
@GetMapping("/id/{id}")
public Mono getById(@PathVariable Long id) {
return memberRepository.findById(id);
}
}
```
`Flux` 는 배열, `Mono` 는 객체로 반환된다.
### WebFlux - Filter
더이상 `javax.servlet.Filter` 을 사용하지 못한다.
필터기능을 하는 방법은 여러가지다.
#### RouterFunctions
```
@SpringBootApplication
public class ReactApplication {
public static void main(String[] args) {
SpringApplication.run(ReactR2dbcApplication.class, args);
}
@Bean
public RouterFunction filterFunction(MemberComponent memberComponent) {
return RouterFunctions
.route(GET("/member/{memberId}")
.and(accept(MediaType.APPLICATION_JSON)), memberComponent::getById)
.filter(new ExampleHandlerFilterFunction());
}
}
@Component
@RequiredArgsConstructor
class MemberComponent {
private final MemberRepository memberRepository;
public Mono getById(ServerRequest request) {
return ServerResponse.ok().contentType(MediaType.TEXT_PLAIN)
.body(BodyInserters.fromValue(
memberRepository.findById(Long.valueOf(request.pathVariable("memberId")))));
}
}
```
#### WebFilter
```
@Component
public class ExampleWebFilter implements WebFilter {
@Override
public Mono filter(ServerWebExchange serverWebExchange,
WebFilterChain webFilterChain) {
serverWebExchange.getResponse().getHeaders().add("web-filter", "web-filter-test");
return webFilterChain.filter(serverWebExchange);
}
}
```
별도의 매핑 조건이 없기때문에 조건문 분기가 필요함
#### HandlerFilterFunction
```
public class ExampleHandlerFilterFunction implements HandlerFilterFunction {
@Override
public Mono filter(ServerRequest request, HandlerFunction handlerFunction) {
if (request.pathVariable("name").equalsIgnoreCase("test")) {
return ServerResponse.status(HttpStatus.FORBIDDEN).build();
}
return handlerFunction.handle(request);
}
}
```
### WebFlux - Exception Handler
**메서드 레벨**에서 오류처리는 `ServerResponse` 에 status, body 등을 설정하면 쉽게 처리할 수 있다.\
또한 클래스 내부에서 기존 WebMVC 에서 사용하던 `@ExceptionHandler` 를 사용해 처리할 수 있다.
```
@Controller public class SimpleController {
@ExceptionHandler public ResponseEntity handle(IOException ex) {
// ...
}
}
```
**글로벌 레벨**에서 오류처리는 `WebExceptionHandler` 인터페이스를 구현해 필터 방식으로 처리할 수 있다.
#### DefaultErrorAttributes
`DefaultErrorAttributes` 는 `WebFlux` 에서 사용하는 에러 핸들러로 기본 필터로 등록되어 있는 에러 핸들러가 `DefaultErrorAttributes` 안의 `getErrorAttributes` 를 호출해 아래와 같은 반환값을 반환한다.
```
{
"timestamp": "...",
"path": "...",
"status": 405,
"error": "Method Not Allowed",
"message": "",
"requestId": "ca35d584-1"
}
```
글로벌 레벨에서 오류처리를 통해 커스텀한 반환값을 설정하고 싶으면 `DefaultErrorAttributes` 의 `getErrorAttributes` 메서드를 오버라이딩 해야한다.
```
@Slf4j
@Component
@RequiredArgsConstructor
public class GlobalErrorAttributes extends DefaultErrorAttributes {
// private final MessageSource messageSource;
@Override
public Map getErrorAttributes(ServerRequest request, ErrorAttributeOptions options) {
Throwable throwable = getError(request);
String acceptLanguage = request.headers().firstHeader("accept-language");
Locale locale = acceptLanguage != null && acceptLanguage.startsWith("ko") ? Locale.KOREA : Locale.getDefault();
log.error("unknown server error:{}, {}" + throwable.getClass().getCanonicalName(), throwable.getMessage());
// throwable 의 종류에 맞게 반환값을 조정
Map map = new HashMap<>();
map.put("code", UNKNOWN_ERROR_CODE);
map.put("error", UNKNOWN_ERROR_TYPE);
// map.put("description", messageSource.getMessage("fms.error.unknown_server_error", null, locale));
return map;
}
}
```
에러 반환시 아래처럼 출력되도록 설정
```
{
"code": "500-00",
"error": "UnknownServerError"
}
```
이제 핸들러에서 `DefaultErrorAttributes` 의 `getErrorAttributes` 가 아닌\
직접 정의한 `GlobalErrorAttributes` 의 `getErrorAttributes` 가 호출되도록 설정하면 된다.
#### AbstractErrorWebExceptionHandler
`AbstractErrorWebExceptionHandler` 는 에러발생시 필터로 등록되어 있는 핸들러\
해당 핸들러보다 더 높은 우선순위를 가진 핸들러로 에러처리하도록 설정
```
@Order(-2) // 기본 에러처리는 -1, 보다 빠르게 설정한다.
@Component
public class GlobalErrorWebExceptionHandler extends AbstractErrorWebExceptionHandler {
// constructors
public GlobalErrorWebExceptionHandler(
GlobalErrorAttributes errorAttributes, WebProperties.Resources resources,
ApplicationContext applicationContext, ServerCodecConfigurer configurer) {
super(errorAttributes, resources, applicationContext);
this.setMessageWriters(configurer.getWriters());
}
@Override
protected RouterFunction getRoutingFunction(ErrorAttributes errorAttributes) {
return RouterFunctions.route(RequestPredicates.all(), this::renderErrorResponse);
}
private Mono renderErrorResponse(ServerRequest request) {
Map errorPropertiesMap = getErrorAttributes(request, ErrorAttributeOptions.defaults());
Integer status = Integer.valueOf(((String) errorPropertiesMap.get("code")).substring(0, 3));
return ServerResponse.status(status)
.contentType(MediaType.APPLICATION_JSON)
.body(BodyInserters.fromValue(errorPropertiesMap));
}
}
```
### WebFlux - WebSocket
이미 `spring-boot-starter-websocket` 모듈에서 제공하는 스프링 웹소켓을 통해 논블록킹으로 메세지를 처리할 수 있을 줄 알았지만 내부에서 블로킹 방식으로 동작하며 리액티브 서버 성능에 영향을 끼친다
`WebFlux` 에선 비동기/논블록킹 방식의 웹소켓 처리를 위해 `org.springframework.web.reactive.socket` 패키지를 제공한다.
> 패키지에 `reactive` 가 붙을뿐 클래스명이나 사용방법이 유사하다.
웹소켓 서버, 웹소켓 클라이언트 모두 제공한다.
웹소켓 서버는 `WebSocketHandler` 를 사용해 소켓 핸들러 역할을 하는 객체인 `Handler` 를 등록한다.
```
public class EchoWebSocketHandler implements WebSocketHandler {
@Override
public Mono handle(WebSocketSession session) {
return session
.receive() // return Flux
.map(wsMessage -> wsMessage.getPayloadAsText()) // websocket 메세지의 payload (string) 흭득
.map(tm -> "Echo: " + tm) // 문자열 변환
.map(tm -> session.textMessage(tm)) // WebsocketSession 을 사용, client 보낼 메세지(payload) 작성
.as(wsMessage -> session.send(wsMessage)); // client 에게 메세지(payload) 전송
}
}
```
웹소켓을 사용하려면 먼저 웹소켓 설정 객체를 `Bean` 으로 등록해야 한다.\
위에서 정의한 핸들러를 `url` 에 매핑하고, Request 요청을 Upgrade 하는 어뎁터를 `Bean` 으로 등록한다.
```
@Configuration
public class WebSocketConfiguration {
@Bean
public HandlerMapping handlerMapping() {
SimpleUrlHandlerMapping mapping = new SimpleUrlHandlerMapping();
mapping.setUrlMap(Collections.singletonMap("/ws/echo", new EchoWebSocketHandler())); // 경로기반 매핑 설정
mapping.setOrder(-1); // 우선순위, 생략시 사용하지 않는 것으로 설정됨
return mapping;
}
@Bean
public HandlerAdapter handlerAdapter() {
return new WebSocketHandlerAdapter();
// WebSocket Handshake (upgrade request) 를 처리하는 HandlerAdapter 생성
}
}
```
> `WebSocketMessage` 는 `payload` 로 `DataBuffer` 를 구현하여 문자열, 바이트코드로 쉽게 형변환 가능
>
> ```
> public class WebSocketMessage {
> private final Type type;
> private final DataBuffer payload;
> ...
> }
> ```
`WebSocketHandler` 를 구현하면 해당 url 에 해당하는 `WebSocketSession` 객체를 사용해 메세지를 받고 보낸다.\
웹소켓 테스트 툴을 사용해 `ws://127.0.0.1:8080/ws/echo` 로 접속, 메세지 전송시 `Echo: ...` 메세지 수신 확인
#### WebSocket Client
웹소켓 클라이언트의 경우 `WebSocketClient` 인터페이스를 구현한 `ReactorNettyWebSocketClient` 를 사용한다.
```
import org.springframework.web.reactive.socket.client.WebSocketClient;
...
@Bean
public CommandLineRunner commandLineRunner() {
return (args) -> {
ReactorNettyWebSocketClient client = new ReactorNettyWebSocketClient();
client.execute(URI.create("http://localhost:8080/ws/echo"),
session -> Flux
.interval(Duration.ofMillis(100))
.map(String::valueOf)
.map(session::textMessage)
.as(session::send))
.subscribe();
};
}
```
정의해둔 `echo` 웹소켓 서버에 0.1 초마다 `interval` 데이터를 문자열로 전송
`client` 역시 `WebsocketHandler` 구현체를 `excute` 함수의 매개변수로 받으며 핸들러 매핑과 핸들러 어뎁터가 `Bean` 으로 등록되지 않을 뿐 웹소켓 서버 코드와 유사하다.
> 참고 코드:
> 안타깝게도 `WebFlux` 에서 제공하는 웹소켓 내용은 위의 기능이 전부이며 STOMP 를 사용한 메세지 매핑 등의 기능은 제공하지 않는다.
#### 다중 통신 Sinks.Many
> 코드 참고:
리액티브의 웹소켓은 WebMVC 에서 사용한 웹소켓과 다른점이 있는데 모든 요청에 대한 응답을 수행할 때 위와같이 람다식이나 함수를 미리 등록해두어야 하고\
두개 이상의 클라이언트들간 통신을 위해서 각 클라이언트의 `session` 을 찾아 데이터를 발행하는데 `Sinks.Many` 클래스를 사용한다.
웹소켓 설정 객체를 `Bean` 으로 등록하고 핸들러 매핑하는 것은 동일하다.
```
@Configuration
public class WebSocketConfig {
@Bean
public HandlerMapping handlerMapping(ChatSocketHandler chatSocketHandler) { // url, handler 매핑
SimpleUrlHandlerMapping mapping = new SimpleUrlHandlerMapping();
mapping.setUrlMap(Collections.singletonMap("/ws/chat", chatSocketHandler));
mapping.setOrder(-1);
return mapping;
}
@Bean
public WebSocketHandlerAdapter handlerAdapter() {
return new WebSocketHandlerAdapter();
}
}
@Slf4j
@Component
@RequiredArgsConstructor
public class ChatSocketHandler implements WebSocketHandler { // handler 정의
private final ObjectMapper mapper;
@Override
public Mono handle(WebSocketSession session) {
WebSocketMessageSubscriber subscriber = new WebSocketMessageSubscriber(session);
return session.receive()
.map(this::toDto)
.doOnNext(subscriber::onNext) // 수신 콜백 함수 등록
.doOnError(subscriber::onError) // 에러 콜백 함수 등록
.doOnCancel(subscriber::onCancel) // 연결끊김 콜백 함수 등록
.zipWith(session.send(subscriber.getMany().asFlux().map(webSocketToClientDto -> // 메세지 발신 콜백 함수 등록
session.textMessage(webSocketToClientDto.getFrom() + ":" + webSocketToClientDto.getMessage()))))
.then();
}
private WebSocketFromClientDto toDto(WebSocketMessage message) {
try {
WebSocketFromClientDto WsDto = mapper.readValue(message.getPayloadAsText(), WebSocketFromClientDto.class);
return WsDto;
} catch (JsonProcessingException e) {
e.printStackTrace();
return null;
}
}
}
```
`session.send` 에 등록되는 `Flux` 발행자는 `Sinks.Many` 로부터 생성되는데\
이는 아래에서 사용하는 코드와 같이 메세지를 집어넣을 수 있는 발행자다.\
Websocket 외에도 SSE (Server Send Events) 에서도 사용된다.
```
@Slf4j
class WebSocketMessageSubscriber {
// 본인을 포함한 다른 클라이언트의 웹소켓 메세지 발행자 맵
public static Map> userMap = new HashMap<>(); //sessionId, sink
private final String id;
@Getter
private final Sinks.Many many; // 본인의 웹소켓 메세지 발행자
public WebSocketMessageSubscriber(WebSocketSession session) {
many = Sinks.many().unicast().onBackpressureBuffer();
id = session.getId();
many.tryEmitNext(WebSocketToClientDto.builder().from("system").message("welcome, " + id).build());
userMap.put(id, many);
}
public void onNext(WebSocketFromClientDto msg) {
log.info("onNext invoked, to:{}, msg:{}", msg.getTo(), msg.getMessage());
Sinks.Many to = userMap.get(msg.getTo());
if (to == null)
many.tryEmitNext(WebSocketToClientDto.builder().from("system").message("no user:" + msg.getTo()).build());
else
to.tryEmitNext(WebSocketToClientDto.builder().from(id).message(msg.getMessage()).build());
}
public void onError(Throwable error) {
//TODO log error
log.error("onError invoked, error:{}, {}", error.getClass().getSimpleName(), error.getMessage());
many.tryEmitNext(WebSocketToClientDto.builder()
.from("system")
.message(id + " on error, error:" + error.getMessage())
.build());
}
public void onCancel() {
log.info("onCancel invoked, id:{}", id);
userMap.remove(id);
for (Map.Entry> entry : userMap.entrySet()) {
if (!entry.getKey().equals(id))
entry.getValue().tryEmitNext(WebSocketToClientDto.builder()
.from("system")
.message(id + " is exit")
.build());
}
}
}
@Getter
@Setter
@Builder
class WebSocketToClientDto {
private String from;
private String message;
}
@Getter
@Setter
class WebSocketFromClientDto {
private String to;
private String message;
}
```
### WebClient
논블록킹 `Http Client`로 기존 스프링 부트에서 대표적인 `Http Client` 로 `RestTemplate`(블록킹) 이 있다.\
내부에 `Flux, Mono` 리액터 객체를 지원하는 매핑이 내장되어 있어 리액티브 서버에 잘 어울린다.
`http://localhost:8080/api/user/{id}` url 을 지원하는 간단한 웹서버 생성
```
@Slf4j
public class TestWebServer {
public static void main(String[] args) {
HttpHandler httpHandler = RouterFunctions.toHttpHandler( // RouterFunction -> HttpHandler 변경
nest(path("/api"), route(GET("/users/{id}"),
request -> {
String id = request.pathVariable("id");
return ServerResponse.ok().syncBody("hello " + id + " user!"); // 반환데이터 동기적으로 생성
}) // end route
) // end nest
);
ReactorHttpHandlerAdapter reactorHttpHandler = new ReactorHttpHandlerAdapter(httpHandler);
DisposableServer server = HttpServer.create()
.host("localhost").port(8080)
.handle(reactorHttpHandler)
.bindNow();
server.onDispose().block();
}
}
```
`WebClient` 를 사용해 위 `url` 에 `Http GET Request` 요청
```
public class TestWebClient {
public static void main(String[] args) throws InterruptedException {
WebClient.create("http://localhost:8080/api") // WebClient 객체 생성 + baseUrl 설정
.get().uri("/users/{id}", 10) // method, uri 설정
.retrieve() // 응답 내용 설정. ResponseSpec 반환
.bodyToMono(String.class) // 응답 body 를 Mono 로 변환
.subscribe(s -> System.out.println(s)); // Mono 에 대한 구독 설정
// hello 10 user!
Thread.sleep(1000); // main thread 종료 방지
}
}
```
위의 `WebClient` 는 `GET` 방식이라 `uri` 만 설정했지만 `API` 에 따라 `cookie, header, body` 모두 설정 가능하다.
HTTP 응답을 처리할 수 있는 메서드가 `retrieve()` 와 `exchage()` 가 있는데
`retrieve()` 는 `ResponseSpec` 을 반환하고 `exchage()` 는 `Mono` 를 반환한다.
만약 `exchange()` 를 사용할 경우 아래와 같이 코드작성
```
public static void main(String[] args) throws InterruptedException {
WebClient.create("http://localhost:8080/api") // WebClient 객체 생성
.get().uri("/users/{id}", 10) // method, uri 설정
.exchange() // Mono 반환
.flatMap(response -> response.bodyToMono(String.class)) // 응답 body 를 Mono 로 변환
.subscribe(s -> System.out.println(s)); // Mono 에 대한 구독 설정
Thread.sleep(1000);
}
```
`exchage` 를 사용하면 `ClientResponse` 에서 제공하는 `Http Response` 의 각종 정보를 조작할 수 있는 여러 메서드로 복잡한 반환 로직 구성이 가능하다.\
`retrieve` 의 경우 `Http status` 만 겨우 조작하여 `DSL` 형식으로 처리할 수 있다.
`WebClient` 는 인터페이스이고 `DefaultWebClient` 가 `WebClient` 의 유일한 구현체이다.\
실제 `DefaultWebClient` 내부에선
#### WebClient Serialize config
`WebClient` 에서 직렬화, 비직렬화를 수행할때 기존생성한 `ObjectMapper` 를 통해 처리할 수 잇다.
```
@Bean
public ObjectMapper objectMapper() {
JavaTimeModule module = new JavaTimeModule();
LocalDateTimeSerializer localDateTimeSerializer = new LocalDateTimeSerializer(DateTimeFormatter.ofPattern("yyyy-MM-dd'T'HH:mm:ss"));
LocalDateTimeDeserializer localDateTimeDeserializer = new LocalDateTimeDeserializer(DateTimeFormatter.ofPattern("yyyy-MM-dd'T'HH:mm:ss"));
module.addSerializer(LocalDateTime.class, localDateTimeSerializer);
module.addDeserializer(LocalDateTime.class, localDateTimeDeserializer);
ObjectMapper objectMapper = new ObjectMapper();
objectMapper.registerModule(module);
objectMapper.disable(SerializationFeature.WRITE_DATES_AS_TIMESTAMPS);
// UnrecognizedPropertyException 처리
objectMapper.configure(DeserializationFeature.FAIL_ON_UNKNOWN_PROPERTIES, false);
// json -> clas 에서 unknown 속성이 있어도 처리
objectMapper.setSerializationInclusion(JsonInclude.Include.NON_NULL); // null 이 아닌것만 변환
objectMapper.setPropertyNamingStrategy(PropertyNamingStrategy.SNAKE_CASE); // snake case 로 변환
return objectMapper;
}
@Bean
public WebClient webClient(ObjectMapper objectMapper) {
ExchangeStrategies jacksonStrategy = ExchangeStrategies.builder()
.codecs(config -> {
config.defaultCodecs().jackson2JsonEncoder(new Jackson2JsonEncoder(objectMapper, MediaType.APPLICATION_JSON));
config.defaultCodecs().jackson2JsonDecoder(new Jackson2JsonDecoder(objectMapper, MediaType.APPLICATION_JSON));
config.defaultCodecs().maxInMemorySize(1024 * 1024 * 50);
}).build();
return WebClient.builder().exchangeStrategies(jacksonStrategy).build();
}
```
주의사항으로 `uri(uriBuilder->..)` 메서드를 사용해 `query parameter` 를 지정할 경우 문자열에 `/` 가 들어갈 `escape` 문자로 인식하기 때문에 base64 문자열로 변환할 수 없다, url encoding 을 진행하지 않는다. 또한 `WebClient` 생성시 `baseUrl` 을 설정하지 않으면 `uribuilder` 를 통해 `scheme, host, port, path` 빌드함수를 모두 호출해야 하기 때문에 번거롭다.
`WebClient` 를 `bean` 으로 생성해 `singleton` 방식으로 사용한다면 `StringBuilder` 를 통해 `uri` 를 직접생성하는 것을 권장.
```
StringBuilder urlBuilder = new StringBuilder(WEATHER_GET_ULTRA_SRT_NCST); /*URL*/
urlBuilder.append("?" + URLEncoder.encode("ServiceKey", "UTF-8") + "=" + URLEncoder.encode(dataGovApiKey, "UTF-8")); /*공공데이터포털에서 받은 인증키*/
urlBuilder.append("&" + URLEncoder.encode("pageNo", "UTF-8") + "=" + URLEncoder.encode("1", "UTF-8")); /*페이지번호*/
urlBuilder.append("&" + URLEncoder.encode("numOfRows", "UTF-8") + "=" + URLEncoder.encode("10", "UTF-8")); /*한 페이지 결과 수*/
urlBuilder.append("&" + URLEncoder.encode("dataType", "UTF-8") + "=" + URLEncoder.encode("JSON", "UTF-8")); /*요청자료형식(XML/JSON)Default: XML*/
urlBuilder.append("&" + URLEncoder.encode("base_date", "UTF-8") + "=" + URLEncoder.encode(LocalDate.now().format(DateTimeFormatter.BASIC_ISO_DATE), "UTF-8")); /*15년 12월 1일발표*/
urlBuilder.append("&" + URLEncoder.encode("base_time", "UTF-8") + "=" + URLEncoder.encode(String.format("%02d00", curHour), "UTF-8")); /*05시 발표*/
urlBuilder.append("&" + URLEncoder.encode("nx", "UTF-8") + "=" + URLEncoder.encode(String.valueOf(nx), "UTF-8")); /*예보지점 X 좌표값*/
urlBuilder.append("&" + URLEncoder.encode("ny", "UTF-8") + "=" + URLEncoder.encode(String.valueOf(ny), "UTF-8")); /*예보지점의 Y 좌표값*/
URI uri = new URL(urlBuilder.toString()).toURI();
```
### SSE(Server-Sent Event)
> 참고:
> 평범한 `HTTP, Websocket, SSE 프로토콜`의 시퀀스 비교이다.\
> 
서버 단방향 통신이라 웹소켓이 비해 속도나 오버헤드 측면에서 SSE 가 효율적이지만 양방향이 안되는 이유로 웹소켓이 주로 사용된다.
`WebFlux` 를 사용하지 않고 `spring-boot-starter-web` 에서 SSE 프로토콜 사용이 가능하다.
```
@Component
@RequiredArgsConstructor
public class TemperatureSensor {
// 스프링 제공 이벤트 발행 구독 지원 클래스
private final ApplicationEventPublisher publisher;
private final ScheduledExecutorService executor = Executors.newSingleThreadScheduledExecutor();
private final Random rnd = new Random();
@PostConstruct
public void startProcessing() {
this.executor.schedule(this::probe, 1, SECONDS); // 데이터 발행 시작
}
private void probe() {
double temperature = 16 + rnd.nextGaussian() * 10;
publisher.publishEvent(new Temperature(temperature)); // 이벤트 데이터 발행
executor.schedule(this::probe, rnd.nextInt(5000), MILLISECONDS); // 5초후 재발행
}
}
```
`ScheduledExecutorService` 와 `ApplicationEventPublisher` 클래스를 사용해 `Temperature` 데이터를 계속 발행한다.\
`@EventListener` 를 사용해 `ApplicationEventPublisher` 에 발행된 데이터를 받을 수 있다.
```
@Slf4j
@RestController
public class TemperatureController {
static final long SSE_SESSION_TIMEOUT = 5 * 1000L;
// 연결 클라이언트 관리 list
private final Set clients = new CopyOnWriteArraySet<>();
// TemperatureSensor 의 ApplicationEventPublisher 에서 발행되는 데이터 대응
@Async
@EventListener
public void handleMessage(Temperature temperature) {
log.info(format("Temperature: %4.2f C, active subscribers: %d", temperature.getValue(), clients.size()));
// 관리되는 클라이언트에게 발행된 Temperature 데이터 전달 및 예외발생 클라이언트 삭제처리
List deadEmitters = new ArrayList<>();
clients.forEach(emitter -> {
try {
Instant start = Instant.now();
emitter.send(temperature, MediaType.APPLICATION_JSON);
log.info("Sent to client, took: {}", Duration.between(start, Instant.now()));
} catch (Exception ignore) {
deadEmitters.add(emitter);
}
});
clients.removeAll(deadEmitters);
}
@RequestMapping(value = "/temperature-stream", method = RequestMethod.GET)
public SseEmitter events(HttpServletRequest request) {
log.info("SSE stream opened for client: " + request.getRemoteAddr());
SseEmitter emitter = new SseEmitter(SSE_SESSION_TIMEOUT); // 5 초간 연결
clients.add(emitter); // 관리 emitter 목록에 추가
// Remove SseEmitter from active clients on error or client disconnect
emitter.onTimeout(() -> clients.remove(emitter));
emitter.onCompletion(() -> clients.remove(emitter));
return emitter;
}
// 위에 설정된 5초가 지나면 AsyncRequestTimeoutException 이 발생하고 호출됨.
@ExceptionHandler(value = AsyncRequestTimeoutException.class)
public ModelAndView handleTimeout(HttpServletResponse rsp) throws IOException {
log.warn("handle timeout");
if (!rsp.isCommitted()) {
rsp.sendError(HttpServletResponse.SC_SERVICE_UNAVAILABLE);
}
return new ModelAndView();
}
}
```
`SseEmitter` 를 지속적으로 유지, 관리한다. 반환 타입이 `ResponseEntity, Map` 과 같은 객체가 아닌 `SseEmitter` 인 것이 어색하다.
`WebFlux` 에선 `Flux` 와 `Spring 5.0` 에 추가된 `ServerSentEvent` 를 사용해 SSE 프로토콜을 지원한다.
```
@RestController
@RequiredArgsConstructor
public class ServerSentController {
private final StocksService stocksService;
@GetMapping("/sse/stocks")
public Flux> streamStocks() {
return stocksService.stream() // Flux 반환
.map(item -> ServerSentEvent
.builder(item)
.event("StockItem")
.id(item.getId())
.build());
}
@GetMapping("/sse/stocks2")
public Flux streamStocks2() {
return stocksService.stream();
}
}
```
`WebFlux` 내부에서 발행원소를 `ServerSentEvent` 로 래핑 하기에 단순 `Flux<>` 만 컨트롤러 메서드에서 반환해도 된다.
## 스프링 시큐리티 with WebFlux
기존 서블릿 기반 스프링 부트는 하나의 스레드에 하나의 연결이 처리되어\
`ThreadLocal` 에 `SecurityContext` 를 저장해 연결동안 보안처리를 진행했지만
리액티브는 하나의 연결에 여러개의 스레드가 꼬여있을 수 있어 `Reactor Context` 를 사용해야 한다.\
`spring-boot-starter-security` 모듈 역시 기존 서블릿 기반 `WebMVC` 에서 `WebFlux` 를 지원할 수 있도록 업데이트 되었다.
### ReactiveSecurityContextHolder
`WebMVC` 에선 `SecurityContextHolder` 에서 `SecurityContext` 를 가져왔다면\
`WebFlux` 에선 `ReactiveSecurityContextHolder` 에서 `SecurityContext` 를 가져온다.
```
@RestController
@RequiredArgsConstructor
@RequestMapping("/api/v1")
public class SecuredProfileController {
private final ProfileService profileService;
@GetMapping("/profiles")
public Mono getProfile() {
return ReactiveSecurityContextHolder.getContext() // Mono 반환
.map(SecurityContext::getAuthentication)
.flatMap(auth -> profileService.getByUser(auth.getName()));
}
}
```
로그인한 유저의 정보를 `SecurityContext` 가져와 `Profile` 에서 검색 후 출력한다.
당연히 `SecurityContext::getAuthentication` 메서드는 `리액터 컨텍스트` 를 사용한다.
```
// ReactiveSecurityContextHolder.java
public static Mono getContext() {
return Mono.subscriberContext()
.filter( c -> c.hasKey(SECURITY_CONTEXT_KEY))
.flatMap( c-> c.>get(SECURITY_CONTEXT_KEY));
}
```
### SecurityWebFilterChain
인증과정을 거치려면 내부 컨텍스트에 엑세스 하려고 하면 `SecurityContext` 가 해당 `리액터 컨텍스트` 에 존재해야하고 `Authentication` 객체가 `SecurityContext` 안에 할당되어야 한다.
이 모든 과정을 `ReactorContextWebFilter` 에 `SecurityWebFilterChain` 적용하고 이를 통해 `리액터 컨텍스트` 안에 `SecurityContext` 객체와 `Authentication` 객체를 집어넣는다.
`SecurityContext` 를 집어 넣는 함수는 `ServerSecurityContextRepository` 를 사용한다.
```
package org.springframework.security.web.server.context;
public interface ServerSecurityContextRepository {
Mono save(ServerWebExchange exchange, SecurityContext context);
Mono load(ServerWebExchange exchange);
}
```
`SecurityContext` 를 특정 `ServerWebExchange`에 저장, 사용할 수 있다.
```
@Configuration
@EnableReactiveMethodSecurity // 별도의 MethodInterceptor 사용시 필요함
public class SecurityConfiguration {
@Bean // ReactorContextWebFilter 에 적용할 시큐리티 필터
public SecurityWebFilterChain securityFilterChainConfigurer(ServerHttpSecurity httpSecurity) {
// 기존 mvc 에서 사용하던 HttpSecurity 에서 webflux 용으로 정의된 ServerHttpSecurity
// 기존 spring security 사용 방식과 크게 다르지 않다.
return httpSecurity
.authorizeExchange()
.anyExchange().permitAll().and()
.httpBasic().and()
.formLogin().and()
.build();
}
private static final Pattern PASSWORD_ALGORITHM_PATTERN = Pattern.compile("^\\{.+}.*$");
private static final String NOOP_PASSWORD_PREFIX = "{noop}";
@Bean // 로그인에 사용할 테스트 사용자 생성
public MapReactiveUserDetailsService reactiveUserDetailsService(ObjectProvider passwordEncoder) {
return new MapReactiveUserDetailsService(
User.withUsername("user")
.password("user")
.passwordEncoder(p -> getOrDeducePassword(p, passwordEncoder.getIfAvailable()))
.roles("USER")
.build(),
User.withUsername("admin")
.password("admin")
.passwordEncoder(p -> getOrDeducePassword(p, passwordEncoder.getIfAvailable()))
.roles("USER", "ADMIN")
.build());
}
private String getOrDeducePassword(String password, PasswordEncoder encoder) {
if (encoder != null || PASSWORD_ALGORITHM_PATTERN.matcher(password).matches()) {
return password;
}
return NOOP_PASSWORD_PREFIX + password;
}
}
```
### WebFlux with JWT
#### Custom SecurityContextRepository, AuthenticationManager
스프링 시큐리티는 `SecurityContextRepository` 를 통해 `SecurityContext` 를 `리액터 컨텍스트`에 저장하고 삭제한다.
default 스프링 시큐리티의 경우 DB 로부터 UserDetails 를 가져와 등록해두고 사용하겠지만 우리는 JWT 를 사용하기에 `ServerSecurityContextRepository` 상속받아 커스터마이징 해야한다.
```
@Component
@RequiredArgsConstructor
public class SecurityContextRepository implements ServerSecurityContextRepository {
private final AuthenticationManager authenticationManager;
@Override
public Mono save(ServerWebExchange swe, SecurityContext sc) {
throw new UnsupportedOperationException("Not supported yet.");
}
@Override
public Mono load(ServerWebExchange swe) {
ServerHttpRequest request = swe.getRequest();
String authToken = request.getHeaders().getFirst(HttpHeaders.AUTHORIZATION);
if (authToken != null) {
Authentication auth = new UsernamePasswordAuthenticationToken(authToken, authToken);
return authenticationManager
.authenticate(auth)
.map(authentication -> new SecurityContextImpl(authentication));
} else {
return Mono.empty();
}
}
}
```
`SecurityContext` 를 생성 및 저장하기 request 헤더로 부터 JWT 토큰을 가져와 `AuthenticationManager`로 넘기는 코드가 `load` 에 정의되어 있다.
```
@Component
@RequiredArgsConstructor
public class AuthenticationManager implements ReactiveAuthenticationManager {
private final JwtTokenUtil jwtUtil;
@Override
public Mono authenticate(Authentication authentication) {
String authToken = authentication.getCredentials().toString();
if (!jwtUtil.validateToken(authToken)) {
return Mono.empty();
}
Claims claims = jwtUtil.getAllClaimsFromToken(authToken);
String role = claims.get("role", String.class);
List authorities = Collections.singletonList(new SimpleGrantedAuthority(role));
return Mono.just(new UsernamePasswordAuthenticationToken(claims.getSubject(), null, authorities));
}
}
```
`AuthenticationManager` 는 전달받은 토큰으로 `role` 을 꺼내어 `Authority` 를 지정하고 `Authentication` 인증 객체를 반환한다.
`SecurityContextPath` 가 리액터 컨텍스트로 변경되었을 뿐 기존 `WebMVC` 모델의 스프링 시큐리티 방식과 비슷하다.
```
@Bean // ReactorContextWebFilter 에 적용할 시큐리티 필터
public SecurityWebFilterChain securityFilterChainConfigurer(ServerHttpSecurity httpSecurity) {
// 기존 mvc 에서 사용하던 HttpSecurity 에서 webflux 용으로 정의된 ServerHttpSecurity
// 기존 spring security 사용 방식과 크게 다르지 않다.
return httpSecurity
.exceptionHandling()
//.authenticationEntryPoint((swe, e) -> Mono.fromRunnable(() -> // 미 로그인
// swe.getResponse().setStatusCode(HttpStatus.UNAUTHORIZED)))
//.accessDeniedHandler((swe, e) -> Mono.fromRunnable(() -> // 미 로그인
// swe.getResponse().setStatusCode(HttpStatus.FORBIDDEN)))
.authenticationEntryPoint((swd, e) -> Mono.error(new AuthenticationCredentialsNotFoundException("")))
.accessDeniedHandler((swe, e) -> Mono.error(new AccessDeniedException("")))
.and()
.authenticationManager(authenticationManager)
.securityContextRepository(securityContextRepository)
.csrf().disable()
.cors().disable()
.httpBasic().disable() // Basic Authentication disable
.formLogin().disable()
.authorizeExchange()
.pathMatchers("/member/join", "/member/login").permitAll()
.pathMatchers("/member/**", "/rent/**").authenticated()
.pathMatchers("/admin/**").hasAnyRole("ROLE_ADMIN")
.anyExchange().permitAll()
.and()
.build();
}
```
마지막으로 `ServerHttpSecurity` 에 커스터마이징한 `SecurityContextRepository, AuthenticationManager` 를 등록하고 별도의 에러 핸들링 처리를 한다면 에러를 반환해 핸들링,\
없다면 `HttpStatus.UNAUTHORIZED` 로 단순 `HTTP Status` 만 반환할 수 있다.
## 스프링 데이터 with WebFlux
기존에 `WebMVC` 방식의 데이터베이스 접근시 `JDBC` 를 구현한 `드라이버 라이브러리` 를 사용해 접근해왔다.
리액티브한 DB 통신도 HTTP 와 다르지 않다.\
이론적으로 DB 접근용 서비스를 생성하고 `WebClient` 를 사용해 DB 데이터를 가져온다면 비동기 DB 접근 라이브러리를 구현한 것 과 다름없다.
다행이도 다양한 DB 벤더사에서 자바 비동기 DB 연결 라이브러리인 `리액티브 드라이버`를 제공함으로 단순히 라이브러리만 추가하면 **데이터베이스 레이어**에 대한 논 블록킹 엑세스를 할 수 있다.
`spring-boot-starter-data-mongodb-reactive`\
`spring-boot-starter-data-cassandra-reactive`\
`spring-boot-starter-data-redis-reactive`\
`spring-boot-starter-data-r2dbc`
스프링 데이터 팀에서 기존에 사용한던 `Repository` 패턴을 리액티브 방식에도 똑같이 사용할 수 있도록 추상화를 통해 구현해두었다.
각 모듈들이 `ReactiveCurdRepository` 인터페이스를 사용해 `Reactor` 라이브러리와 통합되어 자연스럽게 리액티브하게 코드작성이 가능하다.
### 스프링 데이터 몽고DB 리액티브
`NoSQL` 의 경우 각 벤더사에서 통합된 규약이 없다.\
각 벤더사에서 자기들만의 드라이버 라이브러리를 제공하고 스프링 데이터 팀은 스프링에서 해당 라이브러리들을 쉽게 사용할 수 있도록 각종 모듈을 개발하고 있다
`NoSQL DB` 는 최근에 만들어 져서 대부분 벤더사가 `리액티브 드라이버` 를 제공하고 있으며\
스프링 데이터 팀은 몽고DB 에서 제공하는 `리액티브 드라이버` 를 쉽고 편하게 사용할 수 있도록 `spring-boot-starter-data-mongodb-reactive` 모듈을 작성해두었다.
해당 모듈을 사용하면 스프링 팀에서 만든 `Repository` 패턴을 사용해 메서드명 기반으로 쿼리문이 자동 생성/사용 할 수 있다.
#### ReactiveMongoRepository
```
import org.springframework.data.mongodb.repository.ReactiveMongoRepository;
// ReactiveCurdRepository 구현체
@Repository
public interface BookReactiveMongoRepository extends ReactiveMongoRepository {
Mono findOneByTitle(Mono title);
Flux findManyByTitleRegex(String regexp);
@Meta(maxScanDocuments = 3)
Flux findByAuthorsOrderByPublishingYearDesc(Publisher authors);
@Query("{ 'authors.1': { $exists: true } }")
Flux booksWithFewAuthors();
Flux findByPublishingYearBetweenOrderByPublishingYear(
Integer from,
Integer to,
Pageable pageable
);
}
```
#### ReactiveMongoTemplate
`ReactiveMongoRepository` 외에도 `ReactiveMongoTemplate` 를 사용해 쿼리 조작이 가능하다.
```
@Service
@RequiredArgsConstructor
public class RxMongoTemplateQueryService {
private static final String BOOK_COLLECTION = "book";
private final ReactiveMongoTemplate mongoTemplate; // ReactiveMongoTemplate implements ReactiveMongoOperations
public Flux findBooksByTitle(String title) {
Query query = Query.query(new Criteria("title")
.regex(".*" + title + ".*"))
.limit(100);
return mongoTemplate.find(query, Book.class, BOOK_COLLECTION);
}
}
```
#### MongoClient
몽고DB 에서 제공하는 리액티브 드라이버 구현체가 `com.mongodb.reactivestreams.client.MongoClient` 클래스이다.
>
`org.mongodb:mongodb-driver-reactivestreams` 모듈에서 제공하며 `spring-boot-starter-data-mongodb-reactive` 에서 내부적으로 사용한다.
`MongoClient` 클래스를 사용해도 쿼리조작이 가능하다.
```
@Service
@RequiredArgsConstructor
public class RxMongoDriverQueryService {
private final MongoClient mongoClient;
public Flux findBooksByTitle(String title, boolean negate) {
return Flux.defer(() -> {
Bson query = Filters.regex("title", ".*" + title + ".*");
if (negate) query = Filters.not(query);
return mongoClient
.getDatabase("test-db")
.getCollection("book")
.find(query);
}).map(doc -> new Book(
doc.getObjectId("id"), // Document
doc.getString("title"),
doc.getInteger("pubYear")));
}
}
```
#### 트랜잭션(ReactiveMongoTemplate.inTransaction)
`MongoDB 4.0` 버전 이전까지 하나의 문서 에 대해서만 트랜잭션을 제공하는 `Single-Document Transaction` 기능만 있었다.\
하나의 문서에 모든 정보를 삽입하여 사용하기에 하나의 트랜잭션은 하나의 문서만 건들여서 `Single-Document Transaction` 으로도 충분해야 하지만 항상 예외가 있는법,\
결국 여러 문서에 대한 트랜잭션 `Multi-Document Transaction` 기능을 `MongoDB 4.0` 부터 지원한다.
> WiredTiger 스토리지 엔진의 샤딩설정이 되어 있지 않고 복제설정일 경우에만 `Multi-Document Transaction` 을 지원한다.
`ReactiveMongoTemplate` 의 `inTransaction` 메서드를 사용하면
```
private Mono doTransferMoney(String from, String to, Integer amount) {
return mongoTemplate.inTransaction().execute(session -> session
.findOne(queryForOwner(from), Wallet.class)
.flatMap(fromWallet -> session
.findOne(queryForOwner(to), Wallet.class)
.flatMap(toWallet -> {
if (fromWallet.hasEnoughFunds(amount)) {
fromWallet.withdraw(amount);
toWallet.deposit(amount);
return session.save(fromWallet)
.then(session.save(toWallet))
.then(ReactiveMongoContext.getSession())
// An example how to resolve the current session
.doOnNext(tx -> log.info("Current session: {}", tx))
.then(Mono.just(TxResult.SUCCESS));
} else {
return Mono.just(TxResult.NOT_ENOUGH_FUNDS);
}
})))
.onErrorResume(e -> Mono.error(new RuntimeException("Conflict")))
.last();
}
```
### 스프링 데이터 R2DBC
> R2DBC: Reactive Relational Database Connectivity
>
아래와 같은 DBMS 에 대하여 r2dbc 라이브러리를 제공
```
H2 (io.r2dbc:r2dbc-h2)
MariaDB (org.mariadb:r2dbc-mariadb)
Microsoft SQL Server (io.r2dbc:r2dbc-mssql)
MySQL (dev.miku:r2dbc-mysql)
jasync-sql MySQL (com.github.jasync-sql:jasync-r2dbc-mysql)
Postgres (io.r2dbc:r2dbc-postgresql)
Oracle (com.oracle.database.r2dbc:oracle-r2dbc)
```
지금까지 `스프링 JDBC` 혹은 `스프링 데이터 JDBC` 혹은 `JPA` 를 사용해 생성된 `Hikari CP` 안의 연결객체가 `JDBC` 드라이버를 사용해 관계형 DB 를 사용해 왔다.
```
@Repository
public interface BookSpringDataJdbcRepository extends CrudRepository {
@Query("SELECT * FROM book b WHERE b.title = :title")
CompletableFuture findBookByTitleAsync(@Param("title") String title);
}
```
`JDBC, JPA` 등의 `관계형 DB 라이브러리`들은 모두 `동기/블럭킹` 방식으로 동작한다.
다행이도 `spring data jdbc` 를 개발한 `스프링 데이터 Relational` 프로젝트 팀에서\
**리액티브에 적합**한 `자바 DB 드라이버`인 `리액티브 드라이버`를 개발중이다.\
이 `리액티브 드라이버` 를 사용한 프로젝트가 `R2DBC` 프로젝트이다.
더이상 `JDBC` 를 사용하지 않고 리액티브 스택에 적합한 `리액티브 드라이버`를 사용해 DB 에 접근, 데이터를 조작한다.
> 안타깝게도 `JPA` 는 기존 코드가 너무 복잡했는지 리액티브 지원을 하지 않을것으로 보인다.
#### ReactiveCrudRepository
```
@Repository
public interface MemberRepository extends ReactiveCrudRepository {
Mono findByName(String name);
Mono findByUserName(String name);
@Query("SELECT * FROM member WHERE name = :name AND user_name = :userName")
Mono findByNameAndUserName(String name, String userName);
}
```
#### R2dbcEntityTemplate
```
@Service
@RequiredArgsConstructor
public class MemberDynamicRepository {
private final R2dbcEntityTemplate r2dbcEntityTemplate;
public Flux findTest(String userName) {
Query query = Query.query(where("user_name").like("%" + userName + "%"))
.limit(10)
.offset(0);
return r2dbcEntityTemplate.select(Member.class)
.matching(query)
.all();
}
}
```
### 스프링 데이터 Redis
`spring-boot-starter-data-redis-reactive` 모듈을 사용 `ReactiveRedisTemplate` 클래스가 `Redis` 커넥션의 핵심클래스이다.
> 다른 스프링 데이터 프로젝트와 달리 `Repository` 가 존재하지 않음\
> 일반적인 데이터 관리 외에도 구독/발행 구조의 메시지 기능도 지원한다.
> `spring-boot-starter-data-redis-reactive` 모듈은 내부적으로 `Lettuce` 라이브러리를 사용한다.
> , 현재 non blokcing 을 지원하는 redis 라이브러리가 Lettuce 가 유일하다.\
> 또한 `Lettuce` 라이브러리 내에서 `Reactor` 라이브러리를 사용한다.
```
public class Sample {
private String name;
private String description;
}
@Configuration
public class RedisConfig {
@Value("${redis.host}")
private String host;
@Value("${redis.port}")
private Integer port;
@Bean
public ReactiveRedisConnectionFactory reactiveRedisConnectionFactory() {
return new LettuceConnectionFactory(host, port);
}
@Bean
public ReactiveRedisTemplate reactiveRedisTemplate(ReactiveRedisConnectionFactory reactiveRedisConnectionFactory) {
StringRedisSerializer keySerializer = new StringRedisSerializer();
Jackson2JsonRedisSerializer valueSerializer = new Jackson2JsonRedisSerializer<>(Sample.class);
RedisSerializationContext.RedisSerializationContextBuilder builder =
RedisSerializationContext.newSerializationContext(keySerializer);
RedisSerializationContext context = builder.value(valueSerializer).build();
return new ReactiveRedisTemplate(reactiveRedisConnectionFactory, context);
}
}
```
```
@Service
@RequiredArgsConstructor
public class SampleService {
private final ReactiveRedisTemplate redisTemplate;
public Mono put(String key, Sample sample) {
return redisTemplate.opsForValue().set(key, sample);
}
public Mono get(String key) {
return redisTemplate.opsForValue().get(key);
}
public Flux getAll(String keyPattern){
return redisTemplate.keys(keyPattern)
.flatMap(key-> redisTemplate.opsForValue().get(key));
}
public Mono delete(String key) {
return redisTemplate.opsForValue().delete(key);
}
}
```
## 기타
### ListenableFuture
스프링에서 제공하는 `Future` 구현 클래스
```
public interface ListenableFuture extends Future {
void addCallback(ListenableFutureCallback callback);
void addCallback(SuccessCallback successCallback, FailureCallback failureCallback);
default CompletableFuture completable() {
CompletableFuture completable = new DelegatingCompletableFuture<>(this);
addCallback(completable::complete, completable::completeExceptionally);
return completable;
}
}
```
위의 `SseEmitter` 와 마찬가지로 스프링 리액티브의 반환데이터로 사용된다.
```
// 비동기 RestTemplate
AsyncRestTemplate httpClient = new AsyncRestTemplate();
@GetMapping
public ListenableFuture requestData() {
AsyncDatabaseClient databaseClient = new FakeAsyncDatabaseClient();
// /hello 의 호출결과를 CompletableFuture 으로 반환하는 어뎁터
CompletionStage completionStage = AsyncAdapters.toCompletion(httpClient.execute(
"http://localhost:8080/hello",
HttpMethod.GET, null,
new HttpMessageConverterExtractor<>(String.class, messageConverters) // http 의 body 부분 컨버터들 지정
));
// CompletionStage(CompletableFuture 의 인터페이스) 를 ListenableFuture 로 변환
return AsyncAdapters.toListenable(databaseClient.store(completionStage));
}
```
`AsyncRestTemplate.execute` 가 반환하는 `ListenableFuture` 를 `CompletionStage` 로 변환 반환된 `CompletionStage` 를 데이터베이스에 저장후 다시 반환된 `CompletionStage` 를 `ListenableFuture` 로 변환한다.
일반적으로 간단한 비동기 처리는 `Future` 를 구현한 `CompletableFuture`(`CompletionStage` 구현체) 를 주로 사용한다.
메서드 정의시 반환값을 스프링에서 기본 제공하는 `ListenableFuture` 를 구현하는 객체로 반환하기 힘들기에\
위와같은 `AsyncAdapters` 를 사용해 비동기 결과를 `ListenableFuture` 로 변환해주는 어뎁터를 사용하는 것이 편하다.
```
public final class AsyncAdapters {
// ListenableFuture -> completionStage
public static CompletionStage toCompletion(ListenableFuture future) {
CompletableFuture completableFuture = new CompletableFuture<>();
future.addCallback(completableFuture::complete, completableFuture::completeExceptionally);
return completableFuture;
}
// completionStage -> ListenableFuture
public static ListenableFuture toListenable(CompletionStage stage) {
SettableListenableFuture future = new SettableListenableFuture<>();
stage.whenComplete((v, t) -> {
if (t == null) future.set(v);
else future.setException(t);
});
return future;
}
}
```
### AsyncRestTemplate
스프링 리액티브에선 동기방식인 일반 `RestTemplate` 을 사용하지 않고 `AsyncRestTemplate` 를 사용한다.\
흔히사용하는 `execute` 메서드의 반환값이 `ListenableFuture` 객체이다.
```
@Override
public ListenableFuture execute(String url, HttpMethod method, AsyncRequestCallback requestCallback,
ResponseExtractor responseExtractor, Object... uriVariables) throws RestClientException {
URI expanded = getUriTemplateHandler().expand(url, uriVariables);
return doExecute(expanded, method, requestCallback, responseExtractor);
}
```
출처 :