Java 网络编程 —— 实现非阻塞式的服务器

创建阻塞的服务器

当 ServerSocketChannel 与 SockelChannel 采用默认的阻塞模式时，为了同时处理多个客户的连接，必须使用多线程

1. public class EchoServer {
2.    
3.         private int port = 8000;
4.     private ServerSocketChannel serverSocketChannel = null;
5.     private ExecutorService executorService; //线程池
6.     private static final int POOL_MULTIPLE = 4; //线程池中工作线程的数目
7.    
8.     public EchoServer() throws IOException {
9.         //创建一个线程池
10.         executorService = Executors.newFixedThreadPool(
11.             Runtime.getRuntime().availableProcessors() * POOL_MULTIPLE);
12.         //创建一个ServerSocketChannel对象
13.         serverSocketChannel = ServerSocketChannel.open();
14.         //使得在同一个主机上关闭了服务器程序，紧接着再启动该服务器程序时，可以顺利绑定相同的端口
15.         serverSocketChannel.socket().setReuseAddress(true);
16.         //把服务器进程与一个本地端口绑定
17.         serverSocketChannel.socket().bind(new InetSocketAddress(port));
18.         System.out.println("服务器启动");
19.     }
20.    
21.     public void service() {
22.         while (true) {
23.             SocketChannel socketChannel = null;
24.             try {
25.                 socketChannel = serverSocketChannel.accept();
26.                 //处理客户连接
27.                 executorService.execute(new Handler(socketChannel));
28.             } catch(IOException e) {
29.                 e.printStackTrace();
30.             }
31.         }
32.     }
33.    
34.     public static void main(String args[])throws IOException {
35.         new EchoServer().service();
36.     }
37.    
38.     //处理客户连按
39.     class Handler implements Runnable {
41.         private SocketChannel socketChannel;
42.                
43.         public Handler(SocketChannel socketChannel) {
44.             this.socketChannel = socketChannel;
45.         }
46.         
47.         public void run() {
48.             handle(socketChannel);
49.         }
50.         
51.         public void handle(SocketChannel socketChannel) {
52.             try {
53.                 //获得与socketChannel关联的Socket对象
54.                 Socket socket = socketChannel.socket();
55.                 System.out.println("接收到客户连接，来自:" + socket.getInetAddress() + ":" + socket.getPort());
56.                
57.                 BufferedReader br = getReader(socket);
58.                 PrintWriter pw = getWriter(socket);
59.                
60.                 String msg = null;
61.                 while ((msg = br.readLine()) != null) {
62.                     System.out.println(msg);
63.                     pw.println(echo(msg));
64.                     if (msg.equals("bye")) {
65.                         break;
66.                     }
67.                 }
68.             } catch (IOException e) {
69.                 e.printStackTrace();
70.             } finally {
71.                 try {
72.                     if(socketChannel != null) {
73.                         socketChannel.close();
74.                     } catch (IOException e) {
75.                         e.printStackTrace();
76.                     }
77.                 }
78.             }
79.         }
80.     }
81.    
82.     private PrintWriter getWriter(Socket socket) throws IOException {
83.         OutputStream socketOut = socket.getOutputStream();
84.         return new PrintWriter(socketOut,true);
85.     }
86.    
87.     private BufferedReader getReader(Socket socket) throws IOException {
88.         InputStream socketIn = socket.getInputStream();
89.         return new BufferedReader(new InputStreamReader(socketIn));
90.     }
91.    
92.     public String echo(String msg) {
93.         return "echo:" + msg;
94.     }
95. }

复制代码

创建非阻塞的服务器

在非阻塞模式下，EchoServer 只需要启动一个主线程，就能同时处理三件事：

• 接收客户的连接
  
• 接收客户发送的数据
  
• 向客户发回响应数据
  

EchoServer 委托 Selector 来负责监控接收连接就绪事件、读就绪事件和写就绪事件如果有特定事件发生，就处理该事件

1. // 创建一个Selector对象
2. selector = Selector.open();
3. //创建一个ServerSocketChannel对象
4. serverSocketChannel = ServerSocketChannel.open();
5. //使得在同一个主机上关闭了服务器程序，紧接着再启动该服务器程序时
6. //可以顺利绑定到相同的端口
7. serverSocketChannel.socket().setReuseAddress(true);
8. //使ServerSocketChannel工作于非阻塞模式
9. serverSocketChannel.configureBlocking(false):
10. //把服务器进程与一个本地端口绑定
11. serverSocketChannelsocket().bind(new InetSocketAddress(port));

复制代码

EchoServer 类的 service() 方法负责处理本节开头所说的三件事，体现其主要流程的代码如下：

1. public void service() throws IOException {
2.     serverSocketChannel.reqister(selector, SelectionKey.OP_ACCEPT);
3.     //第1层while循环
4.     while(selector.select() > 0) {
5.         //获得Selector的selected-keys集合
6.         Set readyKeys = selector.selectedKeys();
7.         Iterator it = readyKeys.iterator();
8.         //第2层while循环
9.         while (it.hasNext()) {
10.             SelectionKey key = null;
11.             //处理SelectionKey
12.             try {
13.                 //取出一个SelectionKey
14.                 key = (SelectionKey) it.next();
15.                 //把 SelectionKey从Selector 的selected-key 集合中删除
16.                 it.remove();
17.                 1f (key.isAcceptable()) { 处理接收连接就绪事件; }
18.                 if (key.isReadable()) { 处理读就绪水件; }
19.                 if (key.isWritable()) { 处理写就绪事件; }
20.             } catch(IOException e) {
21.                 e.printStackTrace();
22.                 try {
23.                     if(key != null) {
24.                         //使这个SelectionKey失效
25.                         key.cancel();
26.                         //关闭与这个SelectionKey关联的SocketChannel
27.                         key.channel().close();
28.                     }
29.                 } catch(Exception ex) {
30.                     e.printStackTrace();
31.                 }
32.             }
33.         }
34.     }
35. }

复制代码

• 首先由 ServerSocketChannel 向 Selector 注册接收连接就绪事件，如果 Selector 监控到该事件发生,就会把相应的 SelectionKey 对象加入 selected-keys 集合
  
• 第一层 while 循环，不断询问 Selector 已经发生的事件，select() 方法返回当前相关事件已经发生的 SelectionKey 的个数，如果当前没有任何事件发生，该方法会阻塞下去，直到至少有一个事件发生。Selector 的 selectedKeys() 方法返回 selected-keys 集合，它存放了相关事件已经发生的 SelectionKey 对象
  
• 第二层 while 循环，从 selected-keys 集合中依次取出每个 SelectionKey 对象并从集合中删除，，然后调用 isAcceptable()、isReadable() 和 isWritable() 方法判断到底是哪种事件发生了，从而做出相应的处理
  

1. 处理接收连接就绪事件

1. if (key.isAcceptable()) {
2.     //获得与SelectionKey关联的ServerSocketChannel
3.     ServerSocketChannel ssc = (ServerSocketChannel) key.channel();
4.     //获得与客户连接的SocketChannel
5.     SocketChannel socketChannel = (SocketChannel) ssc.accept();
6.     //把Socketchannel设置为非阻塞模式
7.     socketChannel.configureBlocking(false);
8.     //创建一个用于存放用户发送来的数据的级冲区
9.     ByteBuffer buffer = ByteBuffer.allocate(1024);
10.     //Socketchannel向Selector注册读就绪事件和写就绪事件
11.     socketChannel.register(selector, SelectionKey.OP_READ | SelectionKey.OP_WRITE, buffer);
12. }

复制代码

2. 处理读就绪事件

1. public void receive(SelectionKey key) throws IOException {
2.     //获得与SelectionKey关联的附件
3.     ByteBuffer buffer = (ByteBuffer) key.attachment();
4.     //获得与SelectionKey关联的Socketchannel
5.     SocketChannel socketChannel = (SocketChannel)key.channel();
6.     //创建一个ByteBuffer用于存放读到的数据
7.     ByteBuffer readBuff = ByteBuffer.allocate(32);
8.     socketChannel.read(readBuff);
9.     readBuff.flip();
10.     //把buffer的极限设为容量
11.     buffer.limit(buffer.capacity());
12.     //把readBuff中的内容拷贝到buffer
13.     buffer.put(readBuff);
14. }

复制代码

3. 处理写就绪事件

1. public void send(SelectionKey key) throws IOException {
2.     //获得与SelectionKey关联的ByteBuffer
3.     ByteBuffer buffer = (ByteBuffer) key.attachment();
4.     //获得与SelectionKey关联的SocketChannel
5.     SocketChannel socketChannel = (SocketChannel) key.channel();
6.     buffer.flip();
7.     //按照GBK编码把buffer中的字节转换为字符串
8.     String data = decode(buffer);
9.     //如果还没有读到一行数据就返回
10.     if(data.indexOf("\r\n") == -1)
11.         return;
12.     //截取一行数据
13.     String outputData = data.substring(0, data.indexOf("\n") + 1);
14.     //把输出的字符串按照GBK编码转换为字节，把它放在outputBuffer中
15.     ByteBuffer outputBuffer = encode("echo:" + outputData);
16.     //输出outputBuffer的所有字节
17.     while(outputBuffer,hasRemaining())
18.         socketChannel.write(outputBuffer);
19.     //把outputData字符审按照GBK编码，转换为字节，把它放在ByteBuffer
20.     ByteBuffer temp = encode(outputData);
21.     //把buffer的位置设为temp的极限
22.     buffer.position(temp.limit()):
23.     //删除buffer已经处理的数据
24.     buffer.compact();
25.     //如果已经输出了字符串“bye\r\n”，就使SelectionKey失效，并关闭SocketChannel
26.     if(outputData.equals("bye\r\n")) {
27.         key.cancel();
28.         socketChannel.close();
29.     }
30. }

复制代码

完整代码如下：

1. public class EchoServer {
2.    
3.         private int port = 8000;
4.     private ServerSocketChannel serverSocketChannel = null;
5.     private Selector selector;
6.     private Charset charset = Charset.forName("GBK");
8.         public EchoServer() throws IOException {
9.         // 创建一个Selector对象
10.         selector = Selector.open();
11.         //创建一个ServerSocketChannel对象
12.         serverSocketChannel = ServerSocketChannel.open();
13.         //使得在同一个主机上关闭了服务器程序，紧接着再启动该服务器程序时
14.         //可以顺利绑定到相同的端口
15.         serverSocketChannel.socket().setReuseAddress(true);
16.         //使ServerSocketChannel工作于非阻塞模式
17.         serverSocketChannel.configureBlocking(false):
18.         //把服务器进程与一个本地端口绑定
19.         serverSocketChannelsocket().bind(new InetSocketAddress(port));
20.     }
21.    
22.     public void service() throws IOException {
23.         serverSocketChannel.reqister(selector, SelectionKey.OP_ACCEPT);
24.         //第1层while循环
25.         while(selector.select() > 0) {
26.             //获得Selector的selected-keys集合
27.             Set readyKeys = selector.selectedKeys();
28.             Iterator it = readyKeys.iterator();
29.             //第2层while循环
30.             while (it.hasNext()) {
31.                 SelectionKey key = null;
32.                 //处理SelectionKey
33.                 try {
34.                     //取出一个SelectionKey
35.                     key = (SelectionKey) it.next();
36.                     //把 SelectionKey从Selector 的selected-key 集合中删除
37.                     it.remove();
38.                     1f (key.isAcceptable()) {
39.                          //获得与SelectionKey关联的ServerSocketChannel
40.                         ServerSocketChannel ssc = (ServerSocketChannel) key.channel();
41.                         //获得与客户连接的SocketChannel
42.                         SocketChannel socketChannel = (SocketChannel) ssc.accept();
43.                         //把Socketchannel设置为非阻塞模式
44.                         socketChannel.configureBlocking(false);
45.                         //创建一个用于存放用户发送来的数据的级冲区
46.                         ByteBuffer buffer = ByteBuffer.allocate(1024);
47.                         //Socketchannel向Selector注册读就绪事件和写就绪事件
48.                         socketChannel.register(selector, SelectionKey.OP_READ | SelectionKey.OP_WRITE, buffer);
49.                     }
50.                     if (key.isReadable()) { receive(key); }
51.                     if (key.isWritable()) { send(key); }
52.                 } catch(IOException e) {
53.                     e.printStackTrace();
54.                     try {
55.                         if(key != null) {
56.                             //使这个SelectionKey失效
57.                             key.cancel();
58.                             //关闭与这个SelectionKey关联的SocketChannel
59.                             key.channel().close();
60.                         }
61.                     } catch(Exception ex) {
62.                         e.printStackTrace();
63.                     }
64.                 }
65.             }
66.         }
67.     }
68.    
69.     public void receive(SelectionKey key) throws IOException {
70.         //获得与SelectionKey关联的附件
71.         ByteBuffer buffer = (ByteBuffer) key.attachment();
72.         //获得与SelectionKey关联的Socketchannel
73.         SocketChannel socketChannel = (SocketChannel)key.channel();
74.         //创建一个ByteBuffer用于存放读到的数据
75.         ByteBuffer readBuff = ByteBuffer.allocate(32);
76.         socketChannel.read(readBuff);
77.         readBuff.flip();
78.         //把buffer的极限设为容量
79.         buffer.limit(buffer.capacity());
80.         //把readBuff中的内容拷贝到buffer
81.         buffer.put(readBuff);
82.     }
83.    
84.     public void send(SelectionKey key) throws IOException {
85.         //获得与SelectionKey关联的ByteBuffer
86.         ByteBuffer buffer = (ByteBuffer) key.attachment();
87.         //获得与SelectionKey关联的SocketChannel
88.         SocketChannel socketChannel = (SocketChannel) key.channel();
89.         buffer.flip();
90.         //按照GBK编码把buffer中的字节转换为字符串
91.         String data = decode(buffer);
92.         //如果还没有读到一行数据就返回
93.         if(data.indexOf("\r\n") == -1)
94.             return;
95.         //截取一行数据
96.         String outputData = data.substring(0, data.indexOf("\n") + 1);
97.         //把输出的字符串按照GBK编码转换为字节，把它放在outputBuffer中
98.         ByteBuffer outputBuffer = encode("echo:" + outputData);
99.         //输出outputBuffer的所有字节
100.         while(outputBuffer,hasRemaining())
101.             socketChannel.write(outputBuffer);
102.         //把outputData字符审按照GBK编码，转换为字节，把它放在ByteBuffer
103.         ByteBuffer temp = encode(outputData);
104.         //把buffer的位置设为temp的极限
105.         buffer.position(temp.limit()):
106.         //删除buffer已经处理的数据
107.         buffer.compact();
108.         //如果已经输出了字符串“bye\r\n”，就使SelectionKey失效，并关闭SocketChannel
109.         if(outputData.equals("bye\r\n")) {
110.             key.cancel();
111.             socketChannel.close();
112.         }
113.     }
114.    
115.     //解码
116.     public String decode(ByteBuffer buffer) {
117.         CharBuffer charBuffer = charset.decode(buffer);
118.         return charBuffer.toStrinq();
119.     }
120.    
121.     //编码
122.     public ByteBuffer encode(String str) {
123.         return charset.encode(str);
124.     }
125.    
126.     public static void main(String args[])throws Exception {
127.         EchoServer server = new EchoServer();
128.         server.service();
129.     }
130. }

复制代码

阻塞模式与非阻塞模式混合使用

使用非阻塞模式时，ServerSocketChannel 以及 SocketChannel 都被设置为非阻塞模式，这使得接收连接、接收数据和发送数据的操作都采用非阻塞模式，EchoServer 采用一个线程同时完成这些操作
假如有许多客户请求连接，可以把接收客户连接的操作单独由一个线程完成，把接收数据和发送数据的操作由另一个线程完成，这可以提高服务器的并发性能
负责接收客户连接的线程按照阻塞模式工作，如果收到客户连接，就向 Selector 注册读就绪和写就绪事件，否则进入阻塞状态，直到接收到了客户的连接。负责接收数据和发送数据的线程按照非阻塞模式工作，只有在读就绪或写就绪事件发生时，才执行相应的接收数据和发送数据操作

1. public class EchoServer {
2.    
3.         private int port = 8000;
4.     private ServerSocketChannel serverSocketChannel = null;
5.     private Selector selector = null;
6.     private Charset charset = Charset.forName("GBK");
8.         public EchoServer() throws IOException {
9.         selector = Selector.open();
10.         serverSocketChannel = ServerSocketChannel.open();
11.         serverSocketChannel.socket().setReuseAddress(true);
12.         serverSocketChannelsocket().bind(new InetSocketAddress(port));
13.     }
14.    
15.     public void accept() {
16.         while(true) {
17.             try {
18.                 SocketChannel socketChannel = serverSocketChannel.accept();
19.                 socketChannel.configureBlocking(false);
20.                
21.                 ByteBuffer buffer = ByteBuffer.allocate(1024);
22.                 synchronized(gate) {
23.                     selector.wakeup();
24.                     socketChannel.register(selector, SelectionKey.OP_READ | SelectionKey.OP_WRITE, buffer);
25.                 }
26.             } catch(IOException e) {
27.                 e.printStackTrace();
28.             }
29.         }
30.     }
31.    
32.     private Object gate=new Object();
33.    
34.     public void service() throws IOException {
35.         while(true) {
36.             synchronized(gate){}
37.             int n = selector.select();
38.             if(n == 0) continue;
39.             Set readyKeys = selector.selectedKeys();
40.             Iterator it = readyKeys.iterator();
41.             while (it.hasNext()) {
42.                 SelectionKey key = null;
43.                 try {
44.                                     it.remove();
45.                     if (key.isReadable()) {
46.                         receive(key);
47.                     }
48.                     if (key.isWritable()) {
49.                         send(key);
50.                     }
51.                 } catch(IOException e) {
52.                     e.printStackTrace();
53.                     try {
54.                         if(key != null) {
55.                             key.cancel();
56.                             key.channel().close();
57.                         }
58.                     } catch(Exception ex) { e.printStackTrace(); }
59.                 }
60.             }
61.         }
62.     }
63.    
64.     public void receive(SelectionKey key) throws IOException {
65.         ...
66.     }
67.    
68.     public void send(SelectionKey key) throws IOException {
69.         ...
70.     }
71.    
72.     public String decode(ByteBuffer buffer) {
73.         ...
74.     }
75.    
76.     public ByteBuffer encode(String str) {
77.         ...
78.     }
79.    
80.     public static void main(String args[])throws Exception {
81.         final EchoServer server = new EchoServer();
82.         Thread accept = new Thread() {
83.             public void run() {
84.                 server.accept();
85.             }
86.         };
87.         accept.start();
88.                 server.service();
89.     }
90. }

复制代码

注意一点：主线程的 selector select() 方法和 Accept 线程的 register(...) 方法都会造成阻塞，因为他们都会操作 Selector 对象的共享资源 all-keys 集合，这有可能会导致死锁
导致死锁的具体情形是：Selector 中尚没有任何注册的事件，即 all-keys 集合为空，主线程执行 selector.select() 方法时将进入阻塞状态，只有当 Accept 线程向 Selector 注册了事件，并且该事件发生后，主线程才会从 selector.select() 方法返回。然而，由于主线程正在 selector.select() 方法中阻塞，这使得 Acccept 线程也在 register() 方法中阻塞。Accept 线程无法向 Selector 注册事件，而主线程没有任何事件可以监控，所以这两个线程将永远阻塞下去
为了避免对共享资源的竞争，同步机制使得一个线程执行 register() 时，不允许另一个线程同时执行 select() 方法，反之亦然


免责声明：如果侵犯了您的权益，请联系站长，我们会及时删除侵权内容，谢谢合作！