[Java] 자바 스레드와 동시성(5) - Thread Pool (스레드풀), Executor

1. Thread 병렬 작업 시 문제점

지난 몇 개의 과정에서 Thread를 생성하고 실행하는 방법에 대해 알아보았습니다. 이런 식으로 Thread를 실행했을 때 생기는 중요한 결핍 중 하나는 Thread 실행에 대해서 그 어떤 통제도 불가능하다는 점이 있습니다. 어느 특정 시점에서 얼마나 많은 Thread가 실행되고 있는지 알 수 없습니다.

예를 들어, 세 개의 Thread만 어떤 시점에서 실행하고 있게 만들고 싶다고 해보겠습니다. 이러한 조작은 Thread의 실행을 start 메소드를 이용하여 확립할 경우에는 매우 어렵습니다.

또 다른 문제도 있습니다. 예를 들어 Task1과 Task2 중 하나가 완료될 때까지 기다리고 싶다고 해보겠습니다.

Thread를 이용한 기본적인 것으로는 이 상황에 맞는 코드를 작성하는 것이 불가능합니다. 그러므로 Task1 또는 Task2가 실행이 완료될 때까지 기다리는 작업을 하고 싶다고 해보겠습니다.

task1.join();
task2Thread.join();

이런 식으로 코드를 작성하면, Task1과 Task2가 모두 완료될 때까지 기다렸다가 다음 코드가 실행되는 것을 확인 할 수 있습니다.

 

하지만 만약 Task1 혹은 Task2 중 하나가 완료된 후에 실행할 수 있도록 하는 코드를 작성하고 싶다면, join() 메소드 만으로는 구현이 불가능합니다.

또한 Thread를 실행하는 과정은 매우 지루한 과정입니다. 만약에 100개의 task를 실행하고 싶다면, 이와 같은 코드(Task1 task1 = new Task1(); task1.setPriority(1); task1.start();)를 모두 작성한 후에 시작을 실행해 주시면 됩니다. 다시, 이 모든 코드를 작성하고, 실행을 눌러주시면 됩니다. 이 코드의 묶음의 형식으로 처리하는 방법은 없습니다.

 

마지막으로, 가끔 Thread가 실행한 task의 결과를 반환하고 싶은 경우가 생길 수 있습니다. 우리가 지금까지 배운 이야기한 내용으로는 불가능한 내용입니다. 여기서 Executor Service, 즉 실행기를 사용할 수 있습니다. Executor Service는 지금까지 이야기 나눈 이 모든 기능을 아우를 수 있는 기능들을 제공합니다. 다수의 Thread를 한 번에 실행할 수 있고, Thread의 상태를 알 수 있도록 도와주며, 다수의 Thread를 한 번에 실행할 수 있고, Thread의 상태를 알 수 있도록 도와주며, Thread와 사용자가 다음과 같은 논리를 실행할 수 있도록 도와줍니다. Thread1 혹은 Thread2 혹은 Thread3가 완료되면 알려주도록 할 수 있습니다. 이들 중 어떤 Thread가 먼저 실행이 완료되었는지 알 수도 있습니다.

 

또한 모든 Thread가 실행이 완료될 때까지 기다리도록 단 한 문장으로 코드를 작성할 수 있습니다. 이 모든 것들이 Executor Service로 구현 가능한 것들입니다. 다음 단계에서는 Executor Service에 대한 내용을 시작해 보도록 하겠습니다.

 

여기서 Executor Service에 대해 설명하기 전에 스레드풀(ThreadPool)에 대해 알아야하므로, ThreadPool에 대해 먼저 알아보도록하겠습니다.

 

2. Thread Pool 이 나온 배경

데이터베이스 및 웹 서버와 같은 서버 프로그램은 여러 클라이언트의 요청을 반복적으로 실행하며 많은 수의 작은 작업들을 처리하는 데 중점을 둡니다. 서버 응용 프로그램을 이러한 요청을 처리하는 방식은 요청이 도착할 때마다 새 스레드를 만들고 새로 만든 스레드에서 받은 요청을 처리하는 것입니다.

 

이러한 방식은 구현이 간단해 보이지만 큰 단점이 있습니다. 모든 요청에 대해 새 스레드를 생성하는 서버는 실제 요청을 처리하는 것보다 스레드 생성 및 소멸에 더 많은 시간을 소비하고 더 많은 시스템 리소스를 소비합니다.

 

활성(active) 스레드는 시스템 리소스를 소비하기 때문에 동시에 너무 많은 스레드를 생성하는 JVM은 시스템의 메모리 부족(out of memory)을 유발할 수 있습니다. 이것은 생성되는 쓰레드의 수를 제한할 필요성을 느끼게 합니다.

 

3. Thread Pool 이란

스레드 풀은 이전에 생성된 스레드를 재사용하여 현재 작업을 실행하고 스레드 사이클 오버 헤드(thread cycle overhead) 및 리소스 스래싱(resource thrashing) 문제에 대한 해결책을 제공합니다. 요청이 도착하면 스레드가 이미 존재하기 때문에 스레드 생성으로 인한 지연이 제거되어 애플리케이션이 보다 신속하게 응답 할 수 있습니다.

• Java는 Executor 인터페이스(하위 인터페이스인 ExecutorService나, ThreadPoolExecutor 및 이 두 인터페이스를 모두 구현하는 클래스들)를 중심으로 하는 Executor 프레임워크를 제공합니다. Executor를 사용하면 Runnable 객체를 구현하고 Executor로 보내 실행하기만 하면 됩니다.
• 이를 통해 스레딩을 활용할 수 있게 해줍니다. Executor에서 스레드를 관리해주기 때문에 스레드의 내부 동작을 보다는 수행하려는 작업에 집중할 수 있습니다.
• 스레드 풀을 사용하려면 먼저 ExecutorService의 객체를 만들고 작업셋을 전달합니다. ThreadPoolExecutor 클래스를 사용하면 코어 및 최대 풀 크기를 설정할 수 있습니다. 특정 스레드에서 실행되는 실행 가능한 항목은 순차적으로 실행됩니다.

Thread Pool Initialization with size = 3 threads. Task Queue = 5 Runnable Objects

4. Executor Thread Pool Methods

메소드	설명
newFixedThreadPool(int)	고정된 크기의 스레드 풀을 생성합니다.
newCachedThreadPool()	필요에 따라 새 스레드를 생성하는 스레드 풀을 생성하지만 사용 가능한 경우 이전에 생성된 스레드를 재사용합니다.
newSingleThreadExecutor()	단일 스레드를 생성합니다.

고정 스레드 풀의 경우, 현재 모든 스레드가 executor에 의해 실행(스레드 풀에 있는모든 스레드가 active 상태일 경우)되고 있으면 추가로 들어온 작업들은 대기열에 배치되고 스레드가 유휴 상태가 되면 실행됩니다.

 

5. Executor Thread Pool Methods 예제

5.1 newSimgleThreadExecutor()

import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

public class ExecutorServiceMain {
    public static void main(String[] args) {
        // 1. ExecutorService 생성. newSingleThreadExecutor : 한번에 하나의 Thread를 실행하도록 하는 기능을 제공
        ExecutorService executorService = Executors.newSingleThreadExecutor();
        // 2. ExecutorService에 실행할 task를 넘겨줌
        executorService.execute(new Task1());
        executorService.execute(new Thread(new Task2()));

        // 3. ExecutorService 종료. 해당 처리를 해주지 않으면 프로그램이 종료되지 않음
        executorService.shutdown();
    }
}

Task1 Start
101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 
Task1 Finish

Task2 Start
201 202 203 204 205 206 207 208 209 210 211 212 213 214 215 216 217 218 219 220 221 222 223 224 225 226 227 228 229 230 231 232 233 234 235 236 237 238 239 240 241 242 243 244 245 246 247 248 249 250 251 252 253 254 255 256 257 258 259 260 261 262 263 264 265 266 267 268 269 270 271 272 273 274 275 276 277 278 279 280 281 282 283 284 285 286 287 288 289 290 291 292 293 294 295 296 297 298 299 
Task2 Finish

한 번에 한 개의 스레드만 실행 할수 있기 때문에 Task1이 종료 된 후에 Task2가 실행되는 것을 확인할 수 있습니다.

 

import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

public class ExecutorServiceMain {
    public static void main(String[] args) {
        // 1. ExecutorService 생성. newSingleThreadExecutor : 한번에 하나의 Thread를 실행하도록 하는 기능을 제공
        ExecutorService executorService = Executors.newSingleThreadExecutor();
        // 2. ExecutorService에 실행할 task를 넘겨줌
        executorService.execute(new Task1());
        executorService.execute(new Thread(new Task2()));

        //Task3
        System.out.println("Task3 Start");
        for(int i = 301; i <= 399; i++) {
            System.out.print(i + " ");
        }
        System.out.println("\\nTask3 Finish");

        // 3. ExecutorService 종료. 해당 처리를 해주지 않으면 프로그램이 종료되지 않음
        executorService.shutdown();
    }
}

Task1 Start
Task3 Start
101 102 103 104 105 106 301 107 302 108 303 109 304 110 305 306 307 111 308 112 309 113 310 114 311 115 312 116 313 117 314 315 118 316 317 318 319 119 320 120 321 121 322 122 323 123 324 124 325 125 326 126 327 328 127 329 128 330 129 331 130 332 131 333 132 334 133 335 134 336 135 337 136 338 339 137 340 341 138 342 343 344 345 346 347 348 349 350 351 352 139 140 141 353 142 354 143 355 144 356 145 357 146 358 147 359 148 360 149 361 150 362 363 364 365 366 367 151 368 369 370 371 372 373 152 374 153 375 154 376 155 377 156 157 158 159 160 161 378 162 379 163 380 164 165 166 167 381 168 382 169 383 170 384 171 172 385 173 386 174 387 175 176 177 178 179 388 180 389 181 390 182 391 183 184 185 186 187 188 189 190 191 192 193 194 392 195 393 196 197 198 394 395 396 397 199 398 
Task1 Finish
399 
Task2 Start

Task3 Finish
201 202 203 204 205 206 207 208 209 210 211 212 213 214 215 216 217 218 219 220 221 222 223 224 225 226 227 228 229 230 231 232 233 234 235 236 237 238 239 240 241 242 243 244 245 246 247 248 249 250 251 252 253 254 255 256 257 258 259 260 261 262 263 264 265 266 267 268 269 270 271 272 273 274 275 276 277 278 279 280 281 282 283 284 285 286 287 288 289 290 291 292 293 294 295 296 297 298 299 
Task2 Finish

Task1 시작되고 Task3가 시작된 것을 확인 할 수 있고. 위 예제와 동일하게 Task1의 실행이 종료된 후에 Task2가 실행된 것을 알 수 있습니다. Task1과 Task2는 executorService를 통해 실행되었고, Task3는 main 메소드에서 실행되었습니다.

 

5.2 newFixedThreadPool(int)

실행 플로우

1. 실행할 태스크(Runnable Object) 생성
2. Executor를 사용하여 Executor Pool 생성
3. Executor Pool에 작업 전달
4. 실행자 풀 종료

코드

import java.text.SimpleDateFormat;
import java.util.Date;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

// 1. 실행할 태스크(Runnable Object) 생성
class Task implements Runnable {
    private String name;
    public Task(String s) {
        name = s;
    }

    public void run() {
        try {

            // name 필드값을 출력하고 1초 동안 sleep() 5회 반복
            for (int i = 0; i <= 5; i++) {
                if (i == 0) {
                    Date d = new Date();
                    SimpleDateFormat ft = new SimpleDateFormat("hh:mm:ss");
                    System.out.println("Initialization Time for task name - " + name + " = " + ft.format(d));
                } else {
                    Date d = new Date();
                    SimpleDateFormat ft = new SimpleDateFormat("hh:mm:ss");
                    System.out.println("Executing Time for task name - " + name + " = " + ft.format(d));
                }
                Thread.sleep(1000);
            }
            System.out.println(name + " complete");
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }
}

public class FixedThreadMain {
    static final int MAX_T = 3; // 스레드 풀의 최대 스레드 수

    public static void main(String[] args) {
        // Task 5개 생성
        Runnable r1 = new Task("task 1");
        Runnable r2 = new Task("task 2");
        Runnable r3 = new Task("task 3");
        Runnable r4 = new Task("task 4");
        Runnable r5 = new Task("task 5");

        // 2. MAX_T값 만큼 스레드 풀을 생성합니다. 이 때 스레드 풀내 스레드 개수는 고정되어있음
        ExecutorService pool = Executors.newFixedThreadPool(MAX_T);

        // 3. Task 개체를 풀에 전달하여 실행할 수 있습니다.
        pool.execute(r1);
        pool.execute(r2);
        pool.execute(r3);
        pool.execute(r4);
        pool.execute(r5);

        // 4. pool shutdown
        pool.shutdown();
    }
}

Initialization Time for task name - task 3 = 05:02:17
Initialization Time for task name - task 1 = 05:02:17
Initialization Time for task name - task 2 = 05:02:17
Executing Time for task name - task 3 = 05:02:18
Executing Time for task name - task 2 = 05:02:18
Executing Time for task name - task 1 = 05:02:18
Executing Time for task name - task 1 = 05:02:19
Executing Time for task name - task 2 = 05:02:19
Executing Time for task name - task 3 = 05:02:19
Executing Time for task name - task 3 = 05:02:20
Executing Time for task name - task 1 = 05:02:20
Executing Time for task name - task 2 = 05:02:20
Executing Time for task name - task 1 = 05:02:21
Executing Time for task name - task 2 = 05:02:21
Executing Time for task name - task 3 = 05:02:21
Executing Time for task name - task 2 = 05:02:22
Executing Time for task name - task 1 = 05:02:22
Executing Time for task name - task 3 = 05:02:22
task 3 complete
task 2 complete
task 1 complete
Initialization Time for task name - task 5 = 05:02:23
Initialization Time for task name - task 4 = 05:02:23
Executing Time for task name - task 4 = 05:02:24
Executing Time for task name - task 5 = 05:02:24
Executing Time for task name - task 4 = 05:02:25
Executing Time for task name - task 5 = 05:02:25
Executing Time for task name - task 5 = 05:02:26
Executing Time for task name - task 4 = 05:02:26
Executing Time for task name - task 5 = 05:02:27
Executing Time for task name - task 4 = 05:02:27
Executing Time for task name - task 5 = 05:02:28
Executing Time for task name - task 4 = 05:02:28
task 4 complete
task 5 complete

프로그램 실행에서 알 수 있듯이 Task4와 Task5는 스레드 풀 내의 스레드가 유휴 상태가 될 때만 실행됩니다. 그때까지 추가 작업은 대기열에 배치됩니다.

Thread Pool executing first three tasks

Thread Pool executing task 4 and 5

💡 ThreadPool을 사용할 때의 주요 이점 중 하나는 한 번에 100개의 요청을 처리하고 싶지만 JVM 과부하를 줄이기 위해 동일한 요청에 대해 100개의 스레드를 생성하고 싶지 않을 때입니다. 이 방식을 사용하여 10개의 스레드로 구성된 ThreadPool을 만들고 이 ThreadPool에 100개의 요청을 제출할 수 있습니다. ThreadPool은 한 번에 10개의 요청을 처리하기 위해 최대 10개의 스레드를 생성합니다. 단일 스레드의 프로세스 완료 후, ThreadPool은 내부적으로 이 스레드에 11번째 요청을 할당합니다. 나머지 모든 요청에 대해 계속 동일한 작업을 수행합니다.

 

6. Thread Pool 문제점

1. 교착 상태(Deadlock): 모든 멀티 스레드 프로그램에서 교착 상태가 발생할 수 있지만 스레드 풀내의 스레드에서 교착 상태 발생 시 또 다른 교착 상태를 유발합니다. 이 경우 실행 중인 모든 실행 스레드가 실행을 위해 스레드를 사용할 수 없어 대기열에서 대기 중인 블락된 스레드의 결과를 계속 대기 하고 있는 문제가 발생합니다.
2. 스레드 누수(Thread Leakage) : Thread Leakage는 스레드가 풀에서 제거되어 작업을 실행하지만 반환되지 않은 경우 발생합니다. 예를 들어 스레드가 예외를 throw하고 풀 클래스가 이 예외를 catch하지 않으면 스레드는 종료되어 스레드 풀의 크기가 1 감소합니다. 이 작업이 여러 번 반복되면 결국 풀이 비어 있게 되고 다른 요청을 실행하는 데 사용할 수 있는 스레드가 없어지게 됩니다.
3. 리소스 스레싱(Resource Thrashing): 스레드 풀 크기가 매우 크면 스레드 간 컨텍스트 전환에 시간이 낭비됩니다. 최적의 수보다 많은 스레드가 있으면 설명된 대로 자원 낭비로 이어지는 기아 문제(starvation problem)가 발생할 수 있습니다.

 

7. Thread Pool 사용 시 주의 사항

1. 다른 작업의 결과를 동시에 기다리는 작업을 대기열에 넣지 마십시오. 이로 인해 위에서 설명한 교착 상태가 발생할 수 있습니다.
2. 수명이 긴 작업에 스레드를 사용한다면 주의 깊게 살펴봐야합니다. 이로 인해 스레드가 영원히 대기하고 결국 리소스 누수가 발생할 수 있습니다.
3. 스레드 풀은 마지막에 명시적으로 종료되어야 합니다. 이 작업을 하지 않으면 프로그램이 계속 실행되고 끝나지 않습니다. 풀에서 shutdown()을 호출하여 Excutor를 종료합니다. 종료 후 이 Excutor에 다른 작업을 보내려고 하면 RejectedExecutionException이 발생합니다.
4. 스레드 풀을 효과적으로 조정하려면 TasK를 이해해야 합니다. 동일한 스레드 풀을 사용하는 Task가 서로 매우 대조적이라면 작업을 적절하게 조정하기 위해 Task 유형에 따라 다른 스레드 풀을 사용하는 것이 좋습니다.
5. JVM에서 실행할 수 있는 최대 스레드 수를 제한하여 JVM의 메모리 부족 가능성을 줄이는 것이 좋습니다.
6. 처리를 위해 새 스레드를 생성하는 반복문을 구현해야 하는 경우 ThreadPool이 최대 제한에 도달한 후 새 스레드를 생성하지 않기 때문에 ThreadPool을 사용하면 더 빠르게 처리하는 데 도움이 됩니다.
7. 스레드 처리가 완료된 후 ThreadPool은 동일한 스레드를 사용하여 다른 프로세스를 수행할 수 있습니다. (따라서 다른 스레드를 생성하는 시간과 리소스를 절약할 수 있습니다.)

 

8. Thread Pool 튜닝

스레드 풀의 최적 크기는 사용 가능한 프로세서의 수와 작업의 특성에 따라 다릅니다. 계산(computation) 유형 프로세스만 있는 대기열에 대한 N개의 프로세서(CPU) 시스템에서 N 또는 N+1로 최대 스레드 풀 사이즈를 설정하면 최대 효율성을 달성할 수 있습니다. 그러나 작업은 I/O를 기다릴 수 있으며 이러한 경우 비율을 고려합니다. 따라서 아래와 같이 아래의 계산식에 따라 적정 스레드 수를 맞춰주는 것이 좋습니다

적정 스레드 수 = 사용 가능한 CPU 코어 수 * (1 + 요청에 대한 대기 시간/서비스 시간)

 

9. 정리

스레드 풀은 서버 응용 프로그램을 구성하는 데 유용한 도구입니다. 개념적으로는 매우 간단하지만 교착 상태, 리소스 스래싱과 같이 구현 및 사용할 때 주의해야 할 몇 가지 문제가 있습니다. Executor Service를 사용하면 구현이 더 쉬워집니다.

 

---

참고

• https://www.geeksforgeeks.org/thread-pools-java/
• https://www.baeldung.com/thread-pool-java-and-guava