Java AIO 应用实例

项目地址：https://github.com/windwant/aio-test

• Server：

   1 package org.windwant.aio;
   2
   3 import java.io.IOException;
   4 import java.net.InetSocketAddress;
   5 import java.nio.ByteBuffer;
   6 import java.nio.channels.AsynchronousChannelGroup;
   7 import java.nio.channels.AsynchronousServerSocketChannel;
   8 import java.nio.channels.AsynchronousSocketChannel;
   9 import java.nio.channels.CompletionHandler;
  10 import java.nio.charset.Charset;
  11 import java.util.concurrent.ExecutionException;
  12 import java.util.concurrent.Executors;
  13
  14 /**
  15  * AsynchronousServerSocketChannel
  16  */
  17 public class AIOServer implements Runnable{
  18
  19     private int port = 8889;
  20     private int threadSize = 10;
  21     protected AsynchronousChannelGroup asynchronousChannelGroup;
  22
  23     protected AsynchronousServerSocketChannel serverChannel;
  24
  25     public AIOServer(int port, int threadSize) {
  26         this.port = port;
  27         this.threadSize = threadSize;
  28     }
  29
  30     public static void main(String[] args) throws IOException {
  31         new Thread(new AIOServer(8989, 19)).start();
  32     }
  33
  34     public void run() {
  35         try{
  36             asynchronousChannelGroup = AsynchronousChannelGroup.withCachedThreadPool(Executors.newCachedThreadPool(), 10);
  37             serverChannel = AsynchronousServerSocketChannel.open(asynchronousChannelGroup);
  38             serverChannel.bind(new InetSocketAddress(port));
  39             System.out.println("listening on port: " + port);
  40             serverChannel.accept(this, new CompletionHandler<AsynchronousSocketChannel, AIOServer>() {
  41                 final ByteBuffer echoBuffer = ByteBuffer.allocateDirect(1024);
  42
  43                 public void completed(AsynchronousSocketChannel result, AIOServer attachment) {
  44                     System.out.println("reading begin...");
  45                     try {
  46                         System.out.println("远程地址：" + result.getRemoteAddress());
  47                         echoBuffer.clear();
  48                         result.read(echoBuffer).get();
  49                         echoBuffer.flip();
  50                         System.out.println("received : " + Charset.defaultCharset().decode(echoBuffer));
  51                         String msg = "server test msg-" + Math.random();
  52                         System.out.println("server send data: " + msg);
  53                         result.write(ByteBuffer.wrap(msg.getBytes()));
  54                     } catch (IOException e) {
  55                         e.printStackTrace();
  56                     } catch (InterruptedException e) {
  57                         e.printStackTrace();
  58                     } catch (ExecutionException e) {
  59                         e.printStackTrace();
  60                     } finally {
  61                         attachment.serverChannel.accept(attachment, this);// 监听新的请求，递归调用。
  62                     }
  63
  64                 }
  65
  66                 public void failed(Throwable exc, AIOServer attachment) {
  67                     System.out.println("received failed");
  68                     exc.printStackTrace();
  69                     attachment.serverChannel.accept(attachment, this);
  70                 }
  71             });
  72             System.in.read();
  73
  74         }catch (Exception e){
  75             e.printStackTrace();
  76         }
  77     }
  78 }

• Client：

   1 package org.windwant.aio;
   2
   3 import java.io.IOException;
   4 import java.net.InetSocketAddress;
   5 import java.nio.ByteBuffer;
   6 import java.nio.channels.AsynchronousSocketChannel;
   7 import java.nio.channels.CompletionHandler;
   8
   9 /**
  10  * AsynchronousSocketChannel
  11  */
  12 public class AIOClient implements Runnable{
  13
  14     private AsynchronousSocketChannel client;
  15     private String host;
  16     private int port;
  17     public AIOClient(String host, int port) throws IOException {
  18         this.client = AsynchronousSocketChannel.open();
  19         this.host = host;
  20         this.port = port;
  21     }
  22
  23     public static void main(String[] args) {
  24         try {
  25             new Thread(new AIOClient("127.0.0.1", 8989)).start();
  26             System.in.read();
  27         } catch (IOException e) {
  28             e.printStackTrace();
  29         }
  30
  31     }
  32
  33     public void run() {
  34         client.connect(new InetSocketAddress(host, port), null, new CompletionHandler<Void, Object>() {
  35             public void completed(Void result, Object attachment) {
  36                     String msg = "client test msg-" + Math.random();
  37                     client.write(ByteBuffer.wrap(msg.getBytes()));
  38                     System.out.println("client send data:" + msg);
  39             }
  40
  41             public void failed(Throwable exc, Object attachment) {
  42                 System.out.println("client send field...");
  43             }
  44         });
  45
  46         final ByteBuffer byteBuffer = ByteBuffer.allocate(1024);
  47         client.read(byteBuffer, this, new CompletionHandler<Integer, Object>() {
  48             public void completed(Integer result, Object attachment) {
  49                 System.out.println(result);
  50                 System.out.println("client read data: " + new String(byteBuffer.array()));
  51             }
  52
  53             public void failed(Throwable exc, Object attachment) {
  54                 System.out.println("read faield");
  55             }
  56         });
  57     }
  58 }