Java NIO 非阻塞Socket服务器构建

  1 public class EchoServer {
  2     public static SelectorLoop connectionBell;
  3     public static SelectorLoop readBell;
  4     public boolean isReadBellRunning=false;
  5  
  6     public static void main(String[] args) throws IOException {
  7         new EchoServer().startServer();
  8     }
  9      
 10     // 启动服务器
 11     public void startServer() throws IOException {
 12         // 准备好一个闹钟.当有链接进来的时候响.
 13         connectionBell = new SelectorLoop();
 14          
 15         // 准备好一个闹装,当有read事件进来的时候响.
 16         readBell = new SelectorLoop();
 17          
 18         // 开启一个server channel来监听
 19         ServerSocketChannel ssc = ServerSocketChannel.open();
 20         // 开启非阻塞模式
 21         ssc.configureBlocking(false);
 22          
 23         ServerSocket socket = ssc.socket();
 24         socket.bind(new InetSocketAddress("localhost",7878));
 25          
 26         // 给闹钟规定好要监听报告的事件,这个闹钟只监听新连接事件.
 27         ssc.register(connectionBell.getSelector(), SelectionKey.OP_ACCEPT);
 28         new Thread(connectionBell).start();
 29     }
 30      
 31     // Selector轮询线程类
 32     public class SelectorLoop implements Runnable {
 33         private Selector selector;
 34         private ByteBuffer temp = ByteBuffer.allocate(1024);
 35          
 36         public SelectorLoop() throws IOException {
 37             this.selector = Selector.open();
 38         }
 39          
 40         public Selector getSelector() {
 41             return this.selector;
 42         }
 43  
 44         @Override
 45         public void run() {
 46             while(true) {
 47                 try {
 48                     // 阻塞,只有当至少一个注册的事件发生的时候才会继续.
 49                     this.selector.select();
 50                      
 51                     Set<SelectionKey> selectKeys = this.selector.selectedKeys();
 52                     Iterator<SelectionKey> it = selectKeys.iterator();
 53                     while (it.hasNext()) {
 54                         SelectionKey key = it.next();
 55                         it.remove();
 56                         // 处理事件. 可以用多线程来处理.
 57                         this.dispatch(key);
 58                     }
 59                 } catch (IOException e) {
 60                     e.printStackTrace();
 61                 } catch (InterruptedException e) {
 62                     e.printStackTrace();
 63                 }
 64             }
 65         }
 66          
 67         public void dispatch(SelectionKey key) throws IOException, InterruptedException {
 68             if (key.isAcceptable()) {
 69                 // 这是一个connection accept事件, 并且这个事件是注册在serversocketchannel上的.
 70                 ServerSocketChannel ssc = (ServerSocketChannel) key.channel();
 71                 // 接受一个连接.
 72                 SocketChannel sc = ssc.accept();
 73                  
 74                 // 对新的连接的channel注册read事件. 使用readBell闹钟.
 75                 sc.configureBlocking(false);
 76                 sc.register(readBell.getSelector(), SelectionKey.OP_READ);
 77                  
 78                 // 如果读取线程还没有启动,那就启动一个读取线程.
 79                 synchronized(EchoServer.this) {
 80                     if (!EchoServer.this.isReadBellRunning) {
 81                         EchoServer.this.isReadBellRunning = true;
 82                         new Thread(readBell).start();
 83                     }
 84                 }
 85                  
 86             } else if (key.isReadable()) {
 87                 // 这是一个read事件,并且这个事件是注册在socketchannel上的.
 88                 SocketChannel sc = (SocketChannel) key.channel();
 89                 // 写数据到buffer
 90                 int count = sc.read(temp);
 91                 if (count < 0) {
 92                     // 客户端已经断开连接.
 93                     key.cancel();
 94                     sc.close();
 95                     return;
 96                 }
 97                 // 切换buffer到读状态,内部指针归位.
 98                 temp.flip();
 99                 String msg = Charset.forName("UTF-8").decode(temp).toString();
100                 System.out.println("Server received ["+msg+"] from client address:" + sc.getRemoteAddress());
101                  
102                 Thread.sleep(1000);
103                 // echo back.
104                 sc.write(ByteBuffer.wrap(msg.getBytes(Charset.forName("UTF-8"))));
105                  
106                 // 清空buffer
107                 temp.clear();
108             }
109         }
111     }
112  
113 }