Thread 모델 발전 과정

maro

Sep 04, 2024

Contents

Single Threaded Multi Threaded Thread Pool Work Queue

요청을 처리하는 스레드 모델의 발전 과정을 통해 스레드 풀에 대해 이해하고, 스레드 풀을 올바르게 구성하는 방법을 알아봅니다.

Single Threaded

싱글 스레드에서의 서버의 요청 처리를 코드로 표현하면 다음과 같다.


ServerSocket listener = new ServerSocket(8080);
try {
	while (true) {
		Socket socket = listener.accept();
		try {
			handleRequest(socket);
		} catch (IOException e) {
			e.printStackTrace();
		}
	}
} finally {
	listener.close();
}

final static String response =
	"HTTP/1.0 200 OK\r\n" +
	"Content-type: text/plain\r\n" +
	"\r\n" +
	"Hello World\r\n";

public static void handleRequest(Socket socket) throws IOException {
	// Read the input stream, and return "200 OK"
	try {
		BufferedReader in = new BufferedReader(
			new InputStreamReader(socket.getInputStream()));
		
		log.info(in.readLine());

		OutputStream out = socket.getOutputStream();
		out.write(response.getBytes(StandardCharsets.UTF_8));
	} finally {
		socket.close();
	}
}

8080 port를 읽는 소켓에서 요청을 받아들이고 요청을 처리하도록 handleRequest에게 전달한다. 소켓이 받아들인 요청은 동기화 작업이기 때문에 전체 동작이 완료될 때까지 다른 요청은 처리하지 못한다.

Multi Threaded

멀티 스레드 환경에서는 I/O로 인해 스레드가 Blocking되더라도 다른 스레드를 사용하여 비동기로 처리할 수 있다.


// Main loop here
ServerSocket listener = new ServerSocket(8080);
try {
	while (true) {
		Socket socket = listener.accept();
		new Thread( new HandleRequestRunnable(socket) ).start();
	}
} finally {
	listener.close();
}

public static class HandleRequestRunnable implements Runnable {
	final Socket socket;

	public HandleRequestRunnable(Socket socket) {
		this.socket = socket;
	}

	public void run() {
		try {
			handleRequest(socket);
		} catch (IOException e) {
			e.printStackTrace();
		}
	}
}

위와 같은 기술을 ‘thread per request’라고 한다. 이런 요청별 스레드 접근 방식에는 JVM과 OS 모두 리소스를 할당해야 하므로 새 스레드를 생성하고 나중에 이를 삭제하는 데 비용이 많이 들 수 있다. 또한 생성되는 스레드 수에 제한이 없으므로, 한 번에 대량의 요청이 들어올 시 리소스가 고갈될 수 있다.

Thread Pool

스레드 연결 비용이나 무제한 스레드 같은 이슈를 해결하기 위해 일반적으로 스레드 풀을 사용한다. 풀은 모든 스레드를 추적하여 필요할 때 상한까지 새 스레드를 만들고 가능한 경우 유휴 스레드를 재 사용한다.


ServerSocket listener = new ServerSocket(8080);
ExecutorService executor = Executors.newFixedThreadPool(4);
try {
	while (true) {
		Socket socket = listener.accept();
		executor.submit( new HandleRequestRunnable(socket) );
	}
} finally {
	listener.close();
}

스레드를 직접 만드는 대신, 스레드 풀에서 실행할 작업을 제출하는 ExecutorService 를 사용한다. 위 코드에서는 newFixedThreadPool을 사용하고 있지만, 그외에도 newCachedThreadPool을 사용하거나, ThreadPoolExecutor 생성자로 직접 생성할 수 있다.


   public ThreadPoolExecutor(int corePoolSize,
                              int maximumPoolSize,
                              long keepAliveTime,
                              TimeUnit unit,
                              BlockingQueue<Runnable> workQueue)


 public static ExecutorService newFixedThreadPool(int nThreads) {
      return new ThreadPoolExecutor(nThreads, nThreads,
                                    0L, TimeUnit.MILLISECONDS,
                                    new LinkedBlockingQueue<Runnable>());
  }
  
  public static ExecutorService newCachedThreadPool() {
        return new ThreadPoolExecutor(0, Integer.MAX_VALUE,
                                      60L, TimeUnit.SECONDS,
                                      new SynchronousQueue<Runnable>());
  }

newFixedThreadPool와 newCachedThreadPool는 각각 LinkedBlockingQueue와 SynchronousQueue를 인자로 전달하고 있다. SynchronousQueue는 내부 용량이 없어서 put 연산과 take 연산이 동시에 일어나야 한다. 때문에 사용 가능한 스레드가 없다면 새로운 스레드를 생성하고, 만일 최대 스레드 수에 도달했다면 RejectedExecutionHandler의 정책에 따라 처리한다.

newFixedThreadPool은 LinkedBlockingQueue의 사이즈를 지정하지 않아서 Integer.MAX_VALUE까지 대기시킨다.

정리하자면, newFixedThreadPool은 스레드 수는 고정하는 대신 요청을 Integer.MAX_VALUE까지 큐에 쌓아 놓고, newCachedThreadPool는 요청을 큐에 저장하지 않는 대신 Integer.MAX_VALUE까지 스레드 수를 늘려서 처리한다.

Work Queue


public static ExecutorService newBoundedFixedThreadPool(int nThreads, int capacity) {
	return new ThreadPoolExecutor(nThreads, nThreads,
		0L, TimeUnit.MILLISECONDS,
		new LinkedBlockingQueue<Runnable>(capacity),
		new ThreadPoolExecutor.DiscardPolicy());
}

public static void boundedThreadPoolServerSocket() throws IOException {
	ServerSocket listener = new ServerSocket(8080);
	ExecutorService executor = newBoundedFixedThreadPool(4, 16);
	try {
		while (true) {
			Socket socket = listener.accept();
			executor.submit( new HandleRequestRunnable(socket) );
		}
	} finally {
		listener.close();
	}
}