이번엔 Thread를 생성하거나 관리하는 다른 방법에 대해서 배워봅니다. 현업에서는 Thread Pool을 이용하여 작업을 처리하는 경우가 많습니다.

ㅤ

1. Thread Pool

XX문고

저는 서점 가는 것을 좋아합니다. 요즘 나오는 책 트렌드나, 개발 관련 어떤 책들이 나오는지 확인해 보는 것을 좋아합니다. 서점에 괜찮은 책이 있으면 구매하기도 합니다.

저희 동네의 XX 문고에서는 카운터가 이렇게 생겼습니다. 일단 그림을 보면 A열에 대기하는 사람이 가장 없습니다. 그래서 책을 구매하기 위해 A열에서 대기합니다. 그러나 막상 기다리다 보면 B열이 또는 C열이 더 빠르게 처리되는 경우가 있습니다.

사실 이런 방식은 어느 대기열을 기다리든 먼저 온 사람이 먼저 처리가 된다는 보장이 없습니다. 저보다 늦게 도착한 사람이 먼저 계산하고 나가는 경우가 있습니다. 운이 좋게 빨리 처리되는 대기열에 서 있는 사람이 운이 좋은 겁니다. 늦게 도착한 사람이 더 빨리 계산하고 나가면 고객의 기분이 상큼하지는 않을 겁니다.

ㅤ

공항

공항에서 수하물 위탁하는 곳은 대기열이 이렇게 되어있습니다. 우리는 Queue라는 것을 배웠기 때문에 이 방식이 어떤 방식이고 일이 어떻게 처리되는지 알 수 있지요. Queue는 먼저 들어온 것이 먼저 나오는 구조입니다. 각 작업이 양이 다르기 때문에 처리 시간은 다를지라도, 작업 처리 시작을 순서대로 하기 때문에 고객들은 기다리는 것에 대해 별다른 불만이 없을 겁니다.

Thread Pool의 Thread를 수하물 위탁 담당자라고 본다면, Queue를 공항의 대기열이라고 보면 됩니다. 각 작업 처리 완료 시간은 다를지 몰라도 먼저 온 작업을 먼저 시작한다는 것은 동일합니다.

ㅤ

Thread Pool의 구성

Thread Pool은 크게 두 가지로 구성되어 있습니다. Queue는 Submitter로 전달받은 작업을 순차적으로 저장합니다. Thread Array는 Thread Pool이 생성될 때 지정된 수만큼 Thread를 미리 생성해 놓고 보관하는 배열입니다. 그리고 생성된 Thread들은 작업을 전달받을 때까지 대기합니다.

Thread Pool의 작업을 간단히 설명하면 다음과 같습니다.

1. Task Submiiter는 작업을 Thread Pool에 전달합니다.

2. 전달 받은 작업은 Thread Pool의 Queue에 순차적으로 저장됩니다.

3. 유휴 Thread가 존재한다면 Queue에서 작업을 꺼내 처리합니다.

4. 만약에 유휴 Thread가 존재하지 않는다면, 해당 작업은 Queue에서 처리될 때 까지 대기합니다.

5. 작업을 마친 Thread는 Queue에 새로운 작업이 도착할 때 까지 대기합니다.

ㅤ

Thread Pool이 만들어진 이유

우리는 Thread가 필요하면 new 키워드를 이용하여 각 Thread 인스턴스를 생성해왔습니다. 한 번, 두 번 사용하고 반납해야 하는 그런 상황이라면 개별적으로 Thread 인스턴스를 생성하여 사용해도 상관없습니다. 그러나 Java 언어를 이용하여 많이 만들어지는 서버 애플리케이션은 수많은 Thread를 필요로 하며 초 단위로 Thread를 생성하여 사용하고 반납해야 합니다.

Java는 Thread 인스턴스의 start() 메서드가 호출되면 커널 스레드를 할당받아 사용합니다. 커널 스레드를 생성하고 반납하는 연산은 생각보다 비싼 연산입니다. 초 단위로 Thread를 몇 백 개 생성하고 사용하고 반납하고 한다면 애플리케이션은 메모리 부족 현상이 발생할 수 있으며, 반납한 Thread 인스턴스를 메모리에서 해제해야 하기 때문에 Garbage Collector도 굉장히 바빠지며 그만큼 CPU가 해야 할 일이 많아집니다.

그래서 생성과 메모리 해제 비용을 줄이기 위해 Thread를 미리 생성해 놓고 필요할 때만 가져다가 사용하고 커널 스레드를 반납하지 않고 재사용할 수 있도록 하는 Thread Pool이 만들어졌습니다.

ㅤ

Thread Pool의 생성

Java에서 Thread Pool을 생성하는 방법은 여러가지가 있으나 그 중 몇 가지만 소개해드립니다.

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

// 운용하는 Thread 갯수가 고정되어있는 Thread Pool
ExecutorService threadPool1 = Executors.newFixedThreadPool(4);

// 운용하는 Thread 갯수가 1개로 고정되어있는 Thread Pool
ExecutorService threadPool2 = Executors.newSingleThreadExecutor();

// 일정시간 주기적으로 실행해야 하는 작업이 있는 경우 사용하는 Thread Pool
ScheduledExecutorService threadPool3 = Executors.newScheduledThreadPool(4);

// 운용하는 Thread의 갯수를 정하지 않고 상황에 따라서 생성 및 해제하는 Thread Pool
ExecutorService threadPool4 = Executors.newCachedThreadPool();

ㅤ

ExecutorService

ExecutorService는 Executor 인터페이스를 상속 받는 인터페이스입니다. Thread Pool를 구현하기 위한 클래스들은 기본적으로 ExecutorService 인터페이스에 선언된 메서드들을 구현 해야 합니다. ExecutorService에는 Thread Pool에 작업을 전달하기 위한 중요한 메서드가 있습니다.

ㅤ

void execute(Runnable runnable)

execute(runnable) 메서드는 Runnable이 구현된 인스턴스를 전달할 수 있는 메서드입니다. 우리가 Thread 인스턴스를 생성하며 생성자에 runnable을 인스턴스를 전달하는 것처럼 작업을 전달하여 실행할 수 있습니다.

import java.util.concurrent.ExecutorService;

ExecutorService threadPool = …

threadPool.execute(() -> {
    int sum = 0;
    for (int i = 0; i < 100; i++) {
        sum += i;
    }
    System.out.println(sum);
});

execute(runnable) 메서드가 호출되면 Thread Pool에 대기하고 있는 Thread가 해당 작업을 처리합니다.

ㅤ

<T> Future<T> submit(Runnable runnable);

submit(runnable) 메서드는 Runnable이 구현된 인스턴스를 전달할 수 있는 메서드입니다. 다만 execute(runnable) 메서드와 다른 점은 Future라는 객체를 반환하는 메서드 입니다. Future 객체는 Thread가 작업한 내용을 동기적으로 반환값을 받을 수 있도록 도와주는 객체입니다.

말로는 어려우니 예시코드로 확인해봅시다.

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

ExecutorService threadPool = …

Future<Integer> future = threadPool.submit(() -> {
    int sum = 0;
    for (int i = 0; i < 100; i++) {
        sum += i;
    }
    return sum;
});

System.out.println(future.get());
System.out.println("Main Thread Terminated");

submit(runnable) 메서드가 호출되면 Thread Pool에 대기하고 있는 Thread가 해당 작업을 처리합니다. 만약에 처리 결과에 대한 반환 값을 동기적으로 받고 싶다면 Future 객체의 future.get() 메서드를 통해 전달받을 수 있습니다. future.get() 메서드를 호출하는 Thread는 thread.join() 메서드와 유사하게 대기하며, Thread Pool에 전달된 작업이 끝난 후 반환 값을 전달받으며 대기에서 풀려납니다.

0부터 10억까지의 수를 더하는 작업을 Thread Pool을 통해 사용한다면 이렇게 할 수 있습니다.

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

ExecutorService threadPool = …

Future<Long> future1 = threadPool.submit(() -> {
    long sum = 0;
    for (long i = 0; i < 500000000L; i++) {
        sum += i;
    }
    return sum;
});

Future<Long> future2 = threadPool.submit(() -> {
    long sum = 0;
    for (int i = 500000000L; i <= 1000000000L; i++) {
        sum += i;
    }
    return sum;
});

// thread.join()과 비슷하나 반환 값이 있는 join이라고 보면쉽다
long result1 = future1.get();
long result2 = future2.get();

System.out.println(result1 + result2);
System.out.println("Main Thread Terminated");

이렇게 병렬 연산을 통해 연산된 결과 값을 동기적으로 합쳐야 하는 경우 사용될 수 있는 메서드입니다.

ㅤ

2. ExecutorService Thread Pool의 생성

Executors.newFixedThreadPool(size)

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

// 운용하는 Thread 갯수가 고정되어있는 Thread Pool
ExecutorService threadPool = Executors.newFixedThreadPool(4);

정해진 size만큼 Thread를 생성한 Thread Pool 입니다.

ㅤ

사용용도

Thread 생성에 의한 메모리 사용을 고정하고 싶을 때 사용한다. 메모리 자원이 풍족하지 않은 환경에서 Thread Pool을 운용하고 싶을 때 사용한다.

ㅤ

장점

• 과도한 Thread 생성으로 인한 메모리 사용을 제한할 수 있다

ㅤ

단점

• 운용되는 Thread의 수와 처리하는 작업 속도 대비 더 많은 작업이 발생하면, Queue에서 작업이 대기하는 시간이 길어지므로 전체적인 작업 처리 속도가 느려질 수 있다.

ㅤ

Executors.newSingleThreadExecutor()

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

// 운용하는 Thread 갯수가 1개로 고정되어있는 Thread Pool
ExecutorService threadPool = Executors.newSingleThreadExecutor();

한 개의 스레드만 생성된 Thread Pool 입니다.

ㅤ

사용용도

한 개의 Thread를 지속적으로 재사용할 필요가 있을 때 사용한다.

전달된 작업을 순서대로 처리가 필요할 때 사용한다.

ㅤ

장점

• 장점이라기 보단 특정 스레드가 순서대로 작업을 처리하도록 하고 싶을 때 사용할 수 있습니다.

• new Thread()로 생성된 스레드로 여러 작업을 순서대로 처리하려면 별도의 코딩 작업을 통해 처리해야 하나 그런 수고 없이 쉽게 사용할 수 있습니다.

ㅤ

단점

• 단점이라기 보단 작업이 많은 경우, 사용하기에 적합하지 않습니다.

ㅤ

Executors.newCachedThreadPool()

운용하는 Thread의 갯수를 정하지 않고 상황에 따라서 생성 및 해제하는 Thread Pool

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

// 운용하는 Thread의 갯수를 정하지 않고 상황에 따라서 생성 및 해제하는 Thread Pool
ExecutorService threadPool = Executors.newCachedThreadPool();

유동적으로 Thread의 개수를 늘리고 해제하는 방식의 Thread Pool입니다. 최초에는 Thread Pool에 생성된 Thread 수는 0 개이나, 작업이 추가되거나 작업이 많을 경우 Thread를 추가적으로 생성하여 작업을 처리하며, 작업이 처리되면 어느 정도 시간 동안은 Thread Pool에 유지되고 있다가 유휴시간(기본 값 60초)이 지나면 Thread Pool에서 사용하지 않는 Thread는 해제됩니다.

ㅤ

사용용도

규칙적으로 작업이 추가되는 것이 아닌 불규칙적으로 작업이 갑자기 많아질 경우가 빈번할 때 사용한다. 예시로 평소에는 1~2개의 작업을 처리하다가 특정 시간에 작업이 100개로 늘어나는 경우, 고정된 개수의 Thread Pool이라면 모든 작업을 처리할 때까지 시간이 오래 걸릴 수 있으나, 유동적으로 Thread의 개수를 늘리고 해제하는 Thread Pool은 작업이 많아질 때 Thread를 추가 생성하여 유연하게 처리할 수 있다

ㅤ

장점

• 작업의 양이 불규칙적일 수록 유연하게 대처할 수 있다.

ㅤ

단점

• Thread 생성 갯수에 제한이 없다.

• 1000개의 작업이 갑작스럽게 늘어나면 1000개의 Thread가 생성되는 것과 마찬가지이므로 그만큼 메모리가 사용되므로 컴퓨터의 리소스가 빨리 소모될 수 있다. 한정된 메모리로 느리지만 작업 처리를 천천히 처리할 것인지, 메모리를 많이 사용하고 빠르게 작업 처리할 것인지에 따라 사용 여부가 달라진다.

ㅤ

ScheduledExecutorService

ScheduledExecutorService는 ExecutorService 인터페이스를 상속받는 인터페이스입니다. 특정 시간 이후에 작업이 처리되길 원하거나 또는 특정 시간마다 반복된 작업이 필요할 경우 사용될 수 있습니다. 작업의 스케줄을 정할 수 있습니다.

ㅤ

ScheduledFuture<?> schedule(Runnable command, long delay, TimeUnit unit)

schedule 메서드는 특정 시간(delay) 동안 대기한 후에 작업을 처리할 수 있도록 도와주는 메서드입니다. 다만 스케줄링과 관련된 메서드들의 특징은 Thread 스케줄에 따라 실행되므로 10초 후에 실행하라고 지정하더라도 정확한 10초 후가 아닌 10.2초 10.4초에 실행될 수 있습니다.

세 번째 매개변수인 TimeUnit은 long delay 매개변수의 기준을 정하는 것으로 TimeUnit에 따라 delay 매개변수를 지정해야 합니다.

ㅤ

enum TimeUnit

import java.util.concurrent.ScheduledExecutorService;

ScheduledExecutorService threadPool = …

threadPool.schedule(() -> {
    System.out.println("10초 후 실행됨");
}, 10, TimeUnit.SECONDS);

threadPool.schedule(() -> {
    System.out.println("1분 후 실행됨");
}, 1, TimeUnit.MINUTES);

threadPool.schedule(() -> {
    System.out.println("1.2초 후 실행됨");
}, 1200, TimeUnit.MILLISECONDS);

해당 메서드는 ScheduledFuture<?>를 반환하므로 ExecutorService의 submit(runnable) 메서드와 동일하게 동기화된 결과값을 받을 수 있습니다.

import java.util.concurrent.ScheduledExecutorService;

ScheduledExecutorService threadPool = …

ScheduledFuture<Integer> future = threadPool.schedule(() -> {
    System.out.println("1초 후 실행됨");
    int sum = 0;
    for (int i = 0; i < 100; i++) {
        sum += i;
    }
    return sum;
}, 1, TimeUnit.SECONDS);

int result = future.get();

해당 메서드가 실행되면 1초후에 작업을 시작하며, 작업을 처리 할 때 까지 future.get() 메서드를 호출한 Thread는 대기하게 됩니다. 작업이 완료되면 해당 Thread는 대기에서 풀려나며 결과값을 반환 받습니다.

ㅤ

ScheduledFuture<?> scheduleWithFixedDelay(Runnable command, long initialDelay, long delay, TimeUnit unit)

scheduleWithFixedDelay 메서드는 지정된 시간(initialDelay)동안 최초에 한 번 대기한 후에 작업을 처리 후 작업이 처리한 완료 시간부터 지정된 시간(delay) 마다 반복됩니다.

import java.util.concurrent.ScheduledExecutorService;

ScheduledExecutorService threadPool = …

ScheduledFuture<Integer> future = threadPool.scheduleWithFixedDelay(() -> {
    // 5분 정도 걸리는 작업
}, 2, 4, TimeUnit.MINUTES);

int result = future.get();

메서드의 호출시간이 00:00라고 가정합니다.

1. 약 2분 뒤에 작업을 진행합니다. (00:02)

2. 5분정도의 작업이 완료됩니다. (00:07)

3. 4분정도 기다린 후 다시 작업을 진행합니다. (00:11)

4. 5분정도의 작업이 완료됩니다. (00:16)

5. 4분정도 기다린 후 다시 작업을 진행합니다. (00:20)

해당 메서드는 작업의 완료 시점을 기준으로 deley 만큼 기다리고 작업을 반복하는 메서드입니다. 즉 작업이 종료된 시점 부터 대기 시간을 계산합니다.

ㅤ

ScheduledFuture<?> scheduleAtFixedRate(Runnable command, long initialDelay, long period, TimeUnit unit)

scheduleWithFixedDelay와 유사하게 반복적으로 작업하는 것은 동일하지만, 작업 완료 시간과 상관없이

고정된 시간(period)마다 작업을 실행합니다.

import java.util.concurrent.ScheduledExecutorService;

ScheduledExecutorService threadPool = …

ScheduledFuture<Integer> future = threadPool.scheduleAtFixedRate(() -> {
    // 3분 정도 걸리는 작업
}, 2, 4, TimeUnit.MINUTES);

int result = future.get();

메서드가 호출시간이 00:00라고 가정합니다.

1. 약 2분 뒤에 작업을 진행합니다. (00:02)

2. 3분정도의 작업이 완료됩니다. (00:05)

3. 최초 작업 시작 시간을 기준으로 4분 뒤에 작업을 진행합니다. (00:06)

4. 3분정도의 작업이 완료됩니다. (00:09)

5. 두 번째 작업 시작 시간 기준 4분 뒤에 작업을 진행합니다. (00:10)

해당 메서드는 작업의 완료 시점 부터가 아닌 최초 작업이 시작된 시간을 기준으로 period 만큼 대기 시간마다 작업을 반복 실행합니다.

ㅤ

3. ScheduledExecutorService Thread Pool의 생성

Executors.newScheduledThreadPool(size);

import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;

// 일정시간 주기적으로 실행해야 하는 작업이 있는 경우 사용하는 Thread Pool
ScheduledExecutorService threadPool = Executors.newScheduledThreadPool(4);

ㅤ

사용용도

작업을 처리하기 위한 지연시간이 필요하거나 반복적인 작업을 처리할 때 사용한다.

ㅤ

장점

• 지연시간이 필요하거나 반복적인 작업을 처리하기 위해 추가적인 코드를 직접 작성하지 않아도 된다.

ㅤ

Thread Pool에 생성된 모든 Thread가 바쁘다면 어떻게 되나?

지정된 schedule에 따라 실행하는 ScheduledExecutorService의 메서드들은 반복적인 작업 실행을 위한 대기 시간을 정확하게 보장하지는 않습니다.

만약에 Thread Pool에 운용되는 모든 Thread가 바쁘다면 유휴 Thread가 존재할 때까지 작업은 미뤄지게 됩니다. 그러므로 반복적인 작업을 안정적으로 운용하고 싶다면 각 스케줄 되는 작업마다 Thread Pool을 생성하여 운용하는 것을 추천드립니다.

import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;


// A 작업만 처리하는 Thread Pool
ScheduledExecutorService scheduledThreadPoolA = Executors.newFixedThreadPool(2);

scheduledThreadPoolA.scheduleAtFixedRate(() -> {
    // 5분 정도 걸리는 A 작업
}, 2, 4, TimeUnit.MINUTES);


// B 작업만 처리하는 Thread Pool
ScheduledExecutorService scheduledThreadPoolB = Executors.newFixedThreadPool(2);

scheduledThreadPoolB.scheduleAtFixedRate(() -> {
    // 3분 정도 걸리는 B 작업
}, 2, 4, TimeUnit.MINUTES);