基于事件的NIO多线程服务器-宜宾勤盛利枫建站

新闻中心

这里有您想知道的互联网营销解决方案

基于事件的NIO多线程服务器

JDK1.4的NIO有效解决了原有流式IO存在的线程开销的问题，在NIO中使用多线程，主要目的已不是为了应对每个客户端请求而分配独立的服务线程，而是通过多线程充分使用用多个CPU的处理能力和处理中的等待时间，达到提高服务能力的目的。

目前创新互联已为上千余家的企业提供了网站建设、域名、网页空间、网站托管、服务器租用、企业网站设计、洞口网站维护等服务，公司将坚持客户导向、应用为本的策略，正道将秉承"和谐、参与、激情"的文化，与客户和合作伙伴齐心协力一起成长，共同发展。

线程模型

NIO的选择器采用了多路复用（Multiplexing）技术，可在一个选择器上处理多个套接字，通过获取读写通道来进行IO操作。由于网络带宽等原因，在通道的读、写操作中是容易出现等待的，所以在读、写操作中引入多线程，对性能提高明显，而且可以提高客户端的感知服务质量。所以本文的模型将主要通过使用读、写线程池来提高与客户端的数据交换能力。

同时整个服务端的流程处理，建立于事件机制上。在 [接受连接－＞读－＞业务处理－＞写－＞关闭连接 ]这个过程中，触发器将触发相应事件，由事件处理器对相应事件分别响应，完成服务器端的业务处理。
下面我们就来详细看一下这个模型的各个组成部分。

相关事件定义在这个模型中，我们定义了一些基本的事件：

（1）onAccept：

当服务端收到客户端连接请求时，触发该事件。通过该事件我们可以知道有新的客户端呼入。该事件可用来控制服务端的负载。例如，服务器可设定同时只为一定数量客户端提供服务，当同时请求数超出数量时，可在响应该事件时直接抛出异常，以拒绝新的连接。

（2）onAccepted：

当客户端请求被服务器接受后触发该事件。该事件表明一个新的客户端与服务器正式建立连接。

（3）onRead：

当客户端发来数据，并已被服务器控制线程正确读取时，触发该事件。该事件通知各事件处理器可以对客户端发来的数据进行实际处理了。需要注意的是，在本模型中，客户端的数据读取是由控制线程交由读线程完成的，事件处理器不需要在该事件中进行专门的读操作，而只需将控制线程传来的数据进行直接处理即可。

（4）onWrite：

当客户端可以开始接受服务端发送数据时触发该事件，通过该事件，我们可以向客户端发送回应数据。在本模型中，事件处理器只需要在该事件中设置。

（5）onClosed：

当客户端与服务器断开连接时触发该事件。

（6）onError：

当客户端与服务器从连接开始到***断开连接期间发生错误时触发该事件。通过该事件我们可以知道有什么错误发生。

事件回调机制的实现

在这个模型中，事件采用广播方式，也就是所有在册的事件处理器都能获得事件通知。这样可以将不同性质的业务处理，分别用不同的处理器实现，使每个处理器的业务功能尽可能单一。

如下图：整个事件模型由监听器、事件适配器、事件触发器、事件处理器组成。

（事件模型）

1.监听器（Serverlistener）：

这是一个事件接口，定义需监听的服务器事件，如果您需要定义更多的事件，可在这里进行扩展。

 
 
 
 
  
  
  
  public interface Serverlistener  
  
  
  
  { 
  
  
  
  public void onError(String error); 

  
  
  
  public void onAccept() throws Exception; 

  
  
  
  public void onAccepted(Request request) throws Exception; 

  
  
  
  public void onRead(Request request) throws Exception; 

  
  
  
  public void onWrite(Request request, Response response) throws Exception; 

  
  
  
  public void onClosed(Request request) throws Exception; 

  
  
  
  }

2. 事件适配器（EventAdapter）：

对Serverlistener接口实现一个适配器(EventAdapter)，这样的好处是最终的事件处理器可以只处理所关心的事件。

 
 
 
 
  
  
  
  public abstract class EventAdapter  
  
  
  
  implements Serverlistener  
  
  
  
  { 
  
  
  
  public EventAdapter() {} 
  
  
  
  public void onError(String error) {} 
  
  
  
  public void onAccept() throws Exception {} 
  
  
  
  public void onAccepted(Request request) throws Exception {} 
  
  
  
  public void onRead(Request request) throws Exception {} 
  
  
  
  public void onWrite(Request request, Response response) throws Exception {} 
  
  
  
  public void onClosed(Request request) throws Exception {} 
  
  
  
  }

3. 事件触发器（Notifier）：

用于在适当的时候通过触发服务器事件，通知在册的事件处理器对事件做出响应。触发器以Singleton模式实现，统一控制整个服务器端的事件，避免造成混乱。

 
 
 
 
  
  
  
  public class Notifier  
  
  
  
  { 
  
  
  
  private static Arraylist listeners = null; 
  
  
  
  private static Notifier instance = null; 
  
  
  
   
  
  
  
  private Notifier()  
  
  
  
  { 
  
  
  
  listeners = new Arraylist(); 
  
  
  
  } 
  
  
  
   
  
  
  
  /** 
  
  
  
  * 获取事件触发器 
  
  
  
  * @return 返回事件触发器 
  
  
  
  */ 
  
  
  
  public static synchronized Notifier  
  
  
  
  getNotifier()  
  
  
  
  { 
  
  
  
  if (instance == null)  
  
  
  
  { 
  
  
  
  instance = new Notifier(); 
  
  
  
  return instance; 
  
  
  
  } 
  
  
  
  else  
  
  
  
  { 
  
  
  
  return instance; 
  
  
  
  } 
  
  
  
  } 
  
  
  
   
  
  
  
  /** 
  
  
  
  * 添加事件监听器 
  
  
  
  * @param l 监听器 
  
  
  
  */ 
  
  
  
  public void addlistener(Serverlistener l) 
  
  
  
  { 
  
  
  
  synchronized (listeners) 
  
  
  
  { 
  
  
  
  if (!listeners.contains(l)) 
  
  
  
  { 
  
  
  
  listeners.add(l); 
  
  
  
  } 
  
  
  
  } 
  
  
  
  } 
  
  
  
   
  
  
  
  public void fireOnAccept()  
  
  
  
  throws Exception  
  
  
  
  { 
  
  
  
  for (int i = listeners.size() - 1;  
  
  
  
  i >= 0; i--) 
  
  
  
  { 
  
  
  
  ( (Serverlistener) listeners. 
  
  
  
  get(i)).onAccept(); 
  
  
  
  } 
  
  
  
  } 
  
  
  
  // other fire method 
  
  
  
  }

4. 事件处理器（Handler）：

继承事件适配器，对感兴趣的事件进行响应处理，实现业务处理。以下是一个简单的事件处理器实现，它响应onRead事件，在终端打印出从客户端读取的数据。

 
 
 
 
  
  
  
  public class ServerHandler  
  
  
  
  extends EventAdapter  
  
  
  
  { 
  
  
  
  public ServerHandler() {} 
  
  
  
   
  
  
  
  public void onRead(Request request)  
  
  
  
  throws Exception  
  
  
  
  { 
  
  
  
  System.out.println("Received: " + 
  
  
  
  new String(data)); 
  
  
  
  } 
  
  
  
  }

5. 事件处理器的注册。

为了能让事件处理器获得服务线程的事件通知，事件处理器需在触发器中注册。

 
 
 
 
  
  
  
  ServerHandler handler = new ServerHandler(); 
  
  
  
  Notifier.addlistener(handler);

实现NIO多线程服务器

NIO多线程服务器主要由主控服务线程、读线程和写线程组成。

1. 主控服务线程（Server）：

主控线程将创建读、写线程池，实现监听、接受客户端请求，同时将读、写通道提交由相应的读线程（Reader）和写服务线程(Writer），由读写线程分别完成对客户端数据的读取和对客户端的回应操作。

 
 
 
 
  
  
  
  public class Server implements Runnable 
  
  
  
  { 
  
  
  
  private static List wpool = new LinkedList();  
  
  
  
  private static Selector selector; 
  
  
  
  private ServerSocketChannel sschannel; 
  
  
  
  private InetSocketAddress address; 
  
  
  
  protected Notifier notifier; 
  
  
  
  private int port; 
  
  
  
  private static int MAX_THREADS = 4; 
  
  
  
   
  
  
  
  /** 
  
  
  
  * Creat the main thread 
  
  
  
  * @param port server port 
  
  
  
  * @throws java.lang.Exception 
  
  
  
  */ 
  
  
  
  public Server(int port) throws Exception 
  
  
  
  { 
  
  
  
  this.port = port; 
  
  
  
   
  
  
  
  // event dispatcher 
  
  
  
  notifier = Notifier.getNotifier(); 
  
  
  
   
  
  
  
  // create the thread pool for reading and writing 
  
  
  
  for (int i = 0; i < MAX_THREADS; i++) 
  
  
  
  { 
  
  
  
  Thread r = new Reader(); 
  
  
  
  Thread w = new Writer(); 
  
  
  
  r.start(); 
  
  
  
  w.start(); 
  
  
  
  } 
  
  
  
   
  
  
  
  // create nonblocking socket 
  
  
  
  selector = Selector.open(); 
  
  
  
  sschannel = ServerSocketChannel.open(); 
  
  
  
  sschannel.configureBlocking(false); 
  
  
  
  address = new InetSocketAddress(port); 
  
  
  
  ServerSocket ss = sschannel.socket(); 
  
  
  
  ss.bind(address); 
  
  
  
  sschannel.register(selector, SelectionKey.OP_ACCEPT); 
  
  
  
  } 
  
  
  
   
  
  
  
  public void run() 
  
  
  
  { 
  
  
  
  System.out.println("Server started "); 
  
  
  
  System.out.println("Server listening on port: " + port); 
  
  
  
   
  
  
  
  while (true) 
  
  
  
  { 
  
  
  
  try 
  
  
  
  { 
  
  
  
  int num = 0; 
  
  
  
  num = selector.select(); 
  
  
  
   
  
  
  
  if (num > 0) 
  
  
  
  { 
  
  
  
  Set selectedKeys = selector.selectedKeys(); 
  
  
  
  Iterator it = selectedKeys.iterator(); 
  
  
  
  while (it.hasNext()) 
  
  
  
  { 
  
  
  
  SelectionKey key = (SelectionKey) it.next(); 
  
  
  
  it.remove(); 
  
  
  
   
  
  
  
  if ((key.readyOps() & SelectionKey.OP_ACCEPT) == SelectionKey.OP_ACCEPT) 
  
  
  
  { 
  
  
  
  // Accept the new connection 
  
  
  
  ServerSocketChannel ssc =  
  
  
  
  (ServerSocketChannel) key.channel(); 
  
  
  
  notifier.fireOnAccept(); 
  
  
  
   
  
  
  
  SocketChannel sc = ssc.accept(); 
  
  
  
  sc.configureBlocking(false); 
  
  
  
   
  
  
  
  Request request = new Request(sc); 
  
  
  
  notifier.fireOnAccepted(request); 
  
  
  
   
  
  
  
  sc.register(selector, SelectionKey.OP_READ,request); 
  
  
  
  }  
  
  
  
  else if ((key.readyOps() & SelectionKey.OP_READ) == SelectionKey.OP_READ) 
  
  
  
  { 
  
  
  
  Reader.processRequest(key);  
  
  
  
  key.cancel(); 
  
  
  
  }  
  
  
  
  else if ((key.readyOps() & SelectionKey.OP_WRITE) == SelectionKey.OP_WRITE) 
  
  
  
  { 
  
  
  
  Writer.processRequest(key); 
  
  
  
  key.cancel(); 
  
  
  
  } 
  
  
  
  } 
  
  
  
  }  
  
  
  
  //this selector's wakeup method is invoked 
  
  
  
  else 
  
  
  
  { 
  
  
  
  //register new channel for writing to selector 
  
  
  
  addRegister();  
  
  
  
  } 
  
  
  
  }  
  
  
  
  catch (Exception e) 
  
  
  
  { 
  
  
  
  notifier.fireOnError("Error occured in Server: " 
  
  
  
  + e.getMessage()); 
  
  
  
  continue; 
  
  
  
  } 
  
  
  
  } 
  
  
  
  } 
  
  
  
   
  
  
  
  private void addRegister() 
  
  
  
  { 
  
  
  
  synchronized (wpool) 
  
  
  
  { 
  
  
  
  while (!wpool.isEmpty()) 
  
  
  
  { 
  
  
  
  SelectionKey key = (SelectionKey) wpool.remove(0); 
  
  
  
  SocketChannel schannel = (SocketChannel) key.channel(); 
  
  
  
  try 
  
  
  
  { 
  
  
  
  schannel.register(selector, SelectionKey.OP_WRITE, key 
  
  
  
  .attachment()); 
  
  
  
  } 
  
  
  
  catch (Exception e) 
  
  
  
  { 
  
  
  
  try 
  
  
  
  { 
  
  
  
  schannel.finishConnect(); 
  
  
  
  schannel.close(); 
  
  
  
  schannel.socket().close(); 
  
  
  
  notifier.fireOnClosed((Request) key.attachment()); 
  
  
  
  }  
  
  
  
  catch (Exception e1) 
  
  
  
  { 
  
  
  
  } 
  
  
  
  notifier.fireOnError("Error occured in addRegister: " 
  
  
  
  + e.getMessage()); 
  
  
  
  } 
  
  
  
  } 
  
  
  
  } 
  
  
  
  } 
  
  
  
   
  
  
  
  public static void processWriteRequest(SelectionKey key) 
  
  
  
  { 
  
  
  
  synchronized (wpool) 
  
  
  
  { 
  
  
  
  wpool.add(wpool.size(), key); 
  
  
  
  wpool.notifyAll(); 
  
  
  
  } 
  
  
  
  selector.wakeup();  
  
  
  
  } 
  
  
  
  }

2. 读线程（Reader）：

使用线程池技术，通过多个线程读取客户端数据，以充分利用网络数据传输的时间，提高读取效率。

 
 
 
 
  
  
  
  public class Reader extends Thread  
  
  
  
  { 
  
  
  
  public void run()  
  
  
  
  { 
  
  
  
  while (true)  
  
  
  
  { 
  
  
  
  try  
  
  
  
  { 
  
  
  
  SelectionKey key; 
  
  
  
  synchronized (pool)  
  
  
  
  { 
  
  
  
  while (pool.isEmpty())  
  
  
  
  { 
  
  
  
  pool.wait(); 
  
  
  
  } 
  
  
  
  key = (SelectionKey) pool.remove(0); 
  
  
  
  } 
  
  
  
  // 读取客户端数据，并触发onRead事件 
  
  
  
  read(key);  
  
  
  
  } 
  
  
  
  catch (Exception e)  
  
  
  
  { 
  
  
  
  continue; 
  
  
  
  } 
  
  
  
  } 
  
  
  
  } 
  
  
  
  }

3. 写线程（Writer）：

和读操作一样，使用线程池，负责将服务器端的数据发送回客户端。

 
 
 
 
  
  
  
  public final class Writer extends Thread  
  
  
  
  { 
  
  
  
  public void run()  
  
  
  
  { 
  
  
  
  while (true)  
  
  
  
  { 
  
  
  
  try  
  
  
  
  { 
  
  
  
  SelectionKey key; 
  
  
  
  synchronized (pool)  
  
  
  
  { 
  
  
  
  while (pool.isEmpty())  
  
  
  
  { 
  
  
  
  pool.wait(); 
  
  
  
  } 
  
  
  
  key = (SelectionKey) pool.remove(0); 
  
  
  
  } 
  
  
  
   
  
  
  
  write(key);  
  
  
  
  } 
  
  
  
  catch (Exception e)  
  
  
  
  { 
  
  
  
  continue; 
  
  
  
  } 
  
  
  
  } 
  
  
  
  } 
  
  
  
  }

具体应用

NIO多线程模型的实现告一段落，现在我们可以暂且将NIO的各个API和烦琐的调用方法抛于脑后，专心于我们的实际应用中。

我们用一个简单的TimeServer（时间查询服务器）来看看该模型能带来多么简洁的开发方式。

在这个TimeServer中，将提供两种语言（中文、英文）的时间查询服务。我们将读取客户端的查询命令（GB/EN），并回应相应语言格式的当前时间。在应答客户的请求的同时，服务器将进行日志记录。做为示例，对日志记录，我们只是简单地将客户端的访问时间和IP地址输出到服务器的终端上。

1. 实现时间查询服务的事件处理器（TimeHandler）：

 
 
 
 
  
  
  
  public class TimeHandler extends EventAdapter  
  
  
  
  { 
  
  
  
  public TimeHandler() {} 
  
  
  
   
  
  
  
  public void onWrite(Request request, Response response) throws Exception  
  
  
  
  { 
  
  
  
  String command = new String(request.getDataInput()); 
  
  
  
  String time = null; 
  
  
  
  Date date = new Date(); 
  
  
  
   
  
  
  
  if (command.equals("GB"))  
  
  
  
  { 
  
  
  
  DateFormat cnDate = DateFormat.getDateTimeInstance(DateFormat.FulL, 
  
  
  
  DateFormat.FulL, Locale.CHINA); 
  
  
  
  time = cnDate.format(date); 
  
  
  
  } 
  
  
  
  else  
  
  
  
  { 
  
  
  
  DateFormat enDate = DateFormat.getDateTimeInstance(DateFormat.FulL, 
  
  
  
  DateFormat.FulL, Locale.US); 
  
  
  
  time = enDate.format(date); 
  
  
  
  } 
  
  
  
   
  
  
  
  response.send(time.getBytes()); 
  
  
  
  } 
  
  
  
  }

2. 实现日志记录服务的事件处理器（LogHandler）：

 
 
 
 
  
  
  
  public class LogHandler extends EventAdapter  
  
  
  
  { 
  
  
  
  public LogHandler() {} 
  
  
  
   
  
  
  
  public void onClosed(Request request)  
  
  
  
  throws Exception  
  
  
  
  { 
  
  
  
  String log = new Date().toString() + " from " + request.getAddress().toString(); 
  
  
  
  System.out.println(log); 
  
  
  
  } 
  
  
  
   
  
  
  
  public void onError(String error)  
  
  
  
  { 
  
  
  
  System.out.println("Error: " + error); 
  
  
  
  } 
  
  
  
  }

3. 启动程序：

 
 
 
 
  
  
  
  public class Start  
  
  
  
  { 
  
  
  
   
  
  
  
  public static void main(String[] args)  
  
  
  
  { 
  
  
  
  try  
  
  
  
  { 
  
  
  
  LogHandler loger = new LogHandler(); 
  
  
  
  TimeHandler timer = new TimeHandler(); 
  
  
  
  Notifier notifier = Notifier.getNotifier(); 
  
  
  
  notifier.addlistener(loger); 
  
  
  
  notifier.addlistener(timer); 
  
  
  
   
  
  
  
  System.out.println("Server starting "); 
  
  
  
  Server server = new Server(5100); 
  
  
  
  Thread tServer = new Thread(server); 
  
  
  
  tServer.start(); 
  
  
  
  } 
  
  
  
  catch (Exception e)  
  
  
  
  { 
  
  
  
  System.out.println("Server error: " + e.getMessage()); 
  
  
  
  System.exit(-1); 
  
  
  
  } 
  
  
  
  } 
  
  
  
  }

小结

通过例子我们可以看到，基于事件回调的NIO多线程服务器模型，提供了清晰直观的实现方式，可让开发者从NIO及多线程的技术细节中摆脱出来，集中精力关注具体的业务实现。

原文链接：http://www.cnblogs.com/longb/archive/2006/04/04/366800.html

【编辑推荐】

Java NIO的多路复用及reactor
在Java中使用NIO进行网络编程
Java NIO非阻塞服务器示例
基于Java NIO的即时聊天服务器模型
Java解读NIO Socket非阻塞模式

分享题目：基于事件的NIO多线程服务器
文章出自：http://www.aqpgk.com/article/coephss.html

新闻中心

其他资讯