码迷，mamicode.com
首页 > 编程语言 > 详细

Java中的BIO、NIO、AIO-3

时间：2018-06-26 00:48:20      阅读：222      评论：0      收藏：0      [点我收藏+]

标签：文件操作   fun   public   物理链路   users   none   cell   except   存在   

Java中的BIO、NIO、AIO-3

java 

这一篇是代码篇，敲代码有助于理解记忆这些抽象的东西：

参考资料：

目录

Java BIO代码

服务器

  1. package sock; 
  2.  
  3. import java.io.BufferedReader; 
  4. import java.io.IOException; 
  5. import java.io.InputStreamReader; 
  6. import java.io.PrintWriter; 
  7. import java.net.ServerSocket; 
  8. import java.net.Socket; 
  9. import java.util.ArrayList; 
  10. import java.util.LinkedList; 
  11. import java.util.List; 
  12.  
  13.  
  14.  
  15. /* 
  16. 存在多线程对数据结构并发操作不安全的问题 
  17. */ 
  18. public class socketServerT extends ServerSocket{ 
  19. private static final int port = 2018; 
  20. private static boolean isPrint = false; 
  21. private static List<String> user_list = new ArrayList(); 
  22. private static List<ServerThread> threadlist = new ArrayList<>();//应该使用线程安全的集合 
  23. private static LinkedList<String> message = new LinkedList<>(); 
  24.  
  25.  
  26. socketServerT() throws IOException{ 
  27. super(port); 
  28. new PrintOutThread(); 
  29. System.out.println( " server is created"); 
  30. try{ 
  31. while(true){ 
  32. Socket sock = this.accept(); 
  33. new ServerThread(sock); 
  34. } 
  35. }catch( Exception e){ 
  36. e.printStackTrace(); 
  37. } 
  38. } 
  39. class PrintOutThread extends Thread{ 
  40. PrintOutThread(){ 
  41. System.out.println(getName() + "!!!!!"); 
  42. start(); 
  43. } 
  44. public void run(){ 
  45. while(true){ 
  46. if(isPrint){ 
  47. String m = message.getFirst(); 
  48. for(ServerThread i : threadlist){ 
  49. sendMessage(i,m); 
  50. } 
  51. message.removeFirst(); 
  52. isPrint = message.size()>0 ? true:false; 
  53. } 
  54. } 
  55. } 
  56. } 
  57. class ServerThread extends Thread{ 
  58. private BufferedReader rec; 
  59. private PrintWriter send; 
  60. private Socket client; 
  61. private String name; 
  62.  
  63. ServerThread(Socket sock) throws IOException{ 
  64. client = sock; 
  65. rec = new BufferedReader(new InputStreamReader(client.getInputStream())); 
  66. send = new PrintWriter(client.getOutputStream(),true); 
  67. //rec.readLine(); 
  68. System.out.println(getName() + "is created"); 
  69. send.println("connected to chat room, please input your name!!"); 
  70. start(); 
  71. } 
  72. public PrintWriter getSend(){ 
  73. return send; 
  74. } 
  75. public void run(){ 
  76. try{ 
  77. int flag = 0; 
  78. String line = ""; 
  79. while(!line.contains("bye")){ 
  80. line = rec.readLine(); 
  81. if("showuser".equals(line)){ 
  82. send.println(listOneUsers()); 
  83. //line = rec.readLine(); 
  84. continue; 
  85. } 
  86. if(flag == 0){ 
  87. flag ++; 
  88. name = line; 
  89. user_list.add(name); 
  90. threadlist.add(this); 
  91. send.println(name + " begin to chat"); 
  92. pushMessage("client <" + name+"> enter chat room"); 
  93. }else{ 
  94. pushMessage("client <" + name+"> say :" + line); 
  95. } 
  96. //line = rec.readLine(); 
  97. } 
  98.  
  99. }catch (Exception e){ 
  100. e.printStackTrace(); 
  101. }finally { 
  102. try{ 
  103. client.close(); 
  104. rec.close(); 
  105. send.close(); 
  106. }catch (IOException e){ 
  107. e.printStackTrace(); 
  108. } 
  109. threadlist.remove(this); 
  110. user_list.remove(name); 
  111. pushMessage("client <" + name+"> exit"); 
  112. } 
  113.  
  114. } 
  115. } 
  116. public void pushMessage(String mess){ 
  117. message.add(mess); 
  118. isPrint = true; 
  119. } 
  120. public String listOneUsers(){ 
  121. StringBuffer s = new StringBuffer(); 
  122. s.append("---online users---\n"); 
  123. for( String i:user_list){ 
  124. s.append(i + "\n"); 
  125. } 
  126. s.append("---end---\n"); 
  127. return s.toString(); 
  128. } 
  129. public void sendMessage(ServerThread s,String m){ 
  130. //System.out.println("test"); 
  131. PrintWriter p = s.getSend(); 
  132. p.println(m); 
  133. //p.flush(); 
  134. } 
  135. public static void main(String args[]){ 
  136. try{ 
  137. socketServerT s = new socketServerT(); 
  138. }catch(Exception e){ 
  139. e.printStackTrace(); 
  140. } 
  141. } 
  142. } 
  143.  

客户端

  1. package sock; 
  2.  
  3. import java.io.BufferedReader; 
  4. import java.io.IOException; 
  5. import java.io.InputStreamReader; 
  6. import java.io.PrintWriter; 
  7. import java.net.Socket; 
  8.  
  9. public class socketClientT extends Socket{ 
  10. private static final String server = "127.0.0.1"; 
  11. private static final int port = 2018; 
  12.  
  13. private Socket sock; 
  14. private PrintWriter send; 
  15. private BufferedReader rec; 
  16. socketClientT() throws IOException{ 
  17. super(server,port); 
  18. sock = this; 
  19. send = new PrintWriter(sock.getOutputStream(),true); 
  20. rec = new BufferedReader(new InputStreamReader(sock.getInputStream())); 
  21. Thread t = new recvThread(); 
  22. BufferedReader sysBuff = new BufferedReader(new InputStreamReader(System.in)); 
  23. String line = ""; 
  24. while(! line.contains("bye")){ 
  25. line = sysBuff.readLine(); 
  26. send.println(line); 
  27. } 
  28. send.close(); 
  29. rec.close(); 
  30. sysBuff.close(); 
  31. this.close(); 
  32. } 
  33. class recvThread extends Thread{ 
  34. private BufferedReader buff; 
  35. recvThread(){ 
  36. try { 
  37. buff = new BufferedReader(new InputStreamReader(sock.getInputStream())); 
  38. start(); 
  39. } catch (Exception e){ 
  40. e.printStackTrace(); 
  41. } 
  42. } 
  43. public void run(){ 
  44. String res = ""; 
  45. try{ 
  46. while(true){ 
  47. res = buff.readLine(); 
  48. if(res.contains("bye")) 
  49. break; 
  50. System.out.println(res); 
  51. } 
  52. send.close(); 
  53. buff.close(); 
  54. sock.close(); 
  55. }catch (Exception e){ 
  56. e.printStackTrace(); 
  57. } 
  58. } 
  59. } 
  60.  
  61. public static void main(String args[]){ 
  62. try { 
  63. socketClientT s = new socketClientT(); 
  64. }catch (Exception e){ 
  65. e.printStackTrace(); 
  66. } 
  67. } 
  68. } 
  69.  

Java NIO

JDK 1.4的java.util.*;包中引入了新的Java I/O库，其目的是提高IO操作的速度。

简介

NIO我们一般认为是New I/O（也是官方的叫法），因为它是相对于老的I/O类库新增的（其实在JDK 1.4中就已经被引入了，但这个名词还会继续用很久，即使它们在现在看来已经是“旧”的了，所以也提示我们在命名时，需要好好考虑），做了很大的改变。但民间跟多人称之为Non-block I/O，即非阻塞I/O，因为这样叫，更能体现它的特点。而下文中的NIO，不是指整个新的I/O库，而是非阻塞I/O。
    NIO提供了与传统BIO模型中的Socket和ServerSocket相对应的SocketChannel和ServerSocketChannel两种不同的套接字通道实现。
    新增的着两种通道都支持阻塞和非阻塞两种模式。
阻塞模式使用就像传统中的支持一样，比较简单，但是性能和可靠性都不好；非阻塞模式正好与之相反。
    对于低负载、低并发的应用程序，可以使用同步阻塞I/O来提升开发速率和更好的维护性；对于高负载、高并发的（网络）应用，应使用NIO的非阻塞模式来开发。
    下面会先对基础知识进行介绍。

缓冲区Buffer

Buffer是一个对象，包含一些要写入或者读出的数据。
在NIO库中，所有数据都是用缓冲区处理的。在读取数据时，它是直接读到缓冲区中的；在写入数据时，也是写入到缓冲区中。任何时候访问NIO中的数据，都是通过缓冲区进行操作。
缓冲区实际上是一个数组，并提供了对数据结构化访问以及维护读写位置等信息。
具体的缓存区有这些：ByteBuffe、CharBuffer、 ShortBuffer、IntBuffer、LongBuffer、FloatBuffer、DoubleBuffer。他们实现了相同的接口：Buffer。

通道Channel

我们对数据的读取和写入要通过Channel，它就像水管一样，是一个通道。通道不同于流的地方就是通道是双向的，可以用于读、写和同时读写操作。
底层的操作系统的通道一般都是全双工的，所以全双工的Channel比流能更好的映射底层操作系统的API。
Channel主要分两大类：

  • SelectableChannel：用户网络读写
  • FileChannel：用于文件操作

后面代码会涉及的ServerSocketChannel和SocketChannel都是SelectableChannel的子类。

多路复用器

Selector是Java  NIO 编程的基础。
Selector提供选择已经就绪的任务的能力：Selector会不断轮询注册在其上的Channel，如果某个Channel上面发生读或者写事件，这个Channel就处于就绪状态，会被Selector轮询出来，然后通过SelectionKey可以获取就绪Channel的集合，进行后续的I/O操作。
一个Selector可以同时轮询多个Channel，因为JDK使用了epoll()代替传统的select实现，所以没有最大连接句柄1024/2048的限制。所以，只需要一个线程负责Selector的轮询，就可以接入成千上万的客户端。

服务器端代码

server：

  1. /** 
  2. * server 
  3. */ 
  4. public class server { 
  5.  
  6. private static int port = 8000; 
  7. private static serverHandle sHandle; 
  8.  
  9. public static void start() { 
  10. start(port); 
  11. } 
  12. private static synchronized void start(int port) { 
  13. if (sHandle != null) { 
  14. sHandle.setStarted(false); 
  15. } 
  16. sHandle = new serverHandle(port); 
  17. new Thread(sHandle,"server").start(); 
  18.  
  19. } 
  20. public static void main(String[] args) { 
  21. start(); 
  22. } 
  23. } 

serverHandle

  1. import java.io.IOException; 
  2. import java.net.InetSocketAddress; 
  3. import java.net.ServerSocket; 
  4. import java.nio.ByteBuffer; 
  5. import java.nio.channels.SelectionKey; 
  6. import java.nio.channels.Selector; 
  7. import java.nio.channels.ServerSocketChannel; 
  8. import java.nio.channels.SocketChannel; 
  9. import java.util.Iterator; 
  10. import java.util.Set; 
  11.  
  12. //import jdk.internal.org.objectweb.asm.Handle; 
  13.  
  14. /** 
  15. * serverHandle 
  16. */ 
  17. public class serverHandle implements Runnable{ 
  18.  
  19. private ServerSocketChannel serverChannel; 
  20. private Selector selector; 
  21. private volatile boolean started;//各个线程都能看到状态 
  22.  
  23. public serverHandle(int port) { 
  24. try { 
  25. //创建选择器 
  26. selector = Selector.open(); 
  27. //创建serverSocketChannel 
  28. serverChannel = ServerSocketChannel.open(); 
  29. //如果为 true，则此通道将被置于阻塞模式；如果为 false，则此通道将被置于非阻塞模式 
  30. serverChannel.configureBlocking(false); 
  31. //创建InetSocketAddress 
  32. InetSocketAddress socketAddress = new InetSocketAddress(port); 
  33. //得到ServerSocket 
  34. ServerSocket serverSocket = serverChannel.socket(); 
  35. //绑定ServetSocket到一个具体的端口，并设置backlog 
  36. serverSocket.bind(socketAddress, 1024); 
  37. //向selector注册ServerSocketChannel，设置为监听客户端的连接请求 
  38. serverChannel.register(selector, SelectionKey.OP_ACCEPT); 
  39. //标记服务器状态 
  40. started = true; 
  41. System.out.println("服务器已经启动，端口号：" + port); 
  42.  
  43. } catch (IOException e) { 
  44. //TODO: handle exception 
  45. e.printStackTrace(); 
  46. System.exit(1); 
  47. }  
  48. } 
  49.  
  50. public void setStarted(boolean flag) { 
  51. this.started = flag; 
  52. } 
  53. @Override 
  54. public void run() { 
  55. //循环遍历selector 
  56. while (started) { 
  57. try { 
  58. //无论是否有读写事件，selector每个1s唤醒一次 
  59.  
  60. try { 
  61. selector.select(1000); 
  62. } catch (Exception e) { 
  63. e.printStackTrace(); 
  64. } 
  65. //获得状态为ready的selectorkey 
  66. Set<SelectionKey> keys = selector.selectedKeys(); 
  67. Iterator<SelectionKey> iter = keys.iterator(); 
  68. SelectionKey key = null; 
  69. while (iter.hasNext()) { 
  70. key = iter.next(); 
  71. iter.remove(); 
  72. try { 
  73. handle(key); 
  74. } catch (Exception e) { 
  75.  
  76. if (key != null) { 
  77. key.cancel(); 
  78. if (key.channel() != null) { 
  79. key.channel().close(); 
  80. } 
  81. } 
  82. } 
  83. } 
  84. } catch (Throwable t) { 
  85.  
  86. t.printStackTrace(); 
  87. } 
  88. } 
  89. //关闭selector 
  90. if (selector != null) { 
  91. try { 
  92. selector.close(); 
  93. } catch (Exception e) { 
  94.  
  95. e.printStackTrace(); 
  96. } 
  97. } 
  98. } 
  99. private void handle(SelectionKey key) throws IOException { 
  100. //判断kei是否是有效的 
  101. if (key.isValid()) { 
  102.  
  103. //处理新接入的请求消息， 
  104. if (key.isAcceptable()) { 
  105. //通过selectionkey得到ServerSocketChannel，注意转型 
  106. ServerSocketChannel serverSocketChannel = (ServerSocketChannel) key.channel(); 
  107. //通过serversocketchannel的accept方法创建SocketChannel实例 
  108. SocketChannel client = serverSocketChannel.accept(); 
  109. //设置client为非阻塞模式 
  110. client.configureBlocking(false); 
  111. //把client注册到selector,注册事件为读 
  112. client.register(selector, SelectionKey.OP_READ); 
  113. } 
  114.  
  115. //读消息 
  116. if (key.isReadable()) { 
  117. SocketChannel sc = (SocketChannel) key.channel(); 
  118. //创建byteBuffer，大小为1M 
  119. ByteBuffer buffer = ByteBuffer.allocate(1024); 
  120. //读取请求码流，返回读取到的字节数 
  121. int readBytes = sc.read(buffer); 
  122. //读取到字节，对字节进行编码 
  123. if (readBytes > 0) { 
  124. //将缓冲区buffer的position设为0，用于后续对缓冲区的读操作 
  125. buffer.flip(); 
  126. //根据缓冲区可读字节数创建字节数组 
  127. byte[] bytes = new byte[buffer.remaining()]; 
  128. //将缓冲区可读字节数组复制到新建的数组中 
  129. buffer.get(bytes); 
  130. String expresString = new String(bytes, "UTF-8"); 
  131. System.out.println("服务器收到的消息：" + expresString); 
  132.  
  133. //处理数据 
  134. String res = null; 
  135. res = new StringBuffer(expresString).reverse().toString(); 
  136. //写入返回消息 
  137. dowrite(sc,res); 
  138. } else if (readBytes < 0) { 
  139. //链路关闭释放资源 
  140. key.cancel(); 
  141. sc.close(); 
  142. } 
  143. } 
  144. } 
  145. } 
  146. private void dowrite(SocketChannel sc, String res) throws IOException { 
  147. //把字符串编码为字节数组 
  148. byte[] bytes = res.getBytes(); 
  149. //根据数组容量创建ByteBuffer 
  150. ByteBuffer wBuffer = ByteBuffer.allocate(bytes.length); 
  151. //把字节数组复制到buffer中 
  152. wBuffer.put(bytes); 
  153. //flip操作，更改position为0，方便后续的写操作从头开始 
  154. wBuffer.flip(); 
  155. //发送缓冲区的数据 
  156. sc.write(wBuffer); 
  157. } 
  158. } 

客户端代码

client

  1. import java.util.Scanner; 
  2.  
  3. /** 
  4. * clientChannel 
  5. */ 
  6. public class clientChannel { 
  7. private static String host = "127.0.0.1"; 
  8. private static int port = 8000; 
  9. private static clientHandle cHandle; 
  10.  
  11. public static void start() { 
  12. start(host,port); 
  13. } 
  14. public static synchronized void start(String host, int port) { 
  15. if (cHandle != null) { 
  16. cHandle.stop(); 
  17. } 
  18. cHandle = new clientHandle(host, port); 
  19. new Thread(cHandle, "client").start();; 
  20. } 
  21.  
  22. public static Boolean sendMsg(String msg) throws Exception { 
  23. if (msg.contains("q")) { 
  24. return false; 
  25. } 
  26. cHandle.sendMsg(msg); 
  27. return true; 
  28. } 
  29. public static void main(String[] args) { 
  30. try { 
  31. start(); 
  32. Scanner s = new Scanner(System.in); 
  33. String tmp; 
  34. while ((tmp = s.nextLine())!= null) { 
  35. sendMsg(tmp); 
  36. } 
  37. } catch (Exception e) { 
  38. //TODO: handle exception 
  39. e.printStackTrace(); 
  40. } 
  41. //start(); 
  42. } 
  43. } 

clientHandle

  1. import java.net.InetSocketAddress; 
  2. import java.nio.ByteBuffer; 
  3. import java.nio.channels.SelectionKey; 
  4. import java.nio.channels.Selector; 
  5. import java.nio.channels.SocketChannel; 
  6. import java.util.Iterator; 
  7. import java.util.Set; 
  8. import java.io.IOException; 
  9. /** 
  10. * clientHandle 
  11. */ 
  12. public class clientHandle implements Runnable{ 
  13.  
  14. private String host; 
  15. private int port; 
  16. private Selector selector; 
  17. private SocketChannel socketChannel; 
  18. private volatile boolean started; 
  19.  
  20. public clientHandle(String ip, int port) { 
  21. this.host = ip; 
  22. this.port = port; 
  23.  
  24. try { 
  25. //创建选择器 
  26. this.selector = Selector.open(); 
  27. //创建socketchannel 
  28. this.socketChannel = SocketChannel.open(); 
  29. //配置socketChannel为非阻塞模式 
  30. this.socketChannel.configureBlocking(false); 
  31. this.started = true; 
  32.  
  33. } catch (Exception e) { 
  34. //TODO: handle exception 
  35. e.printStackTrace(); 
  36. System.exit(1); 
  37. } 
  38.  
  39. } 
  40.  
  41. public void stop(){ 
  42. this.started = false; 
  43. } 
  44.  
  45. public void doConnection() throws Exception{ 
  46. InetSocketAddress address = new InetSocketAddress(this.host, this.port); 
  47. if (socketChannel.connect(address)) { 
  48. System.out.println("连接服务器成功！！！"); 
  49. } else { 
  50. System.out.println("未连接成功，下一轮继续！！！"); 
  51. //向selector注册socketChannel的连接操作 
  52. socketChannel.register(selector, SelectionKey.OP_CONNECT); 
  53. } 
  54. } 
  55.  
  56. public void handleInput(SelectionKey key) throws Exception{ 
  57. if (key.isValid()) { 
  58. //获取SeclectionKey对应的socketChannel 
  59. SocketChannel sc = (SocketChannel) key.channel(); 
  60. //测试key对应的socketChannel通道是否已完成或未能完成其套接连接操作 
  61. if (key.isConnectable()) { 
  62. //完成连接过程，返回true表示channel已经建立了连接，false表示建立连接失败 
  63. //当使用SocketChannel的connect()函数进行连接的时候，当处于非阻塞模式的情况下，可能连接不是立刻完成的，需要使用 
  64. //finidhConnect()来检查连接是否建立 
  65. if (sc.finishConnect()) { 
  66. } else { 
  67. System.exit(1); 
  68. }  
  69. } 
  70. //读消息，判断key对应的channel是否可读 
  71. if (key.isReadable()) { 
  72. //创建一个缓冲区，用来存读取的数据， 
  73. ByteBuffer buffer = ByteBuffer.allocate(1024); 
  74. //读取数据，返回读取的字节数 
  75. int readSize = sc.read(buffer); 
  76. //读取到字节，对字节进行编码 
  77. if (readSize > 0) { 
  78. //设置缓冲区的limit和position，方面后面读取数据 
  79. buffer.flip(); 
  80. //根据缓冲区的可读字节数创建字节数组 
  81. byte[] bytes = new byte[buffer.remaining()]; 
  82. //把缓冲区的内容复制到字节数组中去 
  83. buffer.get(bytes); 
  84. String res = new String(bytes, "UTF-8"); 
  85. System.out.println("客户端收到的数据为：" + res); 
  86. } else if (readSize < 0) { 
  87. //这种情况说明链路已经关闭，释放资源 
  88. key.cancel(); 
  89. sc.close(); 
  90. } 
  91. } 
  92. } 
  93. } 
  94.  
  95. public void doWrite(SocketChannel sc, String request) throws Exception { 
  96. //将数据转换为字节数组 
  97. byte[] bytes = request.getBytes(); 
  98. //创建字节缓冲区 
  99. ByteBuffer buffer = ByteBuffer.allocate(bytes.length); 
  100. //将字节数组放入字节缓冲区 
  101. buffer.put(bytes); 
  102. //flip操作，调整limit和position 
  103. buffer.flip(); 
  104. //将数据写入到channel中 
  105. sc.write(buffer); 
  106. } 
  107. public void sendMsg(String msg) throws Exception{ 
  108. //这里还没明白为什么要先注册读操作？需要注册读操作才能，知道读状态是否就绪，方便handelInput函数处理！！ 
  109. //但是还有一个疑问，什么时候使用OP_WRITE 
  110. socketChannel.register(selector, SelectionKey.OP_READ); 
  111. doWrite(socketChannel, msg); 
  112. } 
  113. @Override  
  114. public void run() {  
  115. try{  
  116. doConnection();  
  117. }catch(Exception e){  
  118. e.printStackTrace();  
  119. System.exit(1);  
  120. }  
  121. //循环遍历selector  
  122. while(started){  
  123. try{  
  124. //无论是否有读写事件发生，selector每隔1s被唤醒一次  
  125. selector.select(1000);  
  126. //阻塞,只有当至少一个注册的事件发生的时候才会继续.  
  127. // selector.select();  
  128. Set<SelectionKey> keys = selector.selectedKeys();  
  129. Iterator<SelectionKey> it = keys.iterator();  
  130. SelectionKey key = null;  
  131. while(it.hasNext()){  
  132. key = it.next();  
  133. it.remove();  
  134. try{  
  135. handleInput(key);  
  136. }catch(Exception e){  
  137. if(key != null){  
  138. key.cancel();  
  139. if(key.channel() != null){  
  140. key.channel().close();  
  141. }  
  142. }  
  143. }  
  144. }  
  145. }catch(Exception e){  
  146. e.printStackTrace();  
  147. System.exit(1);  
  148. }  
  149. }  
  150. //selector关闭后会自动释放里面管理的资源  
  151. if(selector != null)  
  152. try{  
  153. selector.close();  
  154. }catch (Exception e) {  
  155. e.printStackTrace();  
  156. }  
  157. } 
  158. } 

##测试及解析
测试代码：

  1. import java.util.Scanner; 
  2.  
  3. /** 
  4. * Test 
  5. */ 
  6. public class Test { 
  7.  
  8. public static void main(String[] args) { 
  9. server.start(); 
  10. try { 
  11. Thread.sleep(3000); 
  12. } catch (Exception e) { 
  13. //TODO: handle exception 
  14. e.printStackTrace(); 
  15. } 
  16. clientChannel.start(); 
  17. Scanner s = new Scanner(System.in); 
  18. String tmp; 
  19. try { 
  20. while ((tmp = s.nextLine()) != null) { 
  21. clientChannel.sendMsg(tmp); 
  22. } 
  23. } catch (Exception e) { 
  24. //TODO: handle exception 
  25. e.printStackTrace(); 
  26. } 
  27.  
  28. } 
  29.  
  30. } 

可以看到，创建NIO服务端的主要步骤如下：

  1. 打开ServerSocketChannel，监听客户端连接
  2. 绑定监听端口，设置连接为非阻塞模式
  3. 创建Reactor线程，创建多路复用器并启动线程
  4. 将ServerSocketChannel注册到Reactor线程中的Selector上，监听ACCEPT事件
  5. Selector轮询准备就绪的key
  6. Selector监听到新的客户端接入，处理新的接入请求，完成TCP三次握手，简历物理链路
  7. 设置客户端链路为非阻塞模式
  8. 将新接入的客户端连接注册到Reactor线程的Selector上，监听读操作，读取客户端发送的网络消息
  9. 异步读取客户端消息到缓冲区
  10. 对Buffer编解码，处理半包消息，将解码成功的消息封装成Task
  11. 将应答消息编码为Buffer，调用SocketChannel的write将消息异步发送给客户端

Java AIO

Java nio 2.0的主要改进就是引入了异步IO（包括文件和网络），这里主要介绍下异步网络IO API的使用以及框架的设计，以TCP服务端为例。首先看下为了支持AIO引入的新的类和接口：

** java.nio.channels.AsynchronousChannel**
       标记一个channel支持异步IO操作。

** java.nio.channels.AsynchronousServerSocketChannel**
       ServerSocket的aio版本，创建TCP服务端，绑定地址，监听端口等。

** java.nio.channels.AsynchronousSocketChannel**
       面向流的异步socket channel，表示一个连接。

** java.nio.channels.AsynchronousChannelGroup**
       异步channel的分组管理，目的是为了资源共享。一个AsynchronousChannelGroup绑定一个线程池，这个线程池执行两个任务：处理IO事件和派发CompletionHandler。AsynchronousServerSocketChannel创建的时候可以传入一个AsynchronousChannelGroup，那么通过AsynchronousServerSocketChannel创建的AsynchronousSocketChannel将同属于一个组，共享资。

** java.nio.channels.CompletionHandler**
       异步IO操作结果的回调接口，用于定义在IO操作完成后所作的回调工作。AIO的API允许两种方式来处理异步操作的结果：返回的Future模式或者注册CompletionHandler，推荐用CompletionHandler的方式，这些handler的调用是由AsynchronousChannelGroup的线程池派发的。显然，线程池的大小是性能的关键因素。AsynchronousChannelGroup允许绑定不同的线程池，通过三个静态方法来创建：

 public static AsynchronousChannelGroup withFixedThreadPool(int nThreads, ThreadFactory threadFactory) throws IOException

 public static AsynchronousChannelGroup withCachedThreadPool(ExecutorService executor, int initialSize)

 public static AsynchronousChannelGroup withThreadPool(ExecutorService executor) throws IOException

需要根据具体应用相应调整，从框架角度出发，需要暴露这样的配置选项给用户。

在介绍完了aio引入的TCP的主要接口和类之后，我们来设想下一个aio框架应该怎么设计。参考非阻塞nio框架的设计，一般都是采用Reactor模式，Reactor负责事件的注册、select、事件的派发；相应地，异步IO有个Proactor模式，Proactor负责CompletionHandler的派发，查看一个典型的IO写操作的流程来看两者的区别：

Reactor:  send(msg) -> 消息队列是否为空，如果为空  -> 向Reactor注册OP_WRITE，然后返回 -> Reactor select -> 触发Writable，通知用户线程去处理 ->先注销Writable(很多人遇到的cpu 100%的问题就在于没有注销）,处理Writeable，如果没有完全写入，继续注册OP_WRITE。注意到，写入的工作还是用户线程在处理。

Proactor: send(msg) -> 消息队列是否为空，如果为空,发起read异步调用，并注册CompletionHandler，然后返回。 -> 操作系统负责将你的消息写入，并返回结果（写入的字节数）给Proactor -> Proactor派发CompletionHandler。可见，写入的工作是操作系统在处理，无需用户线程参与。事实上在aio的API中,AsynchronousChannelGroup就扮演了Proactor的角色。

CompletionHandler有三个方法，分别对应于处理成功、失败、被取消（通过返回的Future)情况下的回调处理：

public interface CompletionHandler<V,A> {

     void completed(V result, A attachment);

    void failed(Throwable exc, A attachment);

    void cancelled(A attachment);
}

其中的泛型参数V表示IO调用的结果，而A是发起调用时传入的attchment。

server端代码

server

  1. //package aio; 
  2.  
  3. public class Server { 
  4. public static int clientCount = 0; 
  5. public static int port = 8000; 
  6. public static String hoString = "127.0.0.1"; 
  7.  
  8. public static void start() { 
  9. start(Server.port); 
  10. } 
  11. public static void start(int port) { 
  12. AsyncServerHandler serverHandler = new AsyncServerHandler(port); 
  13. Thread t1 = new Thread(serverHandler); 
  14. t1.start(); 
  15. } 
  16. public static void main(String[] args) { 
  17. start(); 
  18. } 
  19. } 
  20.  

AsyncServerHandler

  1. //package aio; 
  2.  
  3. import java.io.IOException; 
  4. import java.net.InetSocketAddress; 
  5. import java.nio.channels.AsynchronousServerSocketChannel; 
  6. import java.nio.channels.CompletionHandler; 
  7. import java.util.concurrent.CountDownLatch; 
  8.  
  9. public class AsyncServerHandler implements Runnable{ 
  10. private AsynchronousServerSocketChannel serverSocketChannel; 
  11. private CountDownLatch latch; 
  12. public AsyncServerHandler(int port) { 
  13. // TODO Auto-generated constructor stub 
  14. InetSocketAddress address = new InetSocketAddress(port); 
  15. try { 
  16. serverSocketChannel = AsynchronousServerSocketChannel.open(); 
  17. serverSocketChannel.bind(address); 
  18. } catch (IOException e) { 
  19. // TODO Auto-generated catch block 
  20. e.printStackTrace(); 
  21. } 
  22. System.out.println("服务器已经启动，端口号：" + port); 
  23. } 
  24.  
  25. @Override 
  26. public void run() { 
  27. // TODO Auto-generated method stub 
  28. //CountDownLatch初始化  
  29. //它的作用：在完成一组正在执行的操作之前，允许当前的现场一直阻塞  
  30. //此处，让现场在此阻塞，防止服务端执行完成后退出  
  31. //也可以使用while(true)+sleep  
  32. //生成环境就不需要担心这个问题，以为服务端是不会退出的  
  33. this.latch = new CountDownLatch(1); 
  34. serverSocketChannel.accept(this, new AcceptHandler(this.latch)); 
  35.  
  36. try { 
  37. latch.await(); 
  38. } catch (InterruptedException e) { 
  39. // TODO Auto-generated catch block 
  40. e.printStackTrace(); 
  41. } 
  42. } 
  43.  
  44. public AsynchronousServerSocketChannel getServerSocketChannel() { 
  45. return serverSocketChannel; 
  46. } 
  47.  
  48. public void setServerSocketChannel(AsynchronousServerSocketChannel serverSocketChannel) { 
  49. this.serverSocketChannel = serverSocketChannel; 
  50. } 
  51.  
  52. public CountDownLatch getLatch() { 
  53. return latch; 
  54. } 
  55.  
  56. public void setLatch(CountDownLatch latch) { 
  57. this.latch = latch; 
  58. } 
  59.  
  60. } 
  61.  

AcceptHandler

  1. //package aio; 
  2.  
  3. import java.nio.ByteBuffer; 
  4. import java.nio.channels.AsynchronousServerSocketChannel; 
  5. import java.nio.channels.AsynchronousSocketChannel; 
  6. import java.nio.channels.CompletionHandler; 
  7. import java.util.concurrent.CountDownLatch; 
  8.  
  9. public class AcceptHandler implements CompletionHandler<AsynchronousSocketChannel,AsyncServerHandler> { 
  10.  
  11. private CountDownLatch latch; 
  12. public AcceptHandler(CountDownLatch latch) { 
  13. // TODO Auto-generated constructor stub 
  14. this.latch = latch; 
  15. } 
  16. @Override 
  17. public void completed(AsynchronousSocketChannel socketChannel, AsyncServerHandler serverHandler) { 
  18. // TODO Auto-generated method stub 
  19. //进入这个函数说明说明事件处理成功，已经成功的拿到socketChanne 
  20. Server.clientCount ++; 
  21. System.out.println("当前连接的客户数：" + Server.clientCount); 
  22. //继续接受其他客户机的连接， 
  23. AsynchronousServerSocketChannel channel = serverHandler.getServerSocketChannel(); 
  24. channel.accept(serverHandler, this); 
  25. //创建新的buffer,为读取数据做准备 
  26. ByteBuffer buffer = ByteBuffer.allocate(1024); 
  27. socketChannel.read(buffer, buffer, new serverReadHandler(socketChannel)); 
  28.  
  29. } 
  30.  
  31. @Override 
  32. public void failed(Throwable exc, AsyncServerHandler attachment) { 
  33. // TODO Auto-generated method stub 
  34. exc.printStackTrace(); 
  35. this.latch.countDown(); 
  36. } 
  37.  
  38.  
  39. } 
  40.  

serverReadHandler

  1. //package aio; 
  2.  
  3. import java.io.IOException; 
  4. import java.io.UnsupportedEncodingException; 
  5. import java.nio.ByteBuffer; 
  6. import java.nio.channels.AsynchronousSocketChannel; 
  7. import java.nio.channels.CompletionHandler; 
  8.  
  9. public class serverReadHandler implements CompletionHandler<Integer,ByteBuffer> { 
  10. //用于读取半包消息和应答消息 
  11. private AsynchronousSocketChannel serverChannel; 
  12. public serverReadHandler(AsynchronousSocketChannel channel) { 
  13. // TODO Auto-generated constructor stub 
  14. this.serverChannel = channel; 
  15. } 
  16.  
  17.  
  18. @Override 
  19. public void completed(Integer result, ByteBuffer buffer) { 
  20. // TODO Auto-generated method stub 
  21. //操作系统读取IO就绪之后，进入这个函数 
  22. //调整limit和position关系，方便读取 
  23. //buffer.flip(); 
  24. if (buffer.hasRemaining()) { 
  25. byte[] bytes = new byte[buffer.remaining()]; 
  26. buffer.get(bytes); 
  27. String msg = null; 
  28. try { 
  29. msg = new String(bytes, "UTF-8"); 
  30. } catch (UnsupportedEncodingException e) { 
  31. // TODO Auto-generated catch block 
  32. e.printStackTrace(); 
  33. } 
  34. System.out.println("服务器收到消息：" + msg); 
  35.  
  36. String calResult = null; 
  37. StringBuffer stringBuffer = new StringBuffer(msg); 
  38. calResult = stringBuffer.reverse().toString(); 
  39. //向客户端发送结果 
  40. byte[] resultBytes = calResult.getBytes(); 
  41. ByteBuffer rBuffer = ByteBuffer.allocate(resultBytes.length); 
  42. rBuffer.put(resultBytes); 
  43. this.serverChannel.write(rBuffer, rBuffer, new ServerWriteHandler(this.serverChannel)); 
  44. }else { 
  45. System.out.println("服务器没有读取到数据"); 
  46. } 
  47. } 
  48.  
  49. @Override 
  50. public void failed(Throwable exc, ByteBuffer buffer) { 
  51. // TODO Auto-generated method stub 
  52. try { 
  53. this.serverChannel.close(); 
  54. System.out.println("服务器socket关闭~~~"); 
  55. } catch (IOException e) { 
  56. // TODO Auto-generated catch block 
  57. e.printStackTrace(); 
  58. } 
  59. } 
  60.  
  61. } 
  62.  

serverWriteHandler

  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.  
  8. public class ServerWriteHandler implements CompletionHandler<Integer,ByteBuffer> { 
  9. private AsynchronousSocketChannel serverChannel; 
  10. public ServerWriteHandler(AsynchronousSocketChannel channel) { 
  11. // TODO Auto-generated constructor stub  
  12. this.serverChannel = channel; 
  13. } 
  14. @Override 
  15. public void completed(Integer result, ByteBuffer buffer) { 
  16. // TODO Auto-generated method stub 
  17. //调整limit和position的位置，方便下面输出使用 
  18. //buffer.flip(); 
  19. if (buffer.hasRemaining()) { 
  20. System.out.println("服务器输出数据~~~"); 
  21. buffer.clear(); 
  22. //向客户端写入数据 
  23. this.serverChannel.write(buffer, buffer, this); 
  24. } else { 
  25. //读取数据 
  26. ByteBuffer readBuffer = ByteBuffer.allocate(1024); 
  27. this.serverChannel.read(readBuffer, readBuffer, new serverReadHandler(this.serverChannel)); 
  28. } 
  29. } 
  30. @Override 
  31. public void failed(Throwable exc, ByteBuffer buffer) { 
  32. // TODO Auto-generated method stub 
  33. //出现异常关闭socketchannel 
  34. try { 
  35. this.serverChannel.close(); 
  36. } catch (IOException e) { 
  37. // TODO Auto-generated catch block 
  38. e.printStackTrace(); 
  39. } 
  40. } 
  41.  
  42. } 
  43.  

客户端

client

  1. //package aio; 
  2.  
  3. import java.util.Scanner; 
  4.  
  5. public class client { 
  6. private static String DEFAULT_HOST = "127.0.0.1";  
  7. private static int DEFAULT_PORT = 8000;  
  8. private static AsyncClientHandler clientHandle;  
  9. public static void start(){  
  10. start(DEFAULT_HOST,DEFAULT_PORT);  
  11. }  
  12. public static synchronized void start(String ip,int port){  
  13. if(clientHandle!=null)  
  14. return;  
  15. clientHandle = new AsyncClientHandler(ip,port);  
  16. new Thread(clientHandle,"Client").start();  
  17. }  
  18. //向服务器发送消息  
  19. public static boolean sendMsg(String msg) throws Exception{  
  20. if(msg.equals("q")) return false;  
  21. clientHandle.sendMsg(msg);  
  22. return true;  
  23. }  
  24. //@SuppressWarnings("resource")  
  25. public static void main(String[] args) throws Exception{  
  26. //System.out.println("请输入请求消息：");  
  27. start();  
  28. System.out.println("请输入请求消息：");  
  29. Scanner scanner = new Scanner(System.in);  
  30. String tmp = null; 
  31. for (int i = 0; i < 10; i++) { 
  32. tmp = scanner.nextLine(); 
  33. clientHandle.sendMsg(tmp); 
  34. } 
  35.  
  36. }  
  37. } 
  38.  

AsyncClientHandler

  1. //package 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. import java.util.concurrent.CountDownLatch; 
  9.  
  10. public class AsyncClientHandler implements CompletionHandler<Void, AsyncClientHandler>, Runnable{ 
  11. private AsynchronousSocketChannel clientChannel; 
  12. private String host; 
  13. private int port; 
  14. private CountDownLatch latch; 
  15.  
  16. public AsyncClientHandler(String host, int port) { 
  17. // TODO Auto-generated constructor stub 
  18. this.host = host; 
  19. this.port = port; 
  20. try { 
  21. //创建异步客户端通道 
  22. clientChannel = AsynchronousSocketChannel.open(); 
  23. } catch (Exception e) { 
  24. // TODO: handle exception 
  25. e.printStackTrace(); 
  26. } 
  27. } 
  28.  
  29.  
  30. @Override 
  31. public void run() { 
  32. // TODO Auto-generated method stub 
  33. latch = new CountDownLatch(1); 
  34. //创建InetScoketAddress 
  35. InetSocketAddress address = new InetSocketAddress(this.host, this.port); 
  36. //发起异步连接操作，回调参数是这个类本身，如果连接成功会回调completed方法 
  37. clientChannel.connect(address, this, this); 
  38. try { 
  39. latch.await(); 
  40. } catch (InterruptedException e) { 
  41. // TODO Auto-generated catch block 
  42. e.printStackTrace(); 
  43. } 
  44. try { 
  45. clientChannel.close(); 
  46. } catch (IOException e) { 
  47. // TODO Auto-generated catch block 
  48. e.printStackTrace(); 
  49. } 
  50. } 
  51.  
  52. @Override 
  53. public void completed(Void result, AsyncClientHandler attachment) { 
  54. // TODO Auto-generated method stub 
  55. //连接服务器成功，就会调用这个函数 
  56. System.out.println("连接服务器成功！！！！"); 
  57. } 
  58.  
  59. @Override 
  60. public void failed(Throwable exc, AsyncClientHandler attachment) { 
  61. // TODO Auto-generated method stub 
  62. //连接服务器失败，会调用这个函数 
  63. System.out.println("连接服务器失败！！！"); 
  64. exc.printStackTrace(); 
  65.  
  66. try { 
  67. clientChannel.close(); 
  68. latch.countDown(); 
  69. } catch (IOException e) { 
  70. // TODO Auto-generated catch block 
  71. e.printStackTrace(); 
  72. } 
  73. } 
  74.  
  75. //向服务器发送消息 
  76. public void sendMsg(String msg) { 
  77. System.out.println(msg); 
  78. byte[] bytes = msg.getBytes(); 
  79. ByteBuffer buffer = ByteBuffer.allocate(bytes.length); 
  80. buffer.put(bytes); 
  81. //进行flip操作 
  82. buffer.flip(); 
  83. //异步写 
  84. clientChannel.write(buffer, buffer, new WriteHandler(clientChannel,latch)); 
  85.  
  86.  
  87. } 
  88.  
  89.  
  90. } 
  91.  
  92.  

writeHandler

  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.util.concurrent.CountDownLatch; 
  8.  
  9. public class WriteHandler implements CompletionHandler<Integer, ByteBuffer>{ 
  10.  
  11. private AsynchronousSocketChannel clientChannel; 
  12. private CountDownLatch latch; 
  13. public WriteHandler(AsynchronousSocketChannel channel, CountDownLatch latch) { 
  14. // TODO Auto-generated constructor stub 
  15. this.clientChannel = channel; 
  16. this.latch = latch; 
  17. } 
  18. @Override 
  19. public void completed(Integer result, ByteBuffer buffer) { 
  20. // TODO Auto-generated method stub 
  21. System.out.println("发送数据成功!~~~"); 
  22. //进行flip操作 
  23. //buffer.flip(); 
  24. //判断buffer中是否有需要发送的数据，如果有数据就进行发送，如果没有就进行读取操作 
  25. if (buffer.hasRemaining()) { 
  26. System.out.println("进入写数据！！！"); 
  27. clientChannel.write(buffer, buffer, this); 
  28. System.out.println("发送数据成功~~~"); 
  29. } 
  30. //读取数据 
  31. System.out.println("进入读取数据"); 
  32. ByteBuffer readBuffer = ByteBuffer.allocate(1024); 
  33. clientChannel.read(readBuffer, readBuffer, new ReadHandle(clientChannel,latch)); 
  34.  
  35. } 
  36.  
  37. @Override 
  38. public void failed(Throwable exc, ByteBuffer buffer) { 
  39. // TODO Auto-generated method stub 
  40. System.err.println("发送数据失败~~~"); 
  41. try { 
  42. clientChannel.close(); 
  43. latch.countDown(); 
  44. } catch (IOException e) { 
  45. // TODO Auto-generated catch block 
  46. e.printStackTrace(); 
  47. } 
  48.  
  49. } 
  50.  
  51. } 
  52.  

ReadHandle

  1. //package aio; 
  2.  
  3. import java.io.IOException; 
  4. import java.io.UnsupportedEncodingException; 
  5. import java.nio.ByteBuffer; 
  6. import java.nio.channels.AsynchronousSocketChannel; 
  7. import java.nio.channels.CompletionHandler; 
  8. import java.util.concurrent.CountDownLatch; 
  9.  
  10. public class ReadHandle implements CompletionHandler<Integer,ByteBuffer> { 
  11. private AsynchronousSocketChannel clientChannel; 
  12. private CountDownLatch latch; 
  13. public ReadHandle(AsynchronousSocketChannel channel, CountDownLatch latch) { 
  14. // TODO Auto-generated constructor stub 
  15. this.latch = latch; 
  16. this.clientChannel = channel; 
  17. } 
  18. @Override 
  19. public void completed(Integer result, ByteBuffer buffer) { 
  20. // TODO Auto-generated method stub 
  21. //调整buffer的limit和position方便获取收到的数据 
  22. //buffer.flip(); 
  23. System.out.println("读取数据成功！！！"); 
  24. byte[] bytes = new byte[buffer.remaining()]; 
  25. buffer.get(bytes); 
  26. String res; 
  27. try { 
  28. res = new String(bytes, "UTF-8"); 
  29. System.out.println("收到的数据： " + res); 
  30. } catch (UnsupportedEncodingException e) { 
  31. // TODO Auto-generated catch block 
  32. e.printStackTrace(); 
  33. } 
  34.  
  35.  
  36.  
  37.  
  38. } 
  39. @Override 
  40. public void failed(Throwable exc, ByteBuffer attachment) { 
  41. // TODO Auto-generated method stub 
  42. System.err.println("读取数据失败~~~"); 
  43.  
  44. try { 
  45. clientChannel.close(); 
  46. this.latch.countDown(); 
  47. } catch (IOException e) { 
  48. // TODO Auto-generated catch block 
  49. e.printStackTrace(); 
  50. } 
  51. } 
  52.  
  53.  
  54. } 
  55.  

Java中的BIO、NIO、AIO-3

标签：文件操作   fun   public   物理链路   users   none   cell   except   存在   

原文地址：https://www.cnblogs.com/chailinbo/p/9226651.html
踩
(0)
赞
(0)
   
举报
评论 一句话评论（0）
登录后才能评论！
分享档案
更多>
2021年07月29日 (22)
2021年07月28日 (40)
2021年07月27日 (32)
2021年07月26日 (79)
2021年07月23日 (29)
2021年07月22日 (30)
2021年07月21日 (42)
2021年07月20日 (16)
2021年07月19日 (90)
2021年07月16日 (35)
周排行
mamicode.com排行更多图片
更多
友情链接
兰亭集智   国之画   百度统计   站长统计   阿里云   chrome插件   新版天听网
关于我们 - 联系我们 - 留言反馈
© 2014 mamicode.com 版权所有  联系我们:gaon5@hotmail.com
迷上了代码！