[Java] 서버의 동시 요청 처리(스레드풀 이용)

이 게시글은 이것이 자바다(저자 : 신용권, 임경균)의 책과 동영상 강의를 참고하여 개인적으로 정리하는 글임을 알립니다. 

---

서버의 다중 작업

출처 : 이것이 자바다 유튜브 동영상 강의

위와 같이 TCP와 UDP 서버에서는 accept()와 receive() 를 제외한 요청 처리 코드를 별도의 스레드에서 작업을 해야 서버에 들어오는 클라이언트의 무수한 요청을 동시에 처리할 수 있다.

 

스레드를 처리할 때 주의할 점은 클라이언트의 폭증으로 인한 서버의 과도한 스레드 생성을 방지해야 한다는 것이다.

자바에서는 이러한 과도한 스레드 생성을 방지하기 위해 스레드풀을 제공한다.

출처 : 이것이 자바다 유튜브 동영상 강의

스레드풀은 작업 처리 스레드 수를 제한해서 사용하기 때문에 갑작스런 클라이언트 폭증이 발생해도 크게 문제가 되지 않는다.

다만 작업 큐의 대기 작업이 증가되어 클라이언트에서 응답을 늦게 받을 수 있다.

스레드풀
스레드풀의 개념을 비유하자면 은행에서 은행 직원이 스레드이고, 은행을 방문한 고객이 작업이다.
즉, 은행원들이 여러명 앉아있으면 고객은 번호표(작업 큐)를 뽑고 대기하여 차례대로 처리된다.
손님이 많다고 무작정 은행 직원의 수를 늘리는 것이 아니라, 은행 직원의 수는 한정적으로 두고 작업을 처리하는 것이다.
물론, 손님이 많으면 늦게 온 손님은 오래 기다려야 할 수 있다.

 

기존에 작업 처리 내용을 스레드풀에 넣기만 하면 쉽게 구현할 수 있다.

 

아래에 나오는 예제에서 메인 스레드는 Q나 q를 입력받으면 모든 스레드를 종료시켜 프로세스를 종료시키는 작업을 처리하고, 작업 스레드는 클라이언트의 연결이 들어오면 스레드풀의 작업큐에 작업을 넣는 역할을 한다.

 

스레드풀의 스레드는 작업큐에 대기하고 있는 작업들을 순차적으로 처리한다.

---

 

TCP 서버 동시 요청 처리

2023.08.12 - [Java] - [Java] TCP/IP 네트워킹

위 글에서 다뤘던 예제를 스레드풀을 이용해서 클라이언트의 요청을 동시처리

 

스레드풀에 전달되는 작업 객체는 리턴값이 없으면 Runable 익명 구현 객체이고, 리턴 값이 있으면 Callable 익명 구현 객체이다.

Runable과 Callable 둘 다 추상 메소드가 하나만 정의(함수형 인터페이스)되어 있으므로 람다식을 사용할 수 있다.

import java.io.DataInputStream;
import java.io.DataOutputStream;
import java.io.IOException;
import java.net.InetSocketAddress;
import java.net.ServerSocket;
import java.net.Socket;
import java.util.Scanner;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

public class EchoServer {
	private static ServerSocket serverSocket = null;
	private static ExecutorService executorService = Executors.newFixedThreadPool(10); //스레드풀의 스레드 수를 10개로 제한
	
	public static void main(String[] args) {
		System.out.println("--------------------------------------------------------------------");
		System.out.println("서버를 종료하려면 q를 입력하고 Enter 키를 입력하세요.");
		System.out.println("--------------------------------------------------------------------");
		
		//TCP 서버 시작
		startServer();
		
		//키보드 입력
		Scanner scanner = new Scanner(System.in);
		while(true) {
			String key = scanner.nextLine();
			if(key.toLowerCase().equals("q")) {
				break;
			}
		}
		scanner.close();
		
		//TCP 서버 종료
		stopServer();		
	}	

	public static void startServer() {
		//작업 스레드 정의
		Thread thread = new Thread() {
			@Override
			public void run() {
				try {
					//ServerSocket 생성 및 Port 바인딩
					serverSocket = new ServerSocket(50001);	
					System.out.println( "[서버] 시작됨\n");
					
					//연결 수락 및 데이터 통신
					while(true) {
						//연결 수락
						Socket socket = serverSocket.accept();
						
						executorService.execute(() -> { //작업 큐의 작업을 익명 구현 객체로 전달
							try {
								//연결된 클라이언트 정보 얻기
								InetSocketAddress isa = (InetSocketAddress) socket.getRemoteSocketAddress();
								System.out.println("[서버] " + isa.getHostName() + "의 연결 요청을 수락함");
									
								//데이터 받기
								DataInputStream dis = new DataInputStream(socket.getInputStream());
								String message = dis.readUTF();
								
								//데이터 보내기
								DataOutputStream dos = new DataOutputStream(socket.getOutputStream());
								dos.writeUTF(message);
								dos.flush();
								System.out.println( "[서버] 받은 데이터를 다시 보냄: " + message);								
								
								//연결 끊기
								socket.close();
								System.out.println("[서버] " + isa.getHostName() + "의 연결을 끊음\n");
							} catch(IOException e) {
							}
						});
					}
				} catch(IOException e) {
					System.out.println("[서버] " + e.getMessage());
				}
			}
		};
		//스레드 시작
		thread.start();
	}
		
	public static void stopServer() {
		try {
			//ServerSocket을 닫고 Port 언바인딩
			serverSocket.close();
			executorService.shutdownNow();
			System.out.println( "[서버] 종료됨 ");
		} catch (IOException e1) {}
	}
}

 

 

---

 

UDP 서버 동시 요청 처리

2023.08.13 - [Java] - [Java] UDP 네트워킹

위 글에서 다뤘던 예제를 수정한 것으로, 스레드 풀을 이용해서 클라이언트의 요청을 동시 처리

 

스레드풀에 전달되는 작업 객체는 리턴값이 없으면 Runable 익명 구현 객체이고, 리턴 값이 있으면 Callable 익명 구현 객체이다.

Runable과 Callable 둘 다 추상 메소드가 하나만 정의(함수형 인터페이스)되어 있으므로 람다식을 사용할 수 있다.

 

import java.net.DatagramPacket;
import java.net.DatagramSocket;
import java.net.SocketAddress;
import java.util.Scanner;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

public class NewsServer {
	private static DatagramSocket datagramSocket = null;
	private static ExecutorService executorService = Executors.newFixedThreadPool(10); //스레드풀의 스레드 수를 10개로 제한
	
	public static void main(String[] args) throws Exception {
		System.out.println("--------------------------------------------------------------------");
		System.out.println("서버를 종료하려면 q 를 입력하고 Enter 키를 입력하세요.");
		System.out.println("--------------------------------------------------------------------");		
		
		//UDP 서버 시작
		startServer();
		
		//키보드 입력
		Scanner scanner = new Scanner(System.in);
		while(true) {
			String key = scanner.nextLine();
			if(key.toLowerCase().equals("q")) {
				break;
			}
		}
		scanner.close();
		
        //TCP 서버 종료
        stopServer();		
	}	
		
	public static void startServer() {
		//작업 스레드 정의
		Thread thread = new Thread() {
			@Override
			public void run() {
				try {
					//DatagramSocket 생성 및 Port 바인딩
					datagramSocket = new DatagramSocket(50001);
					System.out.println( "[서버] 시작됨");
					
					while(true) {
						//클라이언트가 구독하고 싶은 뉴스 종류 얻기
						DatagramPacket receivePacket = new DatagramPacket(new byte[1024], 1024);
						datagramSocket.receive(receivePacket);
						
						executorService.execute(() -> { //작업 큐의 작업을 익명 구현 객체로 전달
							try {
								String newsKind = new String(receivePacket.getData(), 0, receivePacket.getLength(), "UTF-8");
							
								//클라이언트의 IP와 Port 얻기
								SocketAddress socketAddress = receivePacket.getSocketAddress();
								
								//10개의 뉴스를 클라이언트로 전송
								for(int i=1; i<=10; i++) {
									String data = newsKind + ": 뉴스" + i;
									byte[] bytes = data.getBytes("UTF-8");
									DatagramPacket sendPacket = new DatagramPacket(bytes, 0, bytes.length, socketAddress);
									datagramSocket.send(sendPacket);
								}
							} catch (Exception e) {
							}
						});
					}
				} catch (Exception e) {
					System.out.println("[서버] " + e.getMessage());
				}
			}			
		};
		//스레드 시작
		thread.start();
	}
		
	public static void stopServer() {
		//DatagramSocket을 닫고 Port 언바인딩
		datagramSocket.close();
		executorService.shutdownNow();
		System.out.println( "[서버] 종료됨 ");
	}
}