이전 블로그 글인 "자바 스레드의 대기/통지 모델"을 참고하시기 바랍니다. 이제 대기/통지 메커니즘을 활용한 스레드 풀을 구현해보겠습니다.
코드는 GitHub에 공개되어 있습니다: git@github.com:jiulu313/ThreadPool.git
작업은 단순한 문자열 출력으로 시간이 소요되는 작업을 시뮬레이션합니다. 주요 코드는 다음과 같습니다.
- 작업 인터페이스 정의
1 /*
2 * 작업 인터페이스
3 */
4 public interface RunnableTask {
5 void execute();
6 }
- 구체적인 작업 구현
1 /*
2 * 구체적인 작업
3 */
4 public class PrintJob implements RunnableTask{
5
6 // 현재 스레드 이름 출력 후 1초 대기하여 시간 소요 작업 시뮬레이션
7 @Override
8 public void execute() {
9 System.out.println("작업: " + Thread.currentThread().getName());
10 try {
11 Thread.sleep(1000);
12 } catch (InterruptedException e) {
13 e.printStackTrace();
14 }
15 }
16 }
- 작업자 스레드 구현
1 /*
2 * 작업자 스레드
3 */
4 public class WorkerThread implements Runnable {
5 // 스레드 실행 여부
6 private boolean isActive = true;
7
8 // Thread 객체 저장
9 private Thread workerThread;
10
11 // 작업 대기열 저장 (동기화용)
12 private LinkedList<RunnableTask> taskQueue;
13
14 public void setWorkerThread(Thread thread) {
15 this.workerThread = thread;
16 }
17
18 public void setTaskQueue(LinkedList<RunnableTask> queue) {
19 this.taskQueue = queue;
20 }
21
22 // 작업자 스레드 시작
23 public void start() {
24 if (workerThread != null) {
25 workerThread.start();
26 }
27 }
28
29 // 작업자 스레드 종료
30 public void terminate() {
31 isActive = false;
32 workerThread.interrupt();
33 }
34
35 @Override
36 public void run() {
37 while (isActive) {
38 RunnableTask job = null;
39
40 // 공유 자원 동기화 (작업 대기열)
41 synchronized (taskQueue) {
42
43 // 조건 만족 시 대기
44 while (taskQueue.isEmpty()) {
45 try {
46 // 스레드 대기 및 락 해제
47 taskQueue.wait();
48 } catch (InterruptedException e) {
49 // 외부 중단 감지
50 Thread.currentThread().interrupt();
51 return;
52 }
53 }
54
55 // 조건 충족 시 작업 처리
56 job = taskQueue.removeFirst();
57 }
58
59 // 작업 실행
60 if (job != null) {
61 job.execute();
62 }
63 }
64 }
65 }
- 스레드 풀 생성
1 import java.util.ArrayList;
2 import java.util.LinkedList;
3 import java.util.List;
4
5 public class CustomThreadPool implements ThreadPool {
6 private int maxThreadCount = 10;
7 private int minThreadCount = 1;
8 private int defaultThreadCount = 5;
9
10 // 작업 목록
11 private LinkedList<RunnableTask> taskQueue = new LinkedList<>();
12
13 // 작업자 스레드 목록
14 private LinkedList<WorkerThread> threads = new LinkedList<>();
15
16 // 현재 스레드 수
17 private int threadCount = defaultThreadCount;
18
19
20 @Override
21 public void execute(RunnableTask task) {
22 // 작업 추가 후 통지
23 if (task != null) {
24 synchronized (taskQueue) {
25 // 작업 추가
26 taskQueue.addLast(task);
27 // 대기 중인 스레드에게 통지
28 taskQueue.notify();
29 }
30 }
31 }
32
33 // 스레드 풀 종료
34 @Override
35 public void shutdown() {
36 for (WorkerThread thread : threads) {
37 thread.terminate();
38 }
39 }
40
41 // 작업자 스레드 초기화
42 public void initialize(int count) {
43 if (count > maxThreadCount) {
44 count = maxThreadCount;
45 } else if (count < minThreadCount) {
46 count = minThreadCount;
47 } else {
48 count = defaultThreadCount;
49 }
50
51 for (int i = 0; i < threadCount; i++) {
52 // 작업자 스레드 생성
53 WorkerThread worker = new WorkerThread();
54
55 // 스레드 목록 추가
56 threads.add(worker);
57
58 // 새로운 스레드 생성 및 설정
59 Thread thread = new Thread(worker);
60
61 // 스레드 객체 설정
62 worker.setWorkerThread(thread);
63
64 // 작업 대기열 설정
65 worker.setTaskQueue(taskQueue);
66 }
67 }
68
69 // 스레드 풀 시작
70 public void start(){
71 if(threads != null){
72 for(WorkerThread thread : threads){
73 // 작업자 스레드 시작
74 thread.start();
75 }
76 }
77 }
78
79 // 작업자 스레드 추가 (최대 수 제한)
80 @Override
81 public void increaseWorkers(int count) {
82 if (count <= 0) {
83 return;
84 }
85
86 int available = maxThreadCount - threadCount;
87 if (count > available) {
88 count = available;
89 }
90
91 for (int i = 0; i < count; i++) {
92 WorkerThread worker = new WorkerThread();
93 threads.add(worker);
94 Thread thread = new Thread(worker);
95 thread.start();
96 }
97
98 threadCount = threads.size();
99 }
100
101 // 작업자 스레드 감소 (최소 1개 유지)
102 @Override
103 public void decreaseWorkers(int count) {
104 if(count >= threadCount || count <= 0){
105 return;
106 }
107
108 for(int i =0;i<count;i++){
109 WorkerThread worker = threads.getLast();
110 worker.terminate();
111 }
112
113 threadCount = threads.size();
114 }
115
116 @Override
117 public int getTaskCount() {
118 return taskQueue.size();
119 }
120
121
122 }
- 테스트 클래스 생성
1 public class ThreadPoolTest {
2 public static void main(String[] args) throws InterruptedException {
3 // 1. 스레드 풀 생성
4 CustomThreadPool pool = new CustomThreadPool();
5
6 // 2. 스레드 풀 크기 설정
7 pool.initialize(5);
8
9 // 3. 스레드 풀 시작
10 pool.start();
11
12 // 4. 작업 대기열에 작업 추가
13 for(int i = 0;i<100;i++){
14 pool.execute(new PrintJob());
15 }
16
17 }
18 }
에클립스에서 실행 시 결과 일부:
이상으로 간단한 스레드 풀 구현이 완료되었습니다. 이 예제는 스레드 풀의 기본 원리를 보여줍니다.
실제 업무에서 스레드 풀을 구현할 때 고려해야 할 사항은 더 많고 복잡합니다.
중요한 두 가지 고려 사항:
- 스레드 풀에 몇 개의 스레드를 생성해야 할까요?
스레드 수는 작업 시나리오, 하드웨어 사양, 네트워크 대역폭 등에 따라 달라집니다. 일반적으로 CPU 코어 수 + 1이 권장됩니다. 구체적인 작업 유형에 따라 조정해야 합니다.
대규모 작업: 영화 다운로드(수십분 소요) 중규모 작업: 1MB~10MB 이미지 다운로드 소규모 작업: 1MB 미만 아이콘 다운로드
소규모 작업에는 더 많은 스레드를 사용할 수 있습니다.
- 어떤 종류의 작업 대기열을 사용할지 역시 중요합니다.
이상입니다. 이후에도 스레드 병렬 처리 관련 내용을 탐구할 계획입니다.