[병렬프로그래밍] 6. 작업실행 — 거꾸로 바라본 세상

작업실행

아래 초기 웹 서버 코드는 작업 분할 방식을 비교하는 구조 예시다. handleRequest 구현, 소켓 정리, 포트 권한과 예외 처리를 생략했으므로 그대로 컴파일·운영하는 서버가 아니다.

대부분 병렬 어플리케이션은 ‘작업task’을 실행하는 구조가 효율적으로 구성되어있다.

작업: 추상적이면서 명확하게 구분된 업무의 단위

작업은 원론적으로 완전히 독립적인 동작으로 상태,결과, 부수효과 등에 영향을 받지 않아야함

작업을 실행하는 가장 간단한 방법

단일 스레드에서 작업 목록을 순차적으로 실행하는 방법

6.1.1 작업을 순차적으로 실행

하나의 스레드가 모든 요청을 순차적으로 처리하는 경우

class SingleThreadWebServer {
	public static void main(String[] args) {
		ServerSocket socket = new ServerSocket(80);
		while (true) {
			Socket conn = socket.accept();
			handleRequest(conn);
		}
	}
}

* 실제 적용시 성능이 엄청 떨어짐.

단일 스레드에서 IO작업을 하는동안 CPU가 대기하고 있어야하는 경우가 있고 서버 하드웨어 자원을 제대로 활용하지 못하는 문제가 있다.

6.1.2 작업마다 스레드를 직접 생성

요청이 들어올 때마다 새로운 스레드를 하나씩 만들어 생성하게 되는 경우

class ThreadPerTaskWebServer {
	public static void main(String[] args) {
		ServerSocket socket = new ServerSocket(80);
		while (true) {
			final Socket conn = socket.accept();
			Runnable task = new Runnable() {

				public void run() {
					handleRequest(conn);
				}
			};
			new Thread(task).start();
		}
	}
}

클라이언트의 요청 내용 처리를 메인스레드에서 처리하는 것이 아니라 새로운 스레드를 생성하여 처리

작업을 처리하는 기능이 메인 스레드에서 떨어져 나옴(클라이언트 접속을 기다리는 부분이 빨라짐)
동시에 여러 작업을 병렬로 처리할 수 있기 때문에 두 개 이상의 요청을 받아 동시에 처리가능
실제 작업을 처리하는 스레드의 안전성을 확보해야 한다.
클라이언트가 접속 요청을 전송하는 속도에 비해 요청을 처리해 응답을 넘겨주는 속도가 빨라야한다.

6.1.3 스레드를 많이 생성할 때 문제점

스레드 라이프 사이클 문제

생성과 제거 작업에도 자원이 소모됨
클라이언트의 요청을 처리할 때 딜레이가 생기고 JVM과 운영체제도 기초적입 작업을 수행

자원 낭비

스레드는 시스템자원, 특히 메모리를 소모함
프로세서보다 많은 스레드를 만들어 동작중이라면 대부분의 스레드는 idle 상태에 머무름
많은 스레드는 많은 메모리를 필요로함
GC에 가해지는 부하가 늘어남

안전성 문제

모든 시스템에는 생성할 수 있는 스레드의 개수가 제한되어 있음
Thread 클래스에 필요한 스택의 크기의 제한된 양을 모두 사용하면 OutOfMemoryError 발생하고 해결방법은 별로 없다.
OutOfMemoryError이 발생하지 않도록 미연에 방지해야함

6,2 Executor 프레임워크

순차 스레드, 각각의 작업을 스레드에서 담아 처리하는 방법에는 문제점이 있다.

순차적인 방법 : 응답속도와 성능이 크게 떨어짐

작업별로 스레드를 생성하는 방법 : 관리측면에서 허점이 있음Thread Pool은 스레드를 관리하는 측면에서 메모리를 모두 소모해 버리지 않도록 통제력을 가지고 있다.

public interface Executor {
	void execute(Runnable command);
}

Executor은 비동기적 작업 실행 프레임워크의 근간을 이루는 인터페이스
Task Submission(작업등록) 과 Task Execution(작업실행)을 분리
각 작업의 라이프 사이클을 관리와 모니터링하는 기능 제공
각 작업은 Runnable 형태로 정의
일반적으로 Producer-Consumer 패턴에 기반

6.2.1 ex) Executor를 사용한 웹 서버

class TaskExecutionWebServer {
	private static final int NTHREADS = 100;
	private static final Executor exec = Executors.newFixedThreadPool(NTHREADS);

	public static void main(String[] args) throws IOException {
		ServerSocket socket = new ServerSocket(80);
		while (true) {
			final Socket conn = socket.accept();
			Runnable task = new Runnable() {
				public void run() {
					handleRequest(conn);
				}
			};
			exec.execute(task);
		}
	}
}

6.2.2 실행정책

작업을 등록하는 부분과 실행하는 부분을 서로 분리시켜두면 실행정책(execution policy)를 쉽게 변경할 수 있다. 실행 정책은 ‘무엇을, 어디에서, 언제, 어떻게’ 실행하는지 지정할 수 있다.

작업을 어느 스레드에서 실행할 것인가?
작업을 어떤 순서로 실행할 것인가?(FIFO, LIFO, 기타 다양한 우선순위 정책)
동시에 몇 개의 작업을 병렬로 실행할 것인가?
최대 몇 개까지의 작업이 큐에서 실행을 대기할 수 있게 할 것인가?
시스템 부하가 많이 걸려서 작업을 거절해야할 경우, 어떤 작업을 희생양으로 삼야아 할 것이며, 작업을 요청한 프로그램에 어떻게 알려야 할 것인가.
작업을 실행하기 직전이나 실행한 직후에 어떤 동작이 있어야 하는가?

* 프로그램 어디에서든 간에

new Thread(runnable).start(); 와 같은 코드가 남아 있다면 이런 부분에 유연한 실행 정책을 적용할 준비를 해야할 것이며, 나중을 위해서 Executor를 사용해 구현하는 방안을 심각하게 고려해봐야 한다.

6.2.3 스레드 풀

Thread Fool : 스레드 작업을 처리할 수 있는 작업 공간

실행작업

-> 먼저 큐에서 실행할 다음작업을 가져옴

-> 작업을 실행

-> 실행할 다음 작업이 나타날 때까지 대기하는 일을 반복

장점

매번 스레드를 생성하는 대신 이전에 사용했던 스래드를 재사용하므로 계속해서 스레드를 생성할 필요 없음
여러 개의 요청을 처리하는 데 필요한 시스템 자원이 줄어드는 효과가 있다.
크기를 적절히 조절하면 하드웨어 프로세서가 쉬지 않고 동작할 수 있다
스레드 수를 제한할 수 있지만, 작업 대기 큐가 무한히 늘어나면 여전히 메모리가 부족해질 수 있다. 스레드와 큐를 함께 제한해야 한다.
작업별로 스레드를 생성하는 전략(Thread-per-task)에서 풀을 기반으로 하는 전략(pool-based)으로 변경되어 안전성 측면에서 장점을 가지고 있음

자바 스레드풀 메소드

newFixedThreadPool

처리할 작업이 등록되면 실제 작업할 스레드를 하나씩 생성한다.
생성할 수 있는 스레드의 최대 개수는 제한
제한된 개수까지 스레드를 생성하고 나면 더 이상 생성하지 않고 스레드 수를 유지
스레드가 작업중에 예외가 발생해서 스레드가 종료되면 하나씩 더 생성
newFixedThreadPool의 작업 큐는 사실상 무제한이므로 요청이 처리 속도보다 오래 빠르게 들어오면 대기 작업이 쌓인다. 부하 상한이 중요한 서버에서는 큐 크기와 거절 정책을 명시한 ThreadPoolExecutor를 검토한다.

newCachedThreadPool

현재 Pool에 가지고 있는 스레드의 수가 처리할 작업의 수보다 많아서 쉬는 스레드가 많이 발 생하면 쉬는 스레드를 종료시켜 유연하게 대응
처리할 작업의 수가 많아지면 필요한 만큼 스레드를 생성
스레드의 수에는 제한을 두지 않음

newSingleThreadExecutor

단일 스레드로 동작하는 Executor로 작업을 처리하는 스레드가 단 하나
작업중 Exception이 발생하여 비정상적으로 종료되면 새로운 스레드를 생성하여 나머지 작업 처리
등록된 작업은 설정된 큐에서 지정하는 순서에 따라 순차적으로 처리

newScheduledThreadPool

일정시간 이후에 실행하거나 주기적으로 작업을 실행
스레드의 수가 고정되어 있는 Executor.Timer클래스의 기능과 유사

6.2.4 Executor 동작 주기

Executor를 구현하는 클래스는 대부분 작업을 처리하기 위한 스레드를 생성하도록 되어있음
JVM은 모든 스레드가 종료되기 전에 종료하지 않고 대기하기 때문에 Executor를 제대로 종료하지 않으면 JVM자체가 종료되지 않고 대기하기도 한다.
Executor은 비동기적으로 실행되고 작업의 상태를 특정 시점에 정확하게 파악하기는 어려움

그래서ExecutorService 인터페이스에는 동작 주기를 관리하는 여러 메소드가 있다.

public interface ExecutorService extends Executor {
	void shutdown();
        List<Runnable> shutdownNow();
        boolean isShutdown();
        boolean isTerminated();
        boolean awaitTermination( long timeout, TimeUnit unit)
                  throws InterruptedException;
        //... 작업을 등록할 수 있는 몇가지 추가 메소드
}

내부적으로 ExecutorService가 갖고 있는 동작주기

실행중(Running) : ExecutorService를 처음 생성했을 때
종료 중(shutting down) : 어느시점에 shutdown 메소드를 실행하면 안전한 종료 절차를 진행하기 위해 종료중 상태로 전환(새로운 작업을 등록받지 않고, 이전에 등록되어 있는 작업까지는 모두 끝마칠 수 있다) shutdownNow 메소드는 강제종료를 수행

shutdownNow()는 실행 중 작업에 인터럽트를 요청하고 대기 작업을 반환하지만, 모든 작업을 즉시 강제 종료시키지는 못한다. 작업이 인터럽트에 반응하고 외부 I/O에 적절한 타임아웃을 둘 때 종료가 수렴한다.

종료(terminated) : 스레드 종료

ThreadPoolExecutor : 이미 종료된 절차가 시작되거나 종료된 이후에 새로운 작업을 등록하려하면 실행 거절 핸들러 (reject execution handler)를 통해 오류를 처리

클라이언트 측에서 특정한 형태의 HTTP 요청을 전송하는 방법

public class LifecycleWebServer {
	private final ExecutorService exec = Executors.newFixedThreadPool(1);

	public void start() throws IOException {
		ServerSocket socket = new ServerSocket(80);
		while(!exec.isShutdown()) {
			try {
				final Socket conn = socket.accept();
				exec.execute(new Runnable() {

					public void run() {
						handleRequest(conn);
					}
				});
			} catch (Exception e) {
				if (!exec.isShutdown()) {

				}
			}
		}
	}
	public void stop() {
		exec.shutdown();
	}
	// 이 코드는 구조를 보이는 의사 코드다. accept()에서 기다리는
	// ServerSocket도 닫지 않으면 stop()만으로 루프가 끝나지 않는다.
}

6.2.5 지연작업, 주기적 작업

Timer 클래스는 특정시간 이후에 원하는 작업을 실행하는 지연작업이나 주기적인 작업을 실행할 수 있다. but, Timer 그 자체로는 단점이 있기 때문에 ScheduledThreadPoolExecutor를 사용하는 방법을 추천 (이유 : Timer 클래스는 상대시각과 절대시각도 지원한다. 절대시각을 사용하면 시스템 하드웨어의 시각을 변경시키면 Timer에 스케쥴된 작업도 함께변경됨, 그러나 ScheduledThreadPoolExecutor는 상대시각만 지원)
만약 스케줄 방법을 지원하는 스케줄링 서비스를 구현해야 할 필요가 있다면 BlockingQueue를 구현하면서 ScheduledThreadPoolExecutor와 비슷한 기능을 제공하는 DelayQueue 클래스를 사용하는 것을 추천

6.3 병렬로 처리할 만한 작업

6.3.2 결과가 나올때까지 대기 : Callable과 Future

Runnable은 run메소드의 실행이 끝나면 결과 값을 리턴해 줄 수 없고, 예외가 발생해도 throws 구문으로 표현할 수 없다.
Callable은 call메소드를 실행하고 결과값을 돌려받을 수 있고 Exception도 발생시키도록 되어 있다.
Executor에서 실행한 작업은 생성(created), 등록(submitted), 실행(started), 종료(completed) 네 가지 상태를 통과한다.
Future는 특정 작업이 정상적으로 완료되었는지, 취소되었는지 등에 대한 정보를 확인할 수 있다.

6.3.5 CompletionService : Executor와 BlockingQueue의 연합

CompletionService는 Executor의 기능과 BlockingQueue의 기능을 하나로 모은 인터페이스이다. 필요한 Callable 작업을 등로해 실행시킬 수 있고, take나 poll과 같은 큐 메소드를 사용해 작업이 완료되는 순간 완료된 작업의 Future 인스턴스를 받아올 수 있다.

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

public class CompletionExample {
    public static void main(String[] args) throws Exception {
        ExecutorService pool = Executors.newFixedThreadPool(2);
        try {
            CompletionService<Integer> completed = new ExecutorCompletionService<>(pool);
            for (int number = 1; number <= 3; number++) {
                final int value = number;
                completed.submit(() -> value * value);
            }
            int total = 0;
            for (int i = 0; i < 3; i++) {
                total += completed.take().get();
            }
            System.out.println(total); // 1 + 4 + 9 = 14
        } finally {
            pool.shutdownNow();
        }
    }
}

take()는 가장 먼저 완료된 작업의 Future를 돌려준다. 위 계산의 합은 완료 순서와 무관하게 14지만, 순서가 중요한 화면 표시라면 결과와 원본 위치를 함께 보관해야 한다. take()와 get()은 기다릴 수 있으므로 UI 스레드에서 직접 호출하지 않는다. finally에서 풀을 종료하며, 실서비스에서는 작업이 인터럽트에 반응하는지 확인하고 종료 유예 시간을 정한다.

6.3.7 작업실행 제한 시간