JavaAIO编程

JavaAIO编程

Asynchronous IO： 异步非阻塞的编程方式

与 NIO 不同，当进行读写操作时，只须直接调用 API 的 read 或 write 方法即可。这两种 方法均为异步的，对于读操作而言，当有流可读取时，操作系统会将可读的流传入 read 方 法的缓冲区，并通知应用程序；对于写操作而言，当操作系统将 write 方法传递的流写入完 毕时，操作系统主动通知应用程序。即可以理解为，read/write 方法都是异步的，完成后会 主动调用回调函数。在 JDK1.7 中，这部分内容被称作 NIO.2，主要在 java.nio.channels 包下 增加了下面四个异步通道：

AsynchronousSocketChannel

AsynchronousServerSocketChannel

AsynchronousFileChannel

AsynchronousDatagramChannel

异步非阻塞，服务器实现模式为一个有效请求一个线程，客户端的 I/O 请求都是由 OS 先完成了再通知服务器应用去启动线程进行处理。

AIO 方式使用于连接数目多且连接比较长（重操作）的架构

import java.io.IOException;import java.net.InetSocketAddress;import java.nio.channels.AsynchronousChannelGroup;import java.nio.channels.AsynchronousServerSocketChannel;import java.util.concurrent.ExecutorService;import java.util.concurrent.Executors;import java.util.concurrent.TimeUnit;public class AIOServer { // 线程池， 提高服务端效率。 private ExecutorService service; // 线程组 // private AsynchronousChannelGroup group; // 服务端通道， 针对服务器端定义的通道。 private AsynchronousServerSocketChannel serverChannel; public AIOServer(int port){ init(9999); } private void init(int port){ try { System.out.println("server starting at port : " + port + " ..."); // 定长线程池 service = Executors.newFixedThreadPool(4); /* 使用线程组 group = AsynchronousChannelGroup.withThreadPool(service); serverChannel = AsynchronousServerSocketChannel.open(group); */ // 开启服务端通道， 通过静态方法创建的。 serverChannel = AsynchronousServerSocketChannel.open(); // 绑定监听端口， 服务器启动成功，但是未监听请求。 serverChannel.bind(new InetSocketAddress(port)); System.out.println("server started."); // 开始监听 // accept(T attachment, CompletionHandler) // AIO开发中，监听是一个类似递归的监听操作。每次监听到客户端请求后，都需要处理逻辑开启下一次的监听。 // 下一次的监听，需要服务器的资源继续支持。 serverChannel.accept(this, new AIOServerHandler()); try { TimeUnit.SECONDS.sleep(Integer.MAX_VALUE); } catch (InterruptedException e) { e.printStackTrace(); } } catch (IOException e) { e.printStackTrace(); } } public static void main(String[] args) { new AIOServer(9999); } public ExecutorService getService() { return service; } public void setService(ExecutorService service) { this.service = service; } public AsynchronousServerSocketChannel getServerChannel() { return serverChannel; } public void setServerChannel(AsynchronousServerSocketChannel serverChannel) { this.serverChannel = serverChannel; } }

import java.io.IOException;import java.io.UnsupportedEncodingException;import java.net.InetSocketAddress;import java.nio.ByteBuffer;import java.nio.channels.AsynchronousSocketChannel;import java.util.Scanner;import java.util.concurrent.ExecutionException;public class AIOClient { private AsynchronousSocketChannel channel; public AIOClient(String host, int port) { init(host, port); } private void init(String host, int port) { try { // 开启通道 channel = AsynchronousSocketChannel.open(); // 发起请求，建立连接。 channel.connect(new InetSocketAddress(host, port)); } catch (IOException e) { e.printStackTrace(); } } public void write(String line) { try { ByteBuffer buffer = ByteBuffer.allocate(1024); buffer.put(line.getBytes("UTF-8")); buffer.flip(); channel.write(buffer); } catch (UnsupportedEncodingException e) { e.printStackTrace(); } } public void read() { ByteBuffer buffer = ByteBuffer.allocate(1024); try { // read方法是异步方法，OS实现的。get方法是一个阻塞方法，会等待OS处理结束后再返回。 channel.read(buffer).get(); // channel.read(dst, attachment, handler); buffer.flip(); System.out.println("from server : " + new String(buffer.array(), "UTF-8")); } catch (InterruptedException e) { e.printStackTrace(); } catch (ExecutionException e) { e.printStackTrace(); } catch (UnsupportedEncodingException e) { e.printStackTrace(); } } public void doDestory() { if (null != channel) { try { channel.close(); } catch (IOException e) { e.printStackTrace(); } } } public static void main(String[] args) { AIOClient client = new AIOClient("localhost", 9999); try { System.out.print("enter message send to server > "); Scanner s = new Scanner(System.in); String line = s.nextLine(); client.write(line); client.read(); } finally { client.doDestory(); } }}

import java.io.UnsupportedEncodingException;import java.nio.ByteBuffer;import java.nio.channels.AsynchronousSocketChannel;import java.nio.channels.CompletionHandler;import java.util.Scanner;public class AIOServerHandler implements CompletionHandler { /** * 业务处理逻辑， 当请求到来后，监听成功，应该做什么。 * 一定要实现的逻辑： 为下一次客户端请求开启监听。accept方法调用。 * result参数 ： 就是和客户端直接建立关联的通道。 * 无论BIO、NIO、AIO中，一旦连接建立，两端是平等的。 * result中有通道中的所有相关数据。如：OS操作系统准备好的读取数据缓存，或等待返回数据的缓存。 */ @Override public void completed(AsynchronousSocketChannel result, AIOServer attachment) { // 处理下一次的客户端请求。类似递归逻辑 attachment.getServerChannel().accept(attachment, this); doRead(result); } /** * 异常处理逻辑， 当服务端代码出现异常的时候，做什么事情。 */ @Override public void failed(Throwable exc, AIOServer attachment) { exc.printStackTrace(); } /** * 真实项目中，服务器返回的结果应该是根据客户端的请求数据计算得到的。不是等待控制台输入的。 * @param result */ private void doWrite(AsynchronousSocketChannel result){ try { ByteBuffer buffer = ByteBuffer.allocate(1024); System.out.print("enter message send to client > "); Scanner s = new Scanner(System.in); String line = s.nextLine(); buffer.put(line.getBytes("UTF-8")); // 重点：必须复位，必须复位，必须复位 buffer.flip(); // write方法是一个异步操作。具体实现由OS实现。 可以增加get方法，实现阻塞，等待OS的写操作结束。 result.write(buffer); // result.write(buffer).get(); // 调用get代表服务端线程阻塞，等待写操作完成 } catch (UnsupportedEncodingException e) { e.printStackTrace(); }/* catch (InterruptedException e) { e.printStackTrace(); } catch (ExecutionException e) { e.printStackTrace(); }*/ } private void doRead(final AsynchronousSocketChannel channel){ ByteBuffer buffer = ByteBuffer.allocate(1024); /* * 异步读操作， read(Buffer destination, A attachment, * CompletionHandler handler) * destination - 目的地， 是处理客户端传递数据的中转缓存。 可以不使用。 * attachment - 处理客户端传递数据的对象。 通常使用Buffer处理。 * handler - 处理逻辑 */ channel.read(buffer, buffer, new CompletionHandler() { /** * 业务逻辑，读取客户端传输数据 * attachment - 在completed方法执行的时候，OS已经将客户端请求的数据写入到Buffer中了。 * 但是未复位（flip）。 使用前一定要复位。 */ @Override public void completed(Integer result, ByteBuffer attachment) { try { System.out.println(attachment.capacity()); // 复位 attachment.flip(); System.out.println("from client : " + new String(attachment.array(), "UTF-8")); doWrite(channel); } catch (UnsupportedEncodingException e) { e.printStackTrace(); } } @Override public void failed(Throwable exc, ByteBuffer attachment) { exc.printStackTrace(); } }); }}

版权声明：本文内容由网络用户投稿，版权归原作者所有，本站不拥有其著作权，亦不承担相应法律责任。如果您发现本站中有涉嫌抄袭或描述失实的内容，请联系我们jiasou666@gmail.com 处理，核实后本网站将在24小时内删除侵权内容。