Future를 이용한 Thread Pool 작업 완료 통보

2021.12.13 - [Android : Kotlin/Deep-Dive] - 안드로이드 Thread Pool

(위 글에서 이어집니다.)

1. 블로킹 식 작업 완료 통보

생성한 Thread Pool에서 작업 처리를 요청할 때 submit( ) Method는 작업 처리 결과를 Future 타입의 객체로 반환한다. submit( ) Method에 사용되는 파라미터는 아래와 같다.

반환되는 Future 객체는 submit( )을 통해 완료된 단일 작업 결과가 아니라 모든 작업이 완료될 때까지 기다렸다가 최종 결과를 얻는 데 사용한다. 그렇기 때문에 Future 객체를 지연 완료 객체 (Pending Completion Object)라고도 한다.

 

Future의 get( ) Method를 호출하면 Thread가 수행하는 작업이 완료될 때까지 블로킹되었다가 모든 작업을 완료한 후 처리 결과를 반환한다. 

첫 번째 Method는 작업이 완료될 때까지 블로킹되었다가 V를 통해 처리결과를 반환한다.

두 번째 Method는 timeout 전에 작업이 완료되면 V를 통해 처리결과를 리턴 하지만, 작업이 완료되지 않으면 TimeoutException을 발생시킨다.

아래 표는 세 가지 submit( ) Method에서 Future의 get( ) Method가 리턴하는 타입을 보인다.

출처 : https://palpit.tistory.com/732

블로킹 방식의 작업 완료 통보에서는 작업을 처리하는 Thread가 작업을 모두 완료하기 전까지는 get( ) Method가 블로킹되기 때문에 다른 코드를 실행할 수 없다. 안드로이드에서는 Main Thread (UI Thread)에서 호출되면 안 된다. 블로킹 상태가 되기 때문에 UI와 관련된 이벤트를 처리할 수 없기 때문이다. 그렇기 때문에 get( ) Method를 호출하는 Thread는 새로운 Thread이거나 Thread Pool 내부의 또 다른 Thread가 되어야 한다. 

새로운 Thread에서 get( ) 호출

Thread(Runnable {
    try {
        future.get()
    } catch (e : Exception) {
        e.printStackTrace()
    }
}).start()

Thread Pool 내부의 또 다른 Thread에서 get( ) 호출

executorService.submit(Runnable {
    try {
        future.get()
    } catch (e: Exception) {
        e.printStackTrace()
    }
})

 

Future 객체는 작업 결과를 얻기 위해 사용되는 get( ) Method 이외에도 아래와 같은 Method를 제공한다.

cancel( ) Method는 해당 작업의 실행을 취소한다. 해당 작업을 실행하는 Thread가 중단되어야 하는 경우 파라미터로 true를 넣는다. false인 경우 진행 중인 작업을 완료할 수 있다. Boolean을 반환하는데 이미 작업이 정상적으로 완료되었기 때문에 작업을 취소할 수 없는 경우 false를 반환한다.

isCancelled( ) Method는 작업이 정상적으로 취소되었는지 여부를 Boolean으로 반환한다.

isDone( ) Method는 작업 처리가 완료되었는지 여부를 Boolean으로 반환한다.

---

2. 반환 값이 없는 작업의 완료 통보 (Runnable)

이전 포스팅에서 Runnable과 Callable의 차이를 설명했다. 반환 값이 없는 작업일 경우 Runnable 객체로 생성을 하면 된다. 이 작업의 요청은 submit(Runnable task) Method를 사용하면 된다. Runnable 객체는 반환 값이 없지만 submit( ) Method를 사용하면 Future 객체를 반환하는데, Future 객체로 해당 Thread가 Runnable 작업 처리를 정상적으로 완료했는지 도중에 예외가 발생했는지를 확인할 수 있다.

아래와 같이 사용할 수 있다.

 

private fun executeTask() {
    val executorService = Executors.newFixedThreadPool(
        Runtime.getRuntime().availableProcessors()
    )

    val runnableTask = Runnable {
        var sum = 0
        for (i in 1..10) {
            sum += i
        }
        Log.e("Result", sum.toString())
    }

    val future = executorService.submit(runnableTask)
    Log.e("Task", "Start")

    try {
        future.get()
    } catch (e: InterruptedException) {
        e.printStackTrace()
    } catch (e: ExecutionException) {
        e.printStackTrace()
    }

    executorService.shutdown()
    Log.e("Task", "Done!")
}

 

실행 결과는 다음과 같다.

---

3. 반환 값이 있는 작업의 완료 통보 (Callable)

Thread Pool의 Thread가 작업을 완료한 후 결과를 반환하게 하려면 작업을 Callable로 생성하면 된다. Runnable과는 다르게 Callable은 call( ) Method에 반환 값이 존재하는데, 이 값의 반환 타입과 Future의 제네릭 타입이 일치해야 한다.

아래와 같이 사용할 수 있다.

 

private fun executeTask() {
    val executorService = Executors.newFixedThreadPool(
        Runtime.getRuntime().availableProcessors()
    )

    val callableTask = Callable {
        var sum = 0
        for (i in 1..10) {
            sum += i
        }

        sum  // Int
    }

    val future = executorService.submit(callableTask)
    Log.e("Task", "Start")

    try {
        val callableResult = future.get()  // Int
        Log.e("Result", callableResult.toString())
    } catch (e: InterruptedException) {
        e.printStackTrace()
    } catch (e: ExecutionException) {
        e.printStackTrace()
    }

    executorService.shutdown()
    Log.e("Task", "Done!")
}

 

실행 결과는 다음과 같다.

---

 

4. 외부 객체에 작업 처리 결과 저장

Thread에서 작업한 결과를 특정 외부 Class에 저장할 수도 있다. 결과를 외부에 저장하게 되면 어플리케이션에서 해당 객체를 이용하여 또 다른 작업을 진행할 수 있다. 이때, 이 객체는 공유 객체이며 두 개 이상의 Thread의 작업 결과를 취합하는 목적으로 사용된다.

 

ExecutorService의 submit(Runnable task, V result) Method를 사용한다. V는 result의 타입이다. Method를 호출하면 Future<V>가 반환되는데 이때 Future의 get( )을 호출하면 Thread가 작업을 모두 완료할 때까지 블로킹되었다가 작업을 완료하면 V 타입 객체를 반환한다. 

작업 객체는 Runnable로 생성하는데 결과를 저장하기 위해 외부 객체를 사용해야 하기 때문에 생성자로 해당 객체를 넣어야 한다.

아래와 같이 사용할 수 있다.

 

class MainActivity : AppCompatActivity() {
    override fun onCreate(..) {
        button.setOnClickListener(
            executeTask()
        }
    }
    
    private fun executeTask() {
        val executorService = Executors.newFixedThreadPool(
            Runtime.getRuntime().availableProcessors()
        )

        var result = Result()
        val task1 = Task(result)
        val task2 = Task(result)

        val future1 = executorService.submit(task1, result)
        val future2 = executorService.submit(task2, result)
        Log.e("Task", "Start")

        try {
            result = future1.get()
            result = future2.get()
            Log.e("Result", result.resultValue.toString())
        } catch (e: InterruptedException) {
            e.printStackTrace()
        } catch (e: ExecutionException) {
            e.printStackTrace()
        }

        executorService.shutdown()
        Log.e("Task", "Done!")
    }

    inner class Task(result: Result): Runnable {
        private var result = Result()

        init {
            this.result = result
        }

        override fun run() {
            var sum = 0
            for (i in 1..10) {
                sum += i
            }
            result.addValue(sum)
        }
    }
}

class Result {
    var resultValue = 0
    fun addValue(value: Int) {
        resultValue += value
    }
}

두 개의 작업(Task)를 ThreadPool에 요청하고 각각의 Thread가 작업을 완료한 후 결과 값을 Result 객체에서 합산한다. 

실행 결과는 다음과 같다.

---

5. 완료되는 작업 순으로 통보

Thread Pool에 작업을 요청하는 순서대로 작업 처리가 완료되는 것은 당연히 아니다. 작업의 크기와 스케줄링 방식에 따라 가장 먼저 요청한 작업이 나중에 완료될 수도 있다. CompletionService는 ThreadPool에서 처리가 완료된 작업의 결과를 통보받을 수 있는 클래스이다.

아래 Method를 사용하여 처리가 완료된 작업을 가져올 수 있다.

CompletionService는 ExecutorCompletionService<V>로 구현할 수 있다. 객체를 생성할 때에는 생성자로 ExecutorService를 넘긴다.

작업 처리 요청을 위해 CompletionService의 submit( ) Method를 사용하고 처리가 완료된 작업의 Future를 얻기 위해 poll( ) 또는 take( ) Method를 사용한다.

자세한 설명은 아래 표를 참고한다.

출처 :&nbsp;https://palpit.tistory.com/732

아래와 같이 완료된 작업의 결과부터 통보할 수 있다.

 

private fun executeTask() {
    val executorService = Executors.newFixedThreadPool(
        Runtime.getRuntime().availableProcessors()
    )

    val completionService: CompletionService<Int> = ExecutorCompletionService(executorService)

    val callableTask1 = Callable {
        val list = listOf("a", "b", "c")
        var sum = 0
        for (i in list) {
            sum += list.indexOf(i)
        }

        sum
    }

    val callableTask2 = Callable {
        var sum = 0
        val integerList = listOf(1, 2, 3, 4, 5)
        for (i in integerList) {
            sum += integerList.reversed().indexOf(i)
        }

        sum
    }

    val callableTask3 = Callable {
        var sum = 0
        for (i in 1..10) {
            sum += i
        }

        sum
    }

    completionService.submit(callableTask1)
    completionService.submit(callableTask2)
    completionService.submit(callableTask3)

    Log.e("Task", "Start")

    executorService.submit(Runnable {
        while (true) {
            try {
                val future = completionService.take()
                val result = future.get()
                Log.e("Result", result.toString())
            } catch (e: Exception) {
                e.printStackTrace()
            }
        }
    })

    executorService.shutdownNow()
    Log.e("Task", "Done!")
}

 

take( ) Method를 호출하여 완료된 Callable 작업이 있을 때 까지 블로킹되었다가 완료된 작업의 Future를 get( ) Method를 이용해 결과값을 얻어낸다.

 

실행 결과는 아래와 같다.

완료된 Task의 순서는 callableTask3, callableTask1, callableTask2이다.

---

References

https://developer.android.com/reference/kotlin/java/util/concurrent/Future

Future  |  Android Developers

https://palpit.tistory.com/732

[Java] 멀티 스레드 - 스레드풀(ThreadPool)

https://codechacha.com/ko/java-future/

Java - Future 사용 방법