阻塞式线程池服务器程序实例

Excutor 接口 --- 线程池
            .execute(Runnable task) --- 执行Runnable类型的任务。

Executors  --- 负责生成各种连接池的实例
           .newCachedThreadPool()  --- 有任务时才创建新线程，空闲线程保留60s。
           .newFixedThreadPool(int nThreads)  --- 创建固定数量的线程，空闲线程会一直保留。 
           .newScheduledThreadPool(int corePoolSize) --- 线程池按照时间计划执行任务，允许设置任务执行时  间。参数为最小线程数。繁忙时，可能创建更多线程。
           .newSingleThreadExecutor() --- 只建立一个工作线程，依次执行每个任务。
           .newSingleThreadScheduledExecutor() ----只一个线程，按时间计划执行任务。 
           

ExecutorService --- 负责连接池的管理           
            

import java.io.*;
import java.nio.*;
import java.nio.channels.*;
import java.nio.charset.*;
import java.net.*;
import java.util.*;
import java.util.concurrent.*;

public class EchoServer {
  private int port=8001;
  private ServerSocketChannel serverSocketChannel = null;
  private ExecutorService executorService;   //线程池
  private static final int POOL_MULTIPLE = 4;   //线程池数目
  
  /**
   * 负责构造线程池，启动服务区，并绑定端口
   * @throws IOException
   */
  public EchoServer() throws IOException {
	//创建线程池
    executorService= Executors.newFixedThreadPool(
	    Runtime.getRuntime().availableProcessors() * POOL_MULTIPLE);
    //创建通道
    serverSocketChannel= ServerSocketChannel.open();
    //在同一主机上关闭服务器程序，再启动该服务器服务时，能绑定同一端口。
    serverSocketChannel.socket().setReuseAddress(true);
    //绑定本地端口
    serverSocketChannel.socket().bind(new InetSocketAddress(port));
    System.out.println("服务器启动");
  }
  /**
   * 负责客户端连接，每接受一个客户连接，就把它交给线程池来处理。
   * 1.从线程池中取出空闲线程
   * 2.执行Handler的run()方法。
   */
  public void service() {
    while (true) {
      SocketChannel socketChannel=null;
      try {
        socketChannel = serverSocketChannel.accept();// 从连接池中获取一个连接通道
        executorService.execute(new Handler(socketChannel)); // 在工作线程中进行服务
      }catch (IOException e) {
         e.printStackTrace();
      }
    }
  }

  public static void main(String args[])throws IOException {
    new EchoServer().service();
  }
}
/**
 * 负责客户通信
 * @author Jethro Yan
 *
 */
class Handler implements Runnable{
  private SocketChannel socketChannel;
  public Handler(SocketChannel socketChannel){
    this.socketChannel=socketChannel;
  }
  public void run(){
    handle(socketChannel);
  }
  
  //负责和客户端通信工作。
  public void handle(SocketChannel socketChannel){
    try {
        Socket socket=socketChannel.socket();// 获得socket。
        System.out.println("接收到客户连接，来自: " +
        socket.getInetAddress() + ":" +socket.getPort());

        BufferedReader br =getReader(socket);
        PrintWriter pw = getWriter(socket);

        String msg = null;
        while ((msg = br.readLine()) != null) {
          System.out.println(msg);
          pw.println(echo(msg));
          if (msg.equals("bye"))
            break;
        }
      }catch (IOException e) {
         e.printStackTrace();
      }finally {
         try{
           if(socketChannel!=null)socketChannel.close();
         }catch (IOException e) {e.printStackTrace();}
      }
  }

  private PrintWriter getWriter(Socket socket)throws IOException{
    OutputStream socketOut = socket.getOutputStream();
    return new PrintWriter(socketOut,true);
  }
  private BufferedReader getReader(Socket socket)throws IOException{
    InputStream socketIn = socket.getInputStream();
    return new BufferedReader(new InputStreamReader(socketIn));
  }

  public String echo(String msg) {
    return "echo:" + msg;
  }
}

 

评论

1 楼 ak913 2012-09-08  

Line41：        executorService.execute(new Handler(socketChannel)); // 在工作线程中进行服务 
这一步不是阻塞操作，也就是说如果池中的任务处理速度不够，而新请求也可以源源不断地加入到线程池的等待队列，不会被阻塞