주기적으로 작업을 모아 처리하는 BatchLoader 구현하기

출처: https://javascript.plainenglish.io/adding-skeleton-loading-animation-with-css-e6833f6e1d0a

 

특정 주기마다 작업을 모아서 실행하는 Batch Loader를 Kotlin으로 구현해봤다. 여러가지 쓸모가 있겠지만, 나는 Batch Loader를 UI에서 목록을 표시하고, 표시된 아이템만 따로 불러오는데 유용하게 사용하고 있다. 

 

아이디어

대부분의 아이디어는 이 Typescript BatchLoader에서 얻었다.

잠시 이론적으로 짚고 넘어가면 코드 이해가 쉬울 것같다.

BatchLoader는 일정 주기안에 일어나는 모든 요청들을 받아서 별도의 Queue에 쌓아둔다. 요청을 받으면 CompletableDeferred를 반환한다. CompletableDeferred는 콜백 기반의 비동기 통신 시 성공, 혹은 실패에 대한 응답을 다룰때 유용한 클래스로, Javascript에서의 Promise와 비슷한 역할을 한다. 여기서도 이후에 요청자가 응답을 정상적으로 받을 수 있도록 해준다.

 

주기가 지나면 Queue는 비워지고, 그동안 쌓였던 모든 요청들을 한꺼번에 처리한다. 예시는 서버에 HTTP Request를 보낸다. 그동안 새로운 요청 Req4는 Queue에 계속 쌓이며, Req4는 이번 요청 주기를 놓쳤으므로 다음 주기에 처리된다.

 

벌크처리된 응답으로부터 개별 요청에 대한 CompletableDeferred를 완료 시키고, 요청자는 await()로 모든 처리가 끝날때까지 기다리면된다.

 

구현

 

구현에서는 배치처리를 위한 별도의 CoroutineScope를 가지도록 했으며, 작업에 대한 runner (CompletableDeferred)에 접근/생성하는 부분을 임계영역으로 처리했다. runner의 레퍼런스가 있다면 queue에만 작업을 넣어두고 그 runner의 작업이 끝나길 기다리면되고, runner가 없다면 만든다.

 

실행

fun main(args: Array<String>) = runBlocking {
    val loader = BatchLoader<Int, String>(
        executor = Executors.newSingleThreadExecutor(),
        batch = { keys ->
            Result.success(keys.associateWith { "<$it>" })
        }
    )

    val j = CoroutineScope(Dispatchers.Default).launch {
        run(loader, 1, 3)
    }

    val k = CoroutineScope(Dispatchers.IO).launch {
        run(loader, 4, 8)
    }

    val q = CoroutineScope(Dispatchers.IO).launch {
        run(loader, 6, 10)
    }

    j.join()
    k.join()
    q.join()
}

private suspend fun run(loader: BatchLoader<Int, String>, start: Int, end: Int) {
    for (i in start..end) {
        val result = loader.load(i)
        result.invokeOnCompletion {
            if (it == null) {
                println("$i: ${result.getCompleted()}")
            } else {
                println("Error: ${it.message}")
            }
        }
    }
}