<Spring> SUSU의 Coroutine

📌 SUSU 프로젝트

YAPP 23기에서 진행한 경조사비 관리 서비스이다.

자세한 내용은 아래 링크를 이용하면 된다.

결혼식, 돌잔치, 장례식, 생일 같은 경조사를 챙긴 적 있나요?
주고받은 소중한 마음들, 수수와 함께 경조사비를 똑똑하게 관리해요!

Disquiet*

많관부 ㅎㅎ

 

📌 왜 Coroutine을 썼을까

🔵 Coroutine 이란

자세한 설명은 이 글에서 안 할 것이다.

간단히 소개만 해보겠다.

 

위키백과에는 아래와 같이 쓰여있다.

Coroutines are computer program components that allow execution to be suspended and resumed, generalizing subroutines for cooperative multitasking.

중지, 재개를 지원하는 컴퓨터 프로그램 구성요소

협력적 멀티테스킹을 위한 일반화된 서브 루틴

 

이를 만든 목표는 이러하다

- No dependency on a particular implementation of Futures or other such rich library;
- Cover equally the "async/await" use case and "generator blocks";
- Make it possible to utilize Kotlin coroutines as wrappers for different existing asynchronous APIs (such as Java NIO, different implementations of Futures, etc).

비동기 처리를 Coroutine을 통해 처리할 수 있도록 하는 게 이것의 목표이다.

 

사람들은 흔히 Coroutine을 경량 쓰레드라고 한다.

이는 아래와 같은 특징 때문이다.

1. context-switching 비용이 적다
2. suspend를 이용한 루틴의 중지와 재개를 지원한다.
3. 애플리케이션 수준에서 스케줄링을 관리한다.

정리해보자면 코루틴은 이러하다.

• 실행의 지연과 재개를 허용함으로써, 비선점적 멀티태스킹을 위한 서브 루틴을 일반화한 컴퓨터 프로그램 구성요소
• 비동기 + non-blocking 지원해 주는 경량 스레드

 

🔵 수수의 기술 스택

수수는 Kotlin, Spring을 이용했다.

Spring은 Reactive Stack으로 구성했다.

하지만 R2DBC는 사용하지 않았다.

개발 생산성을 높이기 위해 JPA를 사용했다.

이를 위해 비동기 처리를 해야 했고, 이를 Coroutine을 이용해 처리했다.

 

🔵 Webflux와 Blocking I/O

Webflux는 Netty를 기반으로 작동한다.

Netty는 내부적으로 소수의 EventLoop를 기반으로 작동한다.

EventLoop를 이용하여 이벤트 큐에 쌓인 작업을 순차적으로 처리하는 방식이다.

 

이 시스템에 Blocking 로직이 주어지면 어떻게 될까?

Reactor MeltDown이 발생한다.

이때 앞선 작업이 끝나지 않고 EventLoop를 점유해 버린다면, 큐에 쌓인 작업은 대기하는 상황이 된다.

이는 곧 프로그램의 전체적인 성능 저하로 이어진다.

 

그러므로 Blocking 로직이 EventLoop에서 동작하는 상황을 만들면 안 된다.

실제로 공식 문서에도 이렇게 적혀있다.

Blocking API는 동시성 모델에 좋지 않다.

처리할 거면 다른 스레드에서 처리해라.

이게 공식 문서가 권하는 Webflux에서 Blocking 로직을 처리하는 법이다.

SUSU 서비스에서도 Blocking 로직이 존재한다.

JPA가 이에 해당한다.

JPA가 JDBC 기반으로 작동해서 Blocking 하게 작동한다.

그래서 이 부분을 비동기처리 해야 했고, 이를 코루틴을 이용해 처리했다.

 

🔵 Webflux, Reactor와 Coroutine

Webflux는 Project Reactor를 기반으로 구현되었다.

그래서 내부 구조를 살펴보면 Mono, Flux를 이용하는 것을 볼 수 있다.

Coroutine을 이용하지 않고 구현한다면, Project Reactor를 이용해 구현하면 된다.

 

그렇다면 Webflux + Reactor와 Coroutine은 어떻게 공존할 수 있는가?

이는 Webflux의 RequestMappingHandlerAdapter  handle(...)을확인해 보면 알 수 있다.

@Override
public Mono<HandlerResult> handle(ServerWebExchange exchange, Object handler) {
    ... // 생략
    InvocableHandlerMethod invocableMethod = this.methodResolver.getRequestMappingMethod(handlerMethod);
    ... // 생략
    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));
    ... // 생략
}

요청에 알맞은 Method를 찾아서 실행한다.

해당 invoke를 따라 들어가다 보면, InvocableHandlerMethod의 invokeFunction(...)으로 이어진다.

@Nullable
public static Object invokeFunction(Method method, Object target, Object[] args, boolean isSuspendingFunction, ServerWebExchange exchange) throws InvocationTargetException, IllegalAccessException {
    if (isSuspendingFunction) {
        Object coroutineContext = exchange.getAttribute("org.springframework.web.server.CoWebFilter.context");
        return coroutineContext == null ? CoroutinesUtils.invokeSuspendingFunction(method, target, args) : CoroutinesUtils.invokeSuspendingFunction((CoroutineContext)coroutineContext, method, target, args);
    } else {
        KFunction<?> function = ReflectJvmMapping.getKotlinFunction(method);
        if (function == null) {
            return method.invoke(target, args);
        } else {
            ... // 생략
        }
    }
}

 

이 함수에서 suspend 함수냐 아니면 일반 함수냐에 따라서 분기가 발생한다.

만약 suspend function임을 가정하고 진행한다면, CoroutinesUtils의 invokeSuspendingFunction(...)를 마주하게 된다.

/**
 * Invoke a suspending function and converts it to {@link Mono} or
 * {@link Flux}.
 * @param context the coroutine context to use
 * @param method the suspending function to invoke
 * @param target the target to invoke {@code method} on
 * @param args the function arguments. If the {@code Continuation} argument is specified as the last argument
 * (typically {@code null}), it is ignored.
 * @return the method invocation result as reactive stream
 * @throws IllegalArgumentException if {@code method} is not a suspending function
 * @since 6.0
 */
@SuppressWarnings("deprecation")
public static Publisher<?> invokeSuspendingFunction(CoroutineContext context, Method method, Object target,
        Object... args) {
    Assert.isTrue(KotlinDetector.isSuspendingFunction(method), "'method' must be a suspending function");
    KFunction<?> function = Objects.requireNonNull(ReflectJvmMapping.getKotlinFunction(method));
    if (method.isAccessible() && !KCallablesJvm.isAccessible(function)) {
        KCallablesJvm.setAccessible(function, true);
    }
    Mono<Object> mono = MonoKt.mono(context, (scope, continuation) ->
                KCallables.callSuspend(function, getSuspendedFunctionArgs(method, target, args), continuation))
            .filter(result -> !Objects.equals(result, Unit.INSTANCE))
            .onErrorMap(InvocationTargetException.class, InvocationTargetException::getTargetException);
    ... // 생략
    return mono;
}

설명에도 나와있듯이, suspend function을 실행시키고 Mono, Flux로 변환하여 반환한다.

Mono를 Coroutine에 맞춰 변환한 다음 suspend function을 실행하는 것이다.

 

이에 더불어 Reactor Context를 Coroutine Context로 변환해 주는 작업도 해준다.

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, // support for legacy mono in scope
    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)
}

internal fun CoroutineContext.extendReactorContext(extensions: ContextView): CoroutineContext =
    (this[ReactorContext]?.context?.putAll(extensions) ?: extensions).asCoroutineContext()

 

이를 통해 Reactor + Webflux와 Coroutine은 공존할 수 있게 된다.

 

📌 어디에 어떻게 Coroutine을 썼을까

SUSU 프로젝트에 코루틴이 처리된 곳은 크게 3개이다.

1. Blocking I/O 처리
2. 비동기 처리 
3. 이벤트 비동기 처리

 

🔵 Blocking I/O 처리

Coroutine은 여러 Dispatcher를 지원해 준다.

Dispatcher를 변경하여 코루틴을 실행하고 싶을 때, withContext를 사용한다.

SUSU에서는 MDC 값을 자식 코루틴에 전파하기 위해 아래와 같은 함수를 만들어서 사용했다.

suspend fun <T> withMDCContext(
    context: CoroutineContext = Dispatchers.IO,
    block: suspend () -> T,
): T {
    val contextMap = MDC.getCopyOfContextMap() ?: emptyMap()
    return withContext(context + MDCContext(contextMap)) { block() }
}

 

이 함수는 아래와 같이 사용되었다.

suspend fun findByIdOrNull(uid: Long): User? {
    return withMDCContext(Dispatchers.IO) { userRepository.findByIdOrNull(uid) }
}

IO 작업에 최적화된 Dispatcher.IO를 통해 IO 작업을 처리함으로써 더 효율적인 비동기 처리를 하기 위해 노력했다

 

🔵 비동기 처리

서비스의 latency를 줄이기 위해 다수의 작업을 비동기로 처리했다.

대표적으로 아래와 같다.

val userEnvelopeStatistic = parZip(
        { envelopeStatisticService.getRecentSpentFor1Year(user.uid) },
        { envelopeStatisticService.getMostFrequentRelationship(user.uid) },
        { envelopeStatisticService.getMostFrequentCategory(user.uid) },
        { envelopeStatisticService.getMaxReceivedEnvelope(user.uid) },
        { envelopeStatisticService.getMaxSentEnvelope(user.uid) }
    ) {
    	... // 생략
    }

한 값을 구하기 위해 5개의 작업이 비동기로 처리되고 값이 반환되도록 구성했다.

다수의 코루틴을 생성하여 비동기 처리한 후, 값이 모두 반환될 경우 이후 로직을 처리하도록 했다.

 

위에서 사용한 parzip의 경우 라이브러리를 이용한 것이다

defered와 awaitall을 이용해도 되지만, 이를 이용하면 더 편하다.

사용을 추천한다.

High-level concurrency | Arrow

 

🔵 이벤트 처리

이벤트 처리 함수의 경우, suspend function으로 변환이 불가능하다.

suspend function이 내부적으로 변환되며, Continuation 객체가 파라미터에 들어가게 되면, 에러가 발생한다.

그러므로 이벤트 처리를 비동기적으로 진행하려면 CoroutineScope를 이용하여 새로운 코루틴 스코프를 만들어 사용해야 한다.

CoroutineScope(...)로 사용해도 되지만, 이 또한 MDC 관련 이슈로 따로 만들어서 사용했다.

fun mdcCoroutineScope(context: CoroutineContext, traceId: String): CoroutineScope {
    val contextMap = MDC.getCopyOfContextMap() ?: emptyMap()
    contextMap.plus("traceId" to traceId)
    return CoroutineScope(context + MDCContext(contextMap))
}

 

이 함수는 아래와 같이 사용되었다.

@Component
class SystemActionLogEventListener(
    private val systemActionLogService: SystemActionLogService,
) {
    @EventListener
    fun subscribe(event: SystemActionLogEvent) {
        mdcCoroutineScope(Dispatchers.IO + Job(), event.traceId).launch {
            SystemActionLog(
                ipAddress = event.ipAddress,
                path = event.path,
                httpMethod = event.method,
                userAgent = event.userAgent,
                host = event.host,
                referer = event.referer,
                extra = event.extra
            ).run { systemActionLogService.record(this) }
        }
    }
}

이벤트 처리의 경우 다른 스레드에서 처리될 수 있다는 점을 감안하여, BaseEvent에 MDC traceId가 기본적으로 담기게 구성했다.

이에 이벤트 처리를 이용했더라도, 어떤 요청의 이벤트인지 traceId를 통해 알 수 있도록 구현했다.

 

 

자세한 코드는 아래에서 확인하면 된다

GitHub - ok-su-su/oksusu-susu-api: super 극락 Server