[Concurrency] - ThreadPoolExecutor 를 이용한 SimpleServer 만들기

몇년전쯤에 한 선배 왈..
"너는 기본이 부족한거 같다."

처음 그 소리를 들었을때 느껴지는 그 쓸쓸함이란..

요 몇일 java performance fundamental, java concurrency in practice 를 읽으면서 그 선배의 말이 뇌리를 울리는건 무엇일까? 반성하고 또 반성해야겠다.

항상 잊어버리는 누군가를 위해서.. 작업물을 올린다.

1. SimpleServer.java
- 간단한 서버소켓 샘플에 스레드풀만 추가한 단순한 모습.

package study.threadpool.server;

import java.io.IOException;


import java.net.ServerSocket;
import java.net.Socket;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;

public class SimpleServer {

 // 스레드 풀은 실행할 작업이 없어도 corePoolSize 를 유지한다.
 // 최초에 corePoolSize 만큼 스레드를 생성하지 않고 작업이 실행 될 때 corePoolSize 만큼 차례대로 생성함.
 // prestartAllCoreThreads() 를 사용하면 corePoolSize 만큼 스레드를 미리 생성함.
 final int corePoolSize   = 10;
 // xaximumPoolsize 는 동시에 얼마나 많은 개수의 스레드가 동작할 수 있는지를 제한하는 최대값.
 final int xaximumPoolsize  = 10;
 final int blocdingQueueSize = 100;
 final int port    = 10001;
 // keepAliveTime 는 스레드 유지 시간으로 스레드가 keepAliveTime 이상 대기하고 있으면
 // 해당 스레드는 제거 될 수 있음. 풀의 스레드 개수가 corePoolSize 를 넘어서면 제거될 수 있음.
 final long keepAliveTime = 30L;

 final ThreadPoolExecutor pool;
 final ServerSocket servrerSocket;
 
 final BlockingQueue<Runnable> queue;
 final PoolThreadFactory threadFactory;
 
 public SimpleServer() throws IOException {
  queue    = new ArrayBlockingQueue<Runnable>(blocdingQueueSize);
  threadFactory  = new PoolThreadFactory("SERVER_POOL");
  servrerSocket  = new ServerSocket(port);
  pool    = new ThreadPoolExecutor(corePoolSize,
             xaximumPoolsize,
             keepAliveTime,
             TimeUnit.SECONDS,
             queue,
             threadFactory,
             new ThreadPoolExecutor.CallerRunsPolicy());
 
 }
 
 public void serverStart() {
  try {

   while(true) {
    Socket clientSocket = servrerSocket.accept(); 
    pool.execute(new Work(clientSocket));
   } 
  } catch (IOException e) {
   // TODO Auto-generated catch block
   e.printStackTrace();
  }
 }
 
 public static void main(String[] args) {
  try {
   new SimpleServer().serverStart();
  } catch (IOException e) {
   // TODO Auto-generated catch block
   e.printStackTrace();
  }
 }
}

2. PoolThreadFactory.java
- ThreadFactory 를 상속받아 직접 제작한 스레드를 실행시킨다.
- 스레드 생성과 종료에 따른 로그를 기록하는 스레드를 사용하기 위해서(책 참조)

package study.threadpool.server;

import java.util.concurrent.ThreadFactory;

public class PoolThreadFactory implements ThreadFactory {

 private final String name;
 
 public PoolThreadFactory(String name) {
  this.name = name;
 }
 
 @Override
 public Thread newThread(final Runnable r) {
  final PoolAppThread poolAppThread = new PoolAppThread(r, name);
  PoolAppThread.setDebug(true);   
  return poolAppThread;
 }
}

3. PoolAppThread.java
- 스레드 생성, 종료에 따른 로그를 남기기 위해 제작된 단순한 예제 쓰레드

package study.threadpool.server;

import java.util.concurrent.atomic.AtomicInteger;

public class PoolAppThread extends Thread {

 public  static final String DEFAULT_POOL_NAME  = "PoolAppThread";
 private static volatile boolean debugLifecycle  = false;
 private static final AtomicInteger created   = new AtomicInteger();
 private static final AtomicInteger alive   = new AtomicInteger();
 
 public PoolAppThread(Runnable r) {
  this(r, DEFAULT_POOL_NAME); 
 }
 
 public PoolAppThread(Runnable runnable, String name) {
        super(runnable, name + "-" + created.incrementAndGet());
        setUncaughtExceptionHandler(new Thread.UncaughtExceptionHandler() {
            public void uncaughtException(Thread t,
                                          Throwable e) {
            
             System.out.println("UNCAUGHT in thread " + t.getName() + ":" + e);
            }
        });
    }
 
 public void run() {
        // Copy debug flag to ensure consistent value throughout.
        boolean debug = debugLifecycle;
        if (debug) {
         System.out.println("Created : " + getName());        
        }
        try {
            alive.incrementAndGet();
            super.run();
        } finally {
            alive.decrementAndGet();
            if (debug) {
             System.out.println("Exiting : " + getName());
            }
        }
    }
 
 public static int getThreadsCreated() {
        return created.get();
    }

    public static int getThreadsAlive() {
        return alive.get();
    }

    public static boolean getDebug() {
        return debugLifecycle;
    }

    public static void setDebug(boolean b) {
        debugLifecycle = b;
    }
}

4. Work.java
- 받은 데이터 보여주고 헛소리 전달만 하는 무식한넘.

package study.threadpool.server;

import java.io.IOException;
import java.io.InputStreamReader;
import java.io.PrintWriter;
import java.net.Socket;


public class Work implements Runnable {

 
 Socket socket;
 
 public Work(Socket socket) {
  this.socket = socket;
 }
 
 @Override
 public void run() {
  // TODO Auto-generated method stub
  try {
   PrintWriter out = new PrintWriter(socket.getOutputStream(), true);
   InputStreamReader in = new InputStreamReader(socket.getInputStream());
  
   char[] rBuf = new char[1000];
   int readMsg = in.read(rBuf);
  
   String msg = String.valueOf(rBuf).trim();
  
   System.out.println("read : [" + msg + "]");
  
   try {
    Thread.sleep(50);
   } catch (InterruptedException e) {
    // TODO Auto-generated catch block
    e.printStackTrace();
   }
  
   out.println("이거나 받으삼\n");
  
  } catch (IOException e) {
   // TODO Auto-generated catch block
   System.err.println("Oh, dear me! " + e);
  }
 }
}

성능을 높이기 위해서 여러가지 방법들이 있는거 같지만. 그때는 직접 하나하나 따려가면서 작업하는게 낳을듯.. 지금은 단순히 샘플만..