Future를 활용하여 Timeout 구현하기
개요
코루틴 Timeout이 제대로 동작하지 않은 이유라는 글을 작성하며 코루틴을 활용하여 Timeout을 구현해 보았습니다.
동기적인 코드가 섞여있는 상황이라면 코루틴에 협조적이지 않아 타임아웃이 잘 동작하지 않았던 이슈가 있었고 새로운 스레드를 할당하여 문제를 해결했지만 타임아웃이 발생하더라도 비동기적으로 실행되는 코드들이 계속 동작하는 문제가 있었습니다.
이 문제를 해결하기 위해서 Java 1.5부터 Future를 활용해 볼 수 있습니다.
Future란?
<T> Future<T> submit(Callable<T> task);
Future<?> submit(Runnable task);ExecutorService에서 Runnable, Callable Task에 대해 submit 메서드를 활용하면 Future에 대한 반환값을 받을 수 있습니다.
ExecutorService는 새로운 스레드를 만들어 할당하기 때문에 해당 작업이 어떻게 되는지, 결과에 대한 값을 받아오기 위해 Future를 활용합니다.
V get() throws InterruptedException, ExecutionException;Future의 get 메서드를 활용하면 해당 결과가 완료될 때까지 기다린 뒤 결괏값을 받아올 수 있습니다.
V get(long timeout, TimeUnit unit)
throws InterruptedException, ExecutionException, TimeoutException;이때 Future의 get 메서드에는 timeout을 지정해서 결과를 가져올 수 있습니다.
정해진 시간이 넘어가게 되면 TimeoutException이 발생하게 됩니다.
Future로 Timeout 구현하기
fun <T> executeWithTimeoutAndCancellationAsync(timeMillis: Long, block: () -> T): T {
val result: Future<T> = executors.submit(
Callable { block() }
)
try {
return result.get(timeMillis, TimeUnit.MILLISECONDS)
} catch (e: ExecutionException) {
throw e.cause ?: e
} catch (e: TimeoutException) {
logger.info("Future의 타임아웃을 이용해서 쓰레드를 종료시킨다")
result.cancel(true)
logger.info("취소 여부: ${result.isCancelled}")
logger.info("끝남 여부: ${result.isDone}")
throw e
} catch (e: Exception){
throw e
}
}타임아웃에 대한 인자를 받은 뒤, 실행할 메서드를 block 인자로 받습니다.
이후 해당작업은 executors에 의해 다른 스레드에서 실행되고 해당결과는 Future 객체로 확인할 수 있습니다.
이후에는 결과를 타임아웃을 지정해서 기다립니다.
만약 결과를 가져오기 위한 작업을 수행하다가 예외가 발생하면 ExecutionException이 발생하는데 cause를 활용하여 기존의 예외를 다시 전파하며, 타임아웃 시간이 지나서 TimeoutException이 발생한 경우라면 Future의 cancel 메서드를 호출해서 실행되고 있는 작업을 취소합니다.
주의할 점으로 cause를 활용하지 않으면 기존의 예외가 무시되어 메서드를 활용하는 쪽에서 ExecutionException만 알게 될 수 있습니다.
테스트로 검증하기
@Test
fun `Future를 활용하여 비동기작업도 정리하기`() {
val startTime = System.currentTimeMillis()
try {
val result = executeWithTimeoutAndCancellationAsync(1500L) {
var i = 1L
while (true) {
sleep(200L)
println("count : ${i++}")
}
}
println(result)
} catch (e: TimeoutException) {
val endTime = System.currentTimeMillis()
println("Timeout이 발생하는 시점은 시스템이 시작되고 ${endTime - startTime} 이후 이다.")
}
sleep(5000L)
}200ms의 Thread.sleep()과 함께 무한히 실행되는 작업을 1500ms의 타임아웃을 지정하여 넘겨보았습니다.
예상으로는 7번 정도의 count가 올라가게 되고 그 이후에는 작업이 취소되어 정리되어야 합니다.
결과
실제로 7번의 count가 올라가고 그 뒤 작업은 취소됨을 확인할 수 있습니다.
만약 작업을 취소하지 않는다면
result.cancel(true)
만약 해당 부분을 주석처리한다면 어떻게 될까요?
결과
타임아웃이 발생하였지만 계속하여 내부 작업이 실행되는 것을 확인할 수 있습니다.
Coroutine도 비슷하게 작업을 취소하면 되지 않을까?
Future를 활용하여 비동기 내부 작업을 취소해 보았습니다.
문득 Coroutiine도 비슷하게 작업을 취소하면 내부작업이 정리될 것 같은 생각이 들었습니다.
fun <T> asyncAwaitWithTimeout(timeMillis: Long, block: suspend CoroutineScope.() -> T): T {
return runBlocking {
var job: Job? = null
try {
withTimeout(timeMillis) {
job = CoroutineScope(Dispatchers.IO).async {
block()
}
return@withTimeout (job as Deferred<T>).await()
}
}catch (e: TimeoutCancellationException){
println("timeout 발생했음")
job?.cancelAndJoin()
job?.isActive
throw e
}catch (e: Exception){
throw e
}
}
}async에 대한 실행결과를 Deffered Job으로 받아주고 TimeoutCancellationException이 발생하였을 때 해당 Job을 취소하도록 구성하였습니다.
결과
타임아웃이 count가 7개 증가한 뒤 발생하였지만 작업은 무한하게 실행됩니다.
코루틴에서의 취소요청을 보냈지만 내부함수가 suspend로 구성되어 있지 않아 취소에 협조적이지 않기 때문입니다.
만약 내부함수가 sleep()이 아닌 delay로 구성되어 있었다면 어떨까요?
7번의 count가 올라간 뒤 타임아웃이 발생하였고 해당 Job의 isActive도 false인 것을 테스트를 통해 확인할 수 있습니다.
CompletableFuture의 cancel
Future 대신 Java 1.8의 CompletableFuture를 활용해볼까 고민하기도 하였습니다.
하지만 CompletableFuture의 cancel은 실행중인 스레드에 InterruptedException을 발생시킬 수 없습니다.
mayInterruptIfRunning 인자를 메서드에서 활용하지 않으며 docs에서도 효과가 없다고 기술되어 있습니다.