JAVA AIO编程

admin 管理员组

文章数量: 1086019

JAVA AIO编程

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 方式使用于连接数目多且连接比较长（重操作）的架构，比如相册服务器，充分调
用OS 参与并发操作，编程比较复杂，JDK7 开始支持。

package aio;
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;
import java.util.concurrent.Future;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();}}}

package aio;
import java.io.IOException;
import java.net.InetSocketAddress;
import java.nio.channels.AsynchronousChannelGroup;
import java.nio.channels.AsynchronousServerSocketChannel;
import java.util.Scanner;
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<AsynchronousSocketChannel, ? super T>)// 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;}}

package aio;
import java.io.UnsupportedEncodingException;
import java.nio.ByteBuffer;
import java.nio.channels.AsynchronousSocketChannel;
import java.nio.channels.CompletionHandler;
import java.util.Scanner;
import java.util.concurrent.ExecutionException;public class AIOServerHandler implements CompletionHandler<AsynchronousSocketChannel, AIOServer> {/*** 业务处理逻辑， 当请求到来后，监听成功，应该做什么。* 一定要实现的逻辑： 为下一次客户端请求开启监听。accept方法调用。* result参数 ： 就是和客户端直接建立关联的通道。*  无论BIO、NIO、AIO中，一旦连接建立，两端是平等的。*  result中有通道中的所有相关数据。如：OS操作系统准备好的读取数据缓存，或等待返回数据的缓存。*/@Overridepublic void completed(AsynchronousSocketChannel result, AIOServer attachment) {// 处理下一次的客户端请求。类似递归逻辑attachment.getServerChannel().accept(attachment, this);doRead(result);}/*** 异常处理逻辑， 当服务端代码出现异常的时候，做什么事情。*/@Overridepublic 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<Integer, ? super A> handler)* destination - 目的地， 是处理客户端传递数据的中转缓存。 可以不使用。* attachment - 处理客户端传递数据的对象。 通常使用Buffer处理。* handler - 处理逻辑*/channel.read(buffer, buffer, new CompletionHandler<Integer, ByteBuffer>() {/*** 业务逻辑，读取客户端传输数据* attachment - 在completed方法执行的时候，OS已经将客户端请求的数据写入到Buffer中了。*  但是未复位（flip）。 使用前一定要复位。*/@Overridepublic 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();}}@Overridepublic void failed(Throwable exc, ByteBuffer attachment) {exc.printStackTrace();}});}}

本文标签： JAVA AIO编程