[Spring Webflux] #4 스프링 웹플럭스의 코틀린 코루틴 지원 내부 구현 뜯어보기

스프링 진영에서 코틀린은 참 매력적인 언어다. 이미 많은 유명한 서비스들이 코틀린 스프링을 사용하고 있고, 개인적 경험에서는 자바의 장황한 문법에 비해 간결해진 코드와 Null Safety가 정말 큰 장점으로 느껴진다.

 

특히 웹플럭스에서, 선언형(declarative)으로 메소드 체이닝을 통해 코드를 작성하는 기존의 Reactor와 달리 코틀린 코루틴은 명령형(imperative)으로 Non-blocking 코드를 간단 명료하게 작성할 수 있다.

 

Reactor가 Kotlin Coroutines과 어떻게 호환 되는지, Spring은 코루틴을 어떻게 지원하는지, 그리고 Spring Webflux 어플리케이션에서 코루틴을 어떻게 사용하는지 정리해보았다.

 

본문을 요약하면,

• Reactor(Mono/Flux) <-> Coroutines(suspend/Flow)의 상호 변환은 코루틴 진영의 kotlinx-coroutines-reactor 라이브러리를 통해 이루어진다.
• 스프링 코어를 포함한 스프링 진영의 WebFlux, R2DBC 등의 Spring Reactive는 kotlinx-coroutines-reactor 라이브러리와 리플렉션을 통해 코틀린 코루틴을 라이브러리와 언어 차원에서 지원한다.

이 두가지 내용을 염두에 두고 읽는 것을 추천한다.

Reactor <-> Coroutines간의 변환

Spring WebFlux에서 코루틴을 어떻게 사용하는지 이해하려면 Spring Reactive에서 기본으로 사용되는 Reactor와 코루틴이 어떻게 상호 변환 되는지 이해 해야한다.

우선, 둘의 호환은 코틀린 진영에서 제공하는 kotlinx-coroutines-reactor와 kotlinx-coroutines-reactive 라이브러리를 통해 이루어진다.

// build.gradle.kts
dependencies {
    implementation("org.jetbrains.kotlinx:kotlinx-coroutines-core:${1.4.0 이상 버전}")
    implementation("org.jetbrains.kotlinx:kotlinx-coroutines-reactive:${1.4.0 이상 버전}")
    implementation("org.jetbrains.kotlinx:kotlinx-coroutines-reactor:${1.4.0 이상 버전}")
}

 

Reactor의 코루틴으로 변환은 아래와 같이 대응된다.

// 리턴 값
fun handler(): Mono<Void> -> suspend fun handler()
fun handler(): Mono<T> -> suspend fun handler(): T 또는 suspend fun handler(): T?
fun handler(): Flux<T> -> fun handler(): Flow<T>

// 함수 파라미터
// 지연(laziness)이 필요 없을 때
fun handler(mono: Mono<T>) -> fun handler(value: T)
// 지연(laziness)이 필요 할 때
fun handler(mono: Mono<T>) -> fun handler(supplier: suspend () → T) 또는 fun handler(supplier: suspend () → T?)

 

코루틴의 Flow는 Reactor의 Flux와 동등하지만, 몇 가지 차이점이 있다.

• Flow는 push-based, Flux는 push-pull hybrid
• Backpressure는 중지(suspending)함수를 통해 구현 된다.
• Flow는 하나의 함수(suspend fun collect)만 가지며, 연산자들은 extension으로 구현 할 수 있다.

Reactor -> Coroutines 변환

Reactor의 Mono/Flux -> suspend/Flow 변환은 Extension을 사용해 Mono/Flux에 연산자를 추가하는 식으로 구현되었다.

Mono에 awaitSingle()/ awaitSingleOrNull() 연산자를 사용하여 suspend 함수로 사용하고,

Flux에 asFlow() 연산자를 사용하여 Flow로 변환할 수 있다.

// Mono.kt
package kotlinx.coroutines.reactor

public suspend fun <T> Mono<T>.awaitSingleOrNull(): T? = suspendCancellableCoroutine { cont ->
    injectCoroutineContext(cont.context).subscribe(object : Subscriber<T> {
        private var value: T? = null

        override fun onSubscribe(s: Subscription) {
            cont.invokeOnCancellation { s.cancel() }
            s.request(Long.MAX_VALUE)
        }

        override fun onComplete() {
            cont.resume(value)
            value = null
        }

        override fun onNext(t: T) {
            value = t
        }

        override fun onError(error: Throwable) { cont.resumeWithException(error) }
    })
}

// ReactiveFlow.kt
package kotlinx.coroutines.reactive

public fun <T : Any> Publisher<T>.asFlow(): Flow<T> =
    PublisherAsFlow(this)

 

실제로 Spring Webflux에서 제공하는 Http Client인 WebClient의 경우,

기존의 WebClient.RequestHeadersSpec.exchangeToMono를 사용하여 Mono로 결과를 받아오는 대신, 위의 awaitSingle() 연산자를 사용해 내부적으로 Mono를 Suspend 함수로 변환해 Suspend 함수로 사용할 수 있도록 Extension을 제공한다.

// WebClientExtension.kt
package org.springframework.web.reactive.function.client

suspend fun <T: Any> RequestHeadersSpec<out RequestHeadersSpec<*>>.awaitExchange(
	responseHandler: suspend (ClientResponse) -> T): T =
    exchangeToMono { 
        mono(Dispatchers.Unconfined) { responseHandler.invoke(it) } 
    }.awaitSingle()

Coroutines -> Reactor 변환

suspend 함수 -> Mono의 경우 코루틴 컨텍스트와 suspend CoroutineScope를 받아 Mono로 리턴하는 mono 메서드를 사용한다. 

Flow -> Flux는 Flow에 asFlux 연산자 Extension을 사용하면 된다.

package kotlinx.coroutines.reactor

// Mono.kt
public fun <T> mono(
    context: CoroutineContext = EmptyCoroutineContext,
    block: suspend CoroutineScope.() -> T?
): Mono<T> {
    require(context[Job] === null) { "Mono context cannot contain job in it." +
            "Its lifecycle should be managed via Disposable handle. Had $context" }
    return monoInternal(GlobalScope, context, block)
}

private fun <T> monoInternal(
    scope: CoroutineScope,
    context: CoroutineContext,
    block: suspend CoroutineScope.() -> T?
): Mono<T> = Mono.create { sink ->
    val reactorContext = context.extendReactorContext(sink.currentContext())
    val newContext = scope.newCoroutineContext(context + reactorContext)
    val coroutine = MonoCoroutine(newContext, sink)
    sink.onDispose(coroutine)
    coroutine.start(CoroutineStart.DEFAULT, coroutine, block)
}

// ReactorFlow.kt
public fun <T: Any> Flow<T>.asFlux(
	context: CoroutineContext = EmptyCoroutineContext
): Flux<T> =
    FlowAsFlux(this, Dispatchers.Unconfined + context)

private class FlowAsFlux<T : Any>(
    private val flow: Flow<T>,
    private val context: CoroutineContext
) : Flux<T>() {
    override fun subscribe(subscriber: CoreSubscriber<in T>) {
        val hasContext = !subscriber.currentContext().isEmpty
        val source = if (hasContext) flow.flowOn(subscriber.currentContext().asCoroutineContext()) else flow
        subscriber.onSubscribe(FlowSubscription(source, subscriber, context))
    }
}

 

Spring Framework의 Kotlin Coroutines 지원

Spring Reactive는 상호 운용성을 염두에 두고 Reactive Streams를 기반으로 구축되었다.

그래서 디폴트로 사용하는 Reactor 대신 CompletableFuture, RxJava3, Coroutines와 같은 비동기/반응형 API와도 쉽게 호환되도록 설계 되었다.

 

스프링 코어의 ReactiveAdapterRegistry 클래스는 Reactor의 Mono/Flux같은 Reactive Streams Publisher와 다른 비동기/반응형 타입간 상호 변환을 제공하는 어댑터 함수를 등록한다. 정적 초기화 블럭(static { ... })을 보면 클래스 로더를 사용하여 여러 비동기/반응형 API 사용 여부를 결정한다.

코루틴의 경우 kotlinx.coroutines.reactor.MonoKt가 존재하면,  CoroutinesRegistrar 클래스의 registerAdapter 함수를 호출하여 coroutines.Deffered <-> Mono와 coroutines.Flow <-> Flux의 변환 어댑터 2개를 등록한다.

package org.springframework.core;

public class ReactiveAdapterRegistry {
    public ReactiveAdapterRegistry() {
        if (reactiveStreamsPresent) {
            if (reactorPresent) {
            	(new ReactorRegistrar()).registerAdapters(this);
            }
            if (rxjava3Present) {
            	(new RxJava3Registrar()).registerAdapters(this);
            }
            if (reactorPresent && kotlinCoroutinesPresent) {
            	(new CoroutinesRegistrar()).registerAdapters(this);
            }
            if (mutinyPresent) {
            	(new MutinyRegistrar()).registerAdapters(this);
            }
            if (!reactorPresent) {
            	(new FlowAdaptersRegistrar()).registerAdapters(this);
            }
            }
    }

    static {
        ClassLoader classLoader = ReactiveAdapterRegistry.class.getClassLoader();
        reactiveStreamsPresent = ClassUtils.isPresent("org.reactivestreams.Publisher", classLoader);
        reactorPresent = ClassUtils.isPresent("reactor.core.publisher.Flux", classLoader);
        rxjava3Present = ClassUtils.isPresent("io.reactivex.rxjava3.core.Flowable", classLoader);
        kotlinCoroutinesPresent = ClassUtils.isPresent("kotlinx.coroutines.reactor.MonoKt", classLoader);
        mutinyPresent = ClassUtils.isPresent("io.smallrye.mutiny.Multi", classLoader);
    }
    
    private static class CoroutinesRegistrar {

        @SuppressWarnings("KotlinInternalInJava")
        void registerAdapters(ReactiveAdapterRegistry registry) {
            registry.registerReactiveType(
                ReactiveTypeDescriptor.singleOptionalValue(kotlinx.coroutines.Deferred.class,
                	() -> kotlinx.coroutines.CompletableDeferredKt.CompletableDeferred(null)),
                source -> CoroutinesUtils.deferredToMono((kotlinx.coroutines.Deferred<?>) source),
                source -> CoroutinesUtils.monoToDeferred(Mono.from(source)));

            registry.registerReactiveType(
                ReactiveTypeDescriptor.multiValue(kotlinx.coroutines.flow.Flow.class, kotlinx.coroutines.flow.FlowKt::emptyFlow),
                source -> kotlinx.coroutines.reactor.ReactorFlowKt.asFlux((kotlinx.coroutines.flow.Flow<?>) source),
                kotlinx.coroutines.reactive.ReactiveFlowKt::asFlow);
        }
    }
    
    ...
    
}

 

더 깊이 들어가면, Deffered <-> Mono의 변환을 담당하는 CoroutinesUtils.monoToDeffered와 CoroutinesUtils.defferedToMono와 메소드는 각각 위에서 살펴 보았던 kotlinx.coroutines.reactor.MonoKt의 mono와 awaitSingleOrNull 메서드를 사용하는 것을 확인할 수 있다.

Flow <-> Flux에도 마찬가지로 kotlinx.coroutines.reactor/reactive의 ReactorFlowKt.asFlux와 ReactiveFlowkt.asFlow를 사용하고 있는 것을 볼 수 있다.

 

추가적으로 CoroutinesUtils 클래스에서는 MonoKt의 mono 메서드를 사용하여 suspend 함수를 Mono/Flux로 변환하는 invokeSuspendingFunction도 있다.

package org.springframework.core;

public abstract class CoroutinesUtils {

    public static <T> Mono<T> deferredToMono(Deferred<T> source) {
        return MonoKt.mono(Dispatchers.getUnconfined(),
            (scope, continuation) -> source.await(continuation));
    }

    public static <T> Deferred<T> monoToDeferred(Mono<T> source) {
        return BuildersKt.async(GlobalScope.INSTANCE, Dispatchers.getUnconfined(),
            CoroutineStart.DEFAULT,
            (scope, continuation) -> MonoKt.awaitSingleOrNull(source, continuation));
    }

    public static Publisher<?> invokeSuspendingFunction(
    	CoroutineContext context, Method method, @Nullable Object target, @Nullable Object... args) {

        Assert.isTrue(KotlinDetector.isSuspendingFunction(method), "Method must be a suspending function");
        KFunction<?> function = ReflectJvmMapping.getKotlinFunction(method);
        ...

        Mono<Object> mono = MonoKt.mono(context, (scope, continuation) -> {
            Map<KParameter, Object> argMap = CollectionUtils.newHashMap(args.length + 1);

            // 함수의 파라미터를 argMap에 담는 로직
            ...

            return KCallables.callSuspendBy(function, argMap, continuation);
    	})
        .filter(result -> result != Unit.INSTANCE)
        .onErrorMap(InvocationTargetException.class, InvocationTargetException::getTargetException);

        KType returnType = function.getReturnType();

        // 리플렉션을 사용하여 확인한 함수의 리턴타입에 따라 리턴한다
        // Flow -> Flux, Mono -> Mono, Publisher && !Mono -> Flux
        // 이외 모든 타입 -> Mono
        if (KTypes.isSubtypeOf(returnType, flowType)) {
        	return mono.flatMapMany(CoroutinesUtils::asFlux);
        }
    	if (KTypes.isSubtypeOf(returnType, publisherType)) {
    		if (KTypes.isSubtypeOf(returnType, monoType)) {
    			return mono.flatMap(o -> ((Mono<?>)o));
    		}
    		return mono.flatMapMany(o -> ((Publisher<?>)o));
    	}
    	return mono;
    }
}

Spring WebFlux에서의 Coroutines

Spring WebFlux 어플리케이션에서 엔드 포인트를 정의하기 위해서는 Controller 또는 Router Function 방식을 사용한다.

Spring MVC 개발자들이 익숙한 Controller 방식으로 Reactor를 사용한 예시다.

package day.mercury.coroutines

import org.springframework.web.bind.annotation.GetMapping
import org.springframework.web.bind.annotation.RestController
import reactor.core.publisher.Flux
import reactor.core.publisher.Mono
import java.time.Duration

@RestController
class ExampleController {

    @GetMapping("hi")
    fun hi(): Mono<String> {
        return Mono.just("hi").delayElement(Duration.ofSeconds(1))
    }

    @GetMapping("countdown")
    fun countdown(): Flux<Int> {
        return Flux.fromIterable((1..5).reversed())
            .delayElements(Duration.ofSeconds(1))
    }
}

 

Reactor를 Coroutines로 옮겨 적으려면 아래와 같이 Mono를 suspend 함수로, Flux를 Flow로 치환하여 코드를 작성할 수 있다.

package day.mercury.coroutines

import kotlinx.coroutines.delay
import kotlinx.coroutines.flow.Flow
import kotlinx.coroutines.flow.flow
import org.springframework.web.bind.annotation.GetMapping
import org.springframework.web.bind.annotation.RestController

@RestController
class ExampleController {

    @GetMapping("/hi")
    suspend fun hi(): String {
        delay(1000)
        return "hi"
    }

    @GetMapping("/countdown")
    fun countdown(): Flow<Int> {
        return flow {
            (1..5).reversed().forEach {
                delay(1000)
                emit(it)
            }
        }
    }
}

 

이러한 suspend/Flow와 Kotlin을 언어와 라이브러리 측면에서 Spring WebFlux는 어떻게 지원하는지 열심히 브레이크 포인트를 찍어서 요청으로부터 발생하는 내부 동작을 열심히 디버깅 해보았다.

 

Spring MVC에서는 Dispatcher Servlet이 요청을 적절한 핸들러에 위임한다면, Spring WebFlux에는 DispatcherHandler가 있다.

handle 메서드 맨 마지막에 flatMap으로 handlerRequestWith 메서드를 사용하는 것을 볼 수 있다.

package org.springframework.web.reactive;

public class DispatcherHandler implements WebHandler, PreFlightRequestHandler, ApplicationContextAware {
	
    @Nullable
    private List<HandlerAdapter> handlerAdapters;

    @Override
    public Mono<Void> handle(ServerWebExchange exchange) {
    	...
        return Flux.fromIterable(this.handlerMappings)
            .concatMap(mapping -> mapping.getHandler(exchange))
            .next()
            .switchIfEmpty(createNotFoundError())
            .onErrorResume(ex -> handleResultMono(exchange, Mono.error(ex)))
            .flatMap(handler -> handleRequestWith(exchange, handler));
    }

    private Mono<Void> handleRequestWith(ServerWebExchange exchange, Object handler) {
        ...
        if (this.handlerAdapters != null) {
            for (HandlerAdapter adapter : this.handlerAdapters) {
                if (adapter.supports(handler)) {
                	Mono<HandlerResult> resultMono = adapter.handle(exchange, handler);
                	return handleResultMono(exchange, resultMono);
                }
            }
        }
        return Mono.error(new IllegalStateException("No HandlerAdapter: " + handler));
    }

    ...

}

그리고 handlerRequestWith에서는 매핑된 핸들러를 실질적으로 처리하는 HandlerAdapter 인터페이스의 handle을 호출하게 되는데, @RestController와 같은 RequestMapping으로 핸들러를 등록하면 HandlerAdapter의 구현체인 RequestMappingHandlerAdapter 클래스의 handle을 호출하게 된다.

여기서 handler 메서드는 위에서 보았던 Spring Core에서 Mono/Flux <-> 코루틴 상호 변환 어댑터를 포함한 reactiveAdapterRegistry로 bindingContext를 생성하고, InvocableHandlerMethod 인스턴스의 invoke를 호출한다.

package org.springframework.web.reactive.result.method.annotation;

public class RequestMappingHandlerAdapter
	implements HandlerAdapter, DispatchExceptionHandler, ApplicationContextAware, InitializingBean {

	@Override
	public Mono<HandlerResult> handle(ServerWebExchange exchange, Object handler) {

            ...

            InitBinderBindingContext bindingContext = new InitBinderBindingContext(
                this.webBindingInitializer, this.methodResolver.getInitBinderMethods(handlerMethod),
                this.methodResolver.hasMethodValidator() && handlerMethod.shouldValidateArguments(),
                this.reactiveAdapterRegistry);

            InvocableHandlerMethod invocableMethod = this.methodResolver.getRequestMappingMethod(handlerMethod);

            DispatchExceptionHandler exceptionHandler =
                (exchange2, ex) -> handleException(exchange, ex, handlerMethod, bindingContext);

            Mono<HandlerResult> resultMono = this.modelInitializer
                .initModel(handlerMethod, bindingContext, exchange)
                .then(Mono.defer(() -> invocableMethod.invoke(exchange, bindingContext)))
                .doOnNext(result -> result.setExceptionHandler(exceptionHandler))
                .onErrorResume(ex -> exceptionHandler.handleError(exchange, ex));

            Scheduler optionalScheduler = this.methodResolver.getSchedulerFor(handlerMethod);
            if (optionalScheduler != null) {
                return resultMono.subscribeOn(optionalScheduler);
            }

            return resultMono;
    }

    ...

}

 

InvocableHandlerMethod의 invoke 메서드를 보면,

KotlinDetector.isSuspendingFunction(method)로 리플렉션을 통해 대상 핸들러 함수가 suspend 함수인지 판단한다.

그리고 대상 핸들러가 코틀린으로 작성된 함수라면 KotlinDelegate.invokeFunction을 호출한다.

 

KotlinDelegate의 invokeFunction은 함수 여부에 따라 앞서 보았던 스프링 코어의 CoroutinesUtils.invokeSuspendingFunction을 호출하거나, funciton.callBy를 호출한다.

 

그리고 invoke 메서드의 남은 로직에 따라 BindingContext에 있는 ReactiveAdapter를 이용해 suspend와 Flow와 같은 Coroutines를 Mono/Flux로 적절히 변환한다. (Mono.from(adapter.toPublisher(value))

package org.springframework.web.reactive.result.method;

public class InvocableHandlerMethod extends HandlerMethod {

    public Mono<HandlerResult> invoke(
        ServerWebExchange exchange, BindingContext bindingContext, Object... providedArgs) {

        return getMethodArgumentValuesOnScheduler(exchange, bindingContext, providedArgs).flatMap(args -> {
            ...
            Method method = getBridgedMethod();
            boolean isSuspendingFunction = KotlinDetector.isSuspendingFunction(method);
            try {
                if (KotlinDetector.isKotlinReflectPresent() && KotlinDetector.isKotlinType(method.getDeclaringClass())) {
                    value = KotlinDelegate.invokeFunction(method, getBean(), args, isSuspendingFunction, exchange);
                }
                else {
                    value = method.invoke(getBean(), args);
                }
            }
            catch ...

            HttpStatusCode status = getResponseStatus();
            if (status != null) {
                exchange.getResponse().setStatusCode(status);
            }

            MethodParameter returnType = getReturnType();
            if (isResponseHandled(args, exchange)) {
                Class<?> parameterType = returnType.getParameterType();
                ReactiveAdapter adapter = this.reactiveAdapterRegistry.getAdapter(parameterType);
                boolean asyncVoid = isAsyncVoidReturnType(returnType, adapter);
                if (value == null || asyncVoid) {
                    return (asyncVoid ? Mono.from(adapter.toPublisher(value)) : Mono.empty());
                }
                if (isSuspendingFunction && parameterType == void.class) {
                    return (Mono<HandlerResult>) value;
                }
            }

            HandlerResult result = new HandlerResult(this, value, returnType, bindingContext);
            return Mono.just(result);
        });
    }

    private static class KotlinDelegate {

        public static Object invokeFunction(Method method, Object target, Object[] args, boolean isSuspendingFunction,
            ServerWebExchange exchange) throws InvocationTargetException, IllegalAccessException {

            if (isSuspendingFunction) {
                Object coroutineContext = exchange.getAttribute(COROUTINE_CONTEXT_ATTRIBUTE);
                if (coroutineContext == null) {
                    return CoroutinesUtils.invokeSuspendingFunction(method, target, args);
                }
                else {
                    return CoroutinesUtils.invokeSuspendingFunction((CoroutineContext) coroutineContext, method, target, args);
                }
            }
            else {
                KFunction<?> function = ReflectJvmMapping.getKotlinFunction(method);
                
                if (function == null) {
                	return method.invoke(target, args);
                }
                if (method.isAccessible() && !KCallablesJvm.isAccessible(function)) {
                	KCallablesJvm.setAccessible(function, true);
                }
                Map<KParameter, Object> argMap = CollectionUtils.newHashMap(args.length + 1);

                // 함수의 파라미터를 argMap에 담는 로직
                ...
                Object result = function.callBy(argMap);
                return (result == Unit.INSTANCE ? null : result);
            }
        }
    }
}

Spring WebFlux에서는 코틀린 코루틴을 DispatcherHandler가 요청을 Reactor로 시작해서 핸들러에서 코루틴으로 변환되었다가, 최종적으로 응답에서 다시 Reactor로 변환하는 이러한 과정을 통해 지원한다.

마치며

개인적으로는 담당하고 있는 스프링 프로젝트에 이미 코틀린을 사용하고 있다.

다만 코틀린을 고려하면서 당시 코루틴까지 도입하려 POC를 진행하던 중 Redis 클라이언트 라이브러리인 Lettuce가 Reactive Streams API는 지원하는 반면, Coroutines API는 실험적 지원인 것을 보고 헉,,하고 포기했던 기억이 있다.

지금으로선 Coroutines API를 지원하지 않더라도 kotlinx-coroutines-reactor 라이브러리만 잘 사용해도 Reactive Streams API를 변환해서 사용할 수 있을 것 같다.

 

스프링 진영의 프로젝트의 코드를 뜯어볼 때 마다 참으로 경악스럽다. OOP의 모든 디자인 패턴 집합체를 볼 수 있다.

특히 서비스의 추상화는 그렇다 해도 언어 수준에서까지 추상화의 극을 달리는게 정말,,, 

 

자바의 경량스레드(Virtual Thread)는 언제쯤 웹플럭스나 코루틴을 대체 할 수 있을까. 물론 정식으로 도입된지 1년이 안 된 시점이기도 하다. 갈 길이 너무 멀어보인다. 당장 HikariCP 가상스레드 지원 PR이 1년 넘게 Open인걸,,

참고

Going Reative with Spring, Coroutines and Kotlin Flow

Non-Blocking Spring Boot with Kotlin Coroutines

Spring WebFlux에서는 어떻게 Kotlin Coroutine을 지원하고 있을까? (feat.Context)