Java自定义多线程服务器

// multi.MultiServer.java
package multi;

import java.io.IOException;
import java.net.InetSocketAddress;
import java.net.ServerSocket;
import java.net.Socket;

public class MultiServer {
	private int port = 8000;
	private int backlog = 42;
	private ServerSocket server_socket;
	private ThreadPool threadPool;
	private final int POOL_SIZE = 7;
	
	public MultiServer()throws IOException{
		server_socket = new ServerSocket();
		server_socket.setReuseAddress(true);
		server_socket.bind(new InetSocketAddress(port),backlog);
		int system_cpu_total = Runtime.getRuntime().availableProcessors();
		threadPool = new ThreadPool(POOL_SIZE * system_cpu_total);
		System.out.println("--- server started! ---");
	}
	public void service(){
		while(true){
			try{
				Socket socket = server_socket.accept();
				threadPool.execute(new WorkHandler(socket));
			} catch(Exception e){
				e.printStackTrace();
			}
		}
	}
	public static void main(String[] args)throws Exception {
	    MultiServer server = new MultiServer();
	    server.service();
    }
}

// multi.ThreadPool.java
package multi;

import java.util.LinkedList;

public class ThreadPool extends ThreadGroup{
    private static int threadPoolId;
    private boolean isClosed = false;
    private LinkedList<Runnable> workQueue;
    
    public ThreadPool(int poolSize){
        super("ThreadPool-" + (threadPoolId++));
        setDaemon(true);
        workQueue = new LinkedList<Runnable>();
        for(int i=0;i<poolSize;i++){
            new WorkThread(this,i).start();      // 启动工作者线程
        }
    }
    
    public synchronized void execute(Runnable task){
        if(isClosed){
            throw new IllegalStateException("The " + getName()
            		+ " has closed!");
        }
        if(task != null){
            workQueue.add(task);
            // 一次只有一个线程取任务[getTask() 被 synchronized 修饰]
            // 所有此处没有使用notifyAll();
            notify();           
        }        
    }
    
    protected synchronized Runnable getTask()throws InterruptedException{
        while(workQueue.size() == 0){
            if(isClosed){
                return null;
            }
            wait();
        }
        return workQueue.removeFirst();
    }
    /**
     * 直接关闭线程池
     */
    public synchronized void close(){
        if(!isClosed){
            isClosed = true;
            workQueue.clear();
            interrupt();             // 中断所有线程
        }
    }
    /**
     * 执行工作队列中的任务后关闭线程池
     */
    public void join(){
        synchronized(this){
            isClosed = true;
            notifyAll();
        }
        Thread[] threads = new Thread[activeCount()];
        int count = enumerate(threads);
        for(int i=0;i<count;i++){
            try{
                threads[i].join();
            } catch(InterruptedException ex){
            }
        }
    }
    /**
     * 工作者线程
     * @author dycc
     *
     */
    private class WorkThread extends Thread{
        public WorkThread(ThreadGroup group,int threadId){
            super(group,"WorkThread-" + threadId);
        }
        /**
         * 工作者线程关闭的原因
         * 1. ThreadPool.close() -> 中断所有线程 -> 线程退出;
         * 2. workQueue.size() == 0 && isClosed -> getTask() == null -> 线程退出;
         */
        public void run(){
            while(!isInterrupted()){
                Runnable task = null;
                try{
                    task = getTask();
                } catch(InterruptedException ex){
                }
                if(task == null){
                    return;
                }
                try{
                    task.run();
                } catch(Throwable t){
                    t.printStackTrace();
                }
            }
        }
    }
}

// multi.WorkHandler.java
package multi;

import java.io.BufferedReader;
import java.io.IOException;
import java.io.InputStreamReader;
import java.io.PrintWriter;
import java.net.Socket;
import java.text.SimpleDateFormat;
import java.util.Date;

public class WorkHandler implements Runnable{
	private Socket socket;
	private BufferedReader in;
	private PrintWriter out;
	private SimpleDateFormat simpleFormat = new SimpleDateFormat(
			                                  "yyyy-MM-dd HH:mm:ss:S");
	
	public WorkHandler(Socket socket)throws IOException{
		this.socket = socket;
		in = new BufferedReader(new InputStreamReader(socket.getInputStream()));
		out = new PrintWriter(socket.getOutputStream());
	}
	
	public void run() {
	    try{
	    	// start
	    	String time = simpleFormat.format(new Date());
	    	System.out.println(time + "-> connected succeed.");
	    	String msg = null;
	    	while((msg = in.readLine()) != null){
	    		out.println(echo(msg));
	    		out.flush();        // 此处的flush()非常重要，否则客户端无法收到消息
	    		if(msg.equalsIgnoreCase("bye")){
	    			break;
	    		}
	    	}
	    } catch(IOException e){
	    	e.printStackTrace();
	    } finally{
	    	try{
	    		if(socket != null){
		    		socket.close();
		    	}
	    	} catch(Exception e){
	    		e.printStackTrace();
	    	}
	    	// end
	    	String time = simpleFormat.format(new Date());
	    	System.out.println(time + "-> connection closed.");
	    }
	}
	public String echo(String msg){
		return "server back:" + msg;
	}
}

// Client.java
import java.net.Socket;

import multi.ClientHandler;

public class Client{
    public static void main(String[] args)throws Exception{
        final int length = 10;
        String host = "localhost";
        int port = 8000;
        
        Socket[] sockets = new Socket[length];
        for(int i=0;i<length;i++){
            sockets[i] = new Socket(host,port);
            System.out.println(i + "--- client connected! ---");
            // 
            Thread client_thread = new Thread(new ClientHandler(sockets[i]));
            client_thread.start();
        }
    }
}

// multi.ClientHandler.java
package multi;

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

public class ClientHandler implements Runnable{
	private Socket socket;
	private BufferedReader in;
	private PrintWriter out;
	
	public ClientHandler(Socket socket)throws IOException{
		this.socket = socket;
		in = new BufferedReader(new InputStreamReader(socket.getInputStream()));
		out = new PrintWriter(socket.getOutputStream());
	}
	
	public void run() {
	    try{
	    	out.println("hi baby.");
	    	out.println("bye");
	    	out.flush();            // 此处的flush()非常重要，否则服务端无法收到消息
	    	String msg = null;
	    	while((msg = in.readLine()) != null){
	    		System.out.println(msg);
	    		if(msg.indexOf("bye") > -1){
	    			break;
	    		}
	    	}
	    } catch(IOException e){
	    	e.printStackTrace();
	    } finally{
	    	try{
	    		if(socket != null){
		    		socket.close();
		    	}
	    	} catch(Exception e){
	    		e.printStackTrace();
	    	}	    	
	    }
	}
}