JavaSE7新特性异步非阻塞I/O 网络通信 AIO

深碍√TFBOYSˉ_ 2022-06-18 05:48 146阅读 0赞

Asynchronous I/O,异步I/O操作,以Proactor模式为原型设计.在nio中,当有事件发生时,我们会得到通知,然后再去相应的读和写,在aio中,当我们需要的事件完成时才会得到通知,之后可以直接进行业务处理.  
  
Server端

Java代码

1.  package aio;  
2.    
3.  import java.net.InetSocketAddress;  
4.  import java.nio.channels.AsynchronousChannelGroup;  
5.  import java.nio.channels.AsynchronousServerSocketChannel;  
6.  import java.nio.channels.AsynchronousSocketChannel;  
7.  import java.util.concurrent.ExecutorService;  
8.  import java.util.concurrent.Executors;  
9.  import java.util.concurrent.Future;  
10.   
11. public class AioTcpServer implements Runnable \{  
12.     private AsynchronousChannelGroup asyncChannelGroup;//aio的核心之一通道组.由它负责处理事件,完成之后通知相应的handler  
13.     private AsynchronousServerSocketChannel listener;//端口侦听器  
14.   
15.     public AioTcpServer(int port) throws Exception \{  
16.         ExecutorService executor = Executors.newFixedThreadPool(20);  
17.         asyncChannelGroup = AsynchronousChannelGroup.withThreadPool(executor);  
18.         listener = AsynchronousServerSocketChannel.open(asyncChannelGroup).bind(new InetSocketAddress(port));  
19.     \}  
20.   
21.     public void run() \{  
22.         try \{  
23.             Future<AsynchronousSocketChannel> future = listener.accept(listener, new AioAcceptHandler());  
24.             future.get();//此步为阻塞方法,直到有连接上来为止.  
25.         \} catch (InterruptedException e) \{  
26.             e.printStackTrace();  
27.         \} catch (Exception e) \{  
28.             e.printStackTrace();  
29.         \} finally \{  
30.   
31.         \}  
32.     \}  
33.   
34.     public static void main(String... args) throws Exception \{  
35.         AioTcpServer server = new AioTcpServer(9998);  
36.         new Thread(server).start();  
37.     \}  
38. \}

package aio;
    
    import java.net.InetSocketAddress;
    import java.nio.channels.AsynchronousChannelGroup;
    import java.nio.channels.AsynchronousServerSocketChannel;
    import java.nio.channels.AsynchronousSocketChannel;
    import java.util.concurrent.ExecutorService;
    import java.util.concurrent.Executors;
    import java.util.concurrent.Future;
    
    public class AioTcpServer implements Runnable {
     private AsynchronousChannelGroup asyncChannelGroup;//aio的核心之一通道组.由它负责处理事件,完成之后通知相应的handler
     private AsynchronousServerSocketChannel listener;//端口侦听器
    
     public AioTcpServer(int port) throws Exception {
      ExecutorService executor = Executors.newFixedThreadPool(20);
      asyncChannelGroup = AsynchronousChannelGroup.withThreadPool(executor);
      listener = AsynchronousServerSocketChannel.open(asyncChannelGroup).bind(new InetSocketAddress(port));
     }
    
     public void run() {
      try {
       Future<AsynchronousSocketChannel> future = listener.accept(listener, new AioAcceptHandler());
       future.get();//此步为阻塞方法,直到有连接上来为止.
      } catch (InterruptedException e) {
       e.printStackTrace();
      } catch (Exception e) {
       e.printStackTrace();
      } finally {
    
      }
     }
    
     public static void main(String... args) throws Exception {
      AioTcpServer server = new AioTcpServer(9998);
      new Thread(server).start();
     }
    }

Java代码

1.  package aio;  
2.    
3.  import java.io.IOException;  
4.  import java.nio.ByteBuffer;  
5.  import java.nio.channels.AsynchronousServerSocketChannel;  
6.  import java.nio.channels.AsynchronousSocketChannel;  
7.  import java.nio.channels.CompletionHandler;  
8.  import java.util.concurrent.ExecutionException;  
9.  import java.util.concurrent.Future;  
10.   
11. public class AioAcceptHandler implements CompletionHandler<AsynchronousSocketChannel, AsynchronousServerSocketChannel> \{  
12.     public void cancelled(AsynchronousServerSocketChannel attachment) \{  
13.         System.out.println("cancelled");  
14.     \}  
15.   
16.     public void completed(AsynchronousSocketChannel socket, AsynchronousServerSocketChannel attachment) \{  
17.         try \{  
18.             attachment.accept(attachment, this);//此方法有点递归的意思.目的是继续侦听端口,由channelGroup负责执行.  
19.             System.out.println("有客户端连接:" + socket.getRemoteAddress().toString());  
20.             startRead(socket);  
21.         \} catch (IOException e) \{  
22.             e.printStackTrace();  
23.         \}  
24.     \}  
25.   
26.     public void failed(Throwable exc, AsynchronousServerSocketChannel attachment) \{  
27.         exc.printStackTrace();  
28.     \}  
29.   
30.     public void startRead(AsynchronousSocketChannel socket) \{  
31.         ByteBuffer clientBuffer = ByteBuffer.allocate(1024);  
32.         Future<Integer> future = socket.read(clientBuffer, clientBuffer, new AioReadHandler(socket));  
33.         try \{  
34.             future.get();  
35.         \} catch (InterruptedException e) \{  
36.             e.printStackTrace();  
37.         \} catch (ExecutionException e) \{  
38.             e.printStackTrace();  
39.         \}  
40.     \}  
41. \}

package aio;
    
    import java.io.IOException;
    import java.nio.ByteBuffer;
    import java.nio.channels.AsynchronousServerSocketChannel;
    import java.nio.channels.AsynchronousSocketChannel;
    import java.nio.channels.CompletionHandler;
    import java.util.concurrent.ExecutionException;
    import java.util.concurrent.Future;
    
    public class AioAcceptHandler implements CompletionHandler<AsynchronousSocketChannel, AsynchronousServerSocketChannel> {
     public void cancelled(AsynchronousServerSocketChannel attachment) {
      System.out.println("cancelled");
     }
    
     public void completed(AsynchronousSocketChannel socket, AsynchronousServerSocketChannel attachment) {
      try {
       attachment.accept(attachment, this);//此方法有点递归的意思.目的是继续侦听端口,由channelGroup负责执行.
       System.out.println("有客户端连接:" + socket.getRemoteAddress().toString());
       startRead(socket);
      } catch (IOException e) {
       e.printStackTrace();
      }
     }
    
     public void failed(Throwable exc, AsynchronousServerSocketChannel attachment) {
      exc.printStackTrace();
     }
    
     public void startRead(AsynchronousSocketChannel socket) {
      ByteBuffer clientBuffer = ByteBuffer.allocate(1024);
      Future<Integer> future = socket.read(clientBuffer, clientBuffer, new AioReadHandler(socket));
      try {
       future.get();
      } catch (InterruptedException e) {
       e.printStackTrace();
      } catch (ExecutionException e) {
       e.printStackTrace();
      }
     }
    }

Java代码

1.  package aio;  
2.    
3.  import java.io.IOException;  
4.  import java.nio.ByteBuffer;  
5.  import java.nio.channels.AsynchronousSocketChannel;  
6.  import java.nio.channels.CompletionHandler;  
7.  import java.nio.charset.CharacterCodingException;  
8.  import java.nio.charset.Charset;  
9.  import java.nio.charset.CharsetDecoder;  
10.   
11. public class AioReadHandler implements CompletionHandler<Integer, ByteBuffer> \{  
12.     private AsynchronousSocketChannel socket;  
13.   
14.     public AioReadHandler(AsynchronousSocketChannel socket) \{  
15.         this.socket = socket;  
16.     \}  
17.   
18.     public void cancelled(ByteBuffer attachment) \{  
19.         System.out.println("cancelled");  
20.     \}  
21.   
22.     private CharsetDecoder decoder = Charset.forName("GBK").newDecoder();  
23.   
24.     public void completed(Integer i, ByteBuffer buf) \{  
25.         if (i > 0) \{  
26.             buf.flip();  
27.             try \{  
28.                 System.out.println("收到" + socket.getRemoteAddress().toString() + "的消息:" + decoder.decode(buf));  
29.                 buf.compact();  
30.             \} catch (CharacterCodingException e) \{  
31.                 e.printStackTrace();  
32.             \} catch (IOException e) \{  
33.                 e.printStackTrace();  
34.             \}  
35.             socket.read(buf, buf, this);  
36.         \} else if (i == -1) \{  
37.             try \{  
38.                 System.out.println("客户端断线:" + socket.getRemoteAddress().toString());  
39.                 buf = null;  
40.             \} catch (IOException e) \{  
41.                 e.printStackTrace();  
42.             \}  
43.         \}  
44.     \}  
45.   
46.     public void failed(Throwable exc, ByteBuffer buf) \{  
47.         System.out.println(exc);  
48.     \}  
49. \}
50.   
    
51. Client端  
    
    
    Java代码
52. package aio;  
53.   
54. import java.io.IOException;  
55. import java.net.InetSocketAddress;  
56. import java.nio.ByteBuffer;  
57. import java.nio.channels.AsynchronousChannelGroup;  
58. import java.nio.channels.AsynchronousSocketChannel;  
59. import java.nio.channels.CompletionHandler;  
60. import java.nio.charset.CharacterCodingException;  
61. import java.nio.charset.Charset;  
62. import java.nio.charset.CharsetDecoder;  
63. import java.util.Timer;  
64. import java.util.TimerTask;  
65. import java.util.concurrent.ExecutionException;  
66. import java.util.concurrent.ExecutorService;  
67. import java.util.concurrent.Executors;  
68. import java.util.concurrent.Future;  
69.   
70. public class AioTcpConnector \{  
71.     private AsynchronousChannelGroup asyncChannelGroup;  
72.     private AsynchronousSocketChannel connector;  
73.   
74.     public AioTcpConnector() throws Exception \{  
75.         ExecutorService executor = Executors.newFixedThreadPool(20);  
76.         asyncChannelGroup = AsynchronousChannelGroup.withThreadPool(executor);  
77.     \}  
78.   
79.     private final CharsetDecoder decoder = Charset.forName("GBK").newDecoder();  
80.   
81.     public void start(final String ip, final int port) throws Exception \{  
82.         Timer timer = new Timer();  
83.         timer.scheduleAtFixedRate(new TimerTask() \{  
84.             @Override  
85.             public void run() \{  
86.                 try \{  
87.                     if (connector == null || !connector.isOpen()) \{  
88.                         connector = AsynchronousSocketChannel.open(asyncChannelGroup);  
89.                         //connector.setOption(StandardSocketOption.TCP\_NODELAY, true);  
90.                         //connector.setOption(StandardSocketOption.SO\_REUSEADDR, true);  
91.                         //connector.setOption(StandardSocketOption.SO\_KEEPALIVE, true);  
92.                         Future<Void> future = connector.connect(new InetSocketAddress(ip, port));  
93.                         try \{  
94.                             future.get();  
95.                             handlerRead();  
96.                         \} catch (InterruptedException e) \{  
97.                             e.printStackTrace();  
98.                         \} catch (ExecutionException e) \{  
99.                             System.out.println("尝试连接失败!");  
100.                         \}  
101.                     \}  
102.                 \} catch (IOException e) \{  
103.                     e.printStackTrace();  
104.                 \}  
105.             \}  
106.         \}, 1, 10 \* 1000);  
107.     \}  
108.   
109.     public void handlerRead() throws InterruptedException, ExecutionException \{  
110.         try \{  
111.             System.out.println("与服务器连接成功:" + connector.getRemoteAddress());  
112.         \} catch (IOException e1) \{  
113.             e1.printStackTrace();  
114.         \}  
115.         final ByteBuffer buf = ByteBuffer.allocate(1024);  
116.         Future<Integer> rows = connector.read(buf, buf, new CompletionHandler<Integer, ByteBuffer>() \{  
117.             public void cancelled(ByteBuffer attachment) \{  
118.                 System.out.println("cancelled");  
119.             \}  
120.   
121.             public void completed(Integer i, ByteBuffer in) \{  
122.                 if (i > 0) \{  
123.                     in.flip();  
124.                     try \{  
125.                         System.out.println("收到" + connector.getRemoteAddress().toString() + "的消息:" + decoder.decode(in));  
126.                         in.compact();  
127.                     \} catch (CharacterCodingException e) \{  
128.                         e.printStackTrace();  
129.                     \} catch (IOException e) \{  
130.                         e.printStackTrace();  
131.                     \}  
132.                     connector.read(in, in, this);  
133.                 \} else if (i == -1) \{  
134.                     try \{  
135.                         System.out.println("与服务器连接断线:" + connector.getRemoteAddress());  
136.                         connector.close();  
137.                     \} catch (IOException e) \{  
138.                         e.printStackTrace();  
139.                     \}  
140.                     in = null;  
141.                 \} else if (i == 0) \{  
142.                     System.out.println(i);  
143.                 \}  
144.             \}  
145.   
146.             public void failed(Throwable exc, ByteBuffer buf) \{  
147.                 System.out.println(exc);  
148.             \}  
149.         \});  
150.         rows.get();  
151.     \}  
152.   
153.     public static void main(String... args) throws Exception \{  
154.         AioTcpConnector client = new AioTcpConnector();  
155.         client.start("192.168.1.30", 9998);  
156.     \}  
157. \}