程序员社区

使用AIO 完成网络通信

前言

jdk7中新增了一些与文件(网络)I/O相关的一些api。这些API被称为NIO.2,或称为AIO(Asynchronous I/O)。AIO最大的一个特性就是异步能力,这种能力对socket与文件I/O都起作用。AIO其实是一种在读写操作结束之前允许进行其他操作的I/O处理。AIO是对JDK1.4中提出的同步非阻塞I/O(NIO)的进一步增强。

正文

首先创建异步的时间服务器处理类,然后启动线程将AsyncTimeServerHandler装载进去执行。

public class TimeServer {
    /**
     * @param args
     * @throws IOException
     */
    public static void main(String[] args) throws IOException {
        // 采用默认值
        int port = 8080;
        if (args != null && args.length > 0) {
            try {
                port = Integer.valueOf(args[0]);
            } catch (NumberFormatException e) {

            }
        }
        AsyncTimeServerHandler timeServer = new AsyncTimeServerHandler(port);
        new Thread(timeServer, "AIO-AsyncTimeServerHandler-001").start();
    }
}

下面是AsyncTimeServerHandler的代码

public class AsyncTimeServerHandler implements Runnable {

    private int port;

    CountDownLatch latch;
    AsynchronousServerSocketChannel asynchronousServerSocketChannel;

    public AsyncTimeServerHandler(int port) {
    this.port = port;
    try {
        asynchronousServerSocketChannel = AsynchronousServerSocketChannel .open();
        asynchronousServerSocketChannel.bind(new InetSocketAddress(port));
        System.out.println("The time server is start in port : " + port);
    } catch (IOException e) {
        e.printStackTrace();
    }
    }

    /*
     * (non-Javadoc)
     * 
     * @see java.lang.Runnable#run()
     */
    @Override
    public void run() {
       //这里使用门栓,将服务器线程阻塞住,防止执行完直接退出(因为整个过程都是异步的过程)。
    latch = new CountDownLatch(1);
    doAccept();
    try {
        latch.await();
    } catch (InterruptedException e) {
        e.printStackTrace();
    }
    }

    public void doAccept() {
    asynchronousServerSocketChannel.accept(this, new AcceptCompletionHandler());
    }

在构造方法中,我们首先创建一个异步的服务端通道AsynchronousServerSocketChannel,然后调用它的bind方法绑定监听端口。

接下来,初始化CountDownLatch对象,它的作用是在完成一组正在执行的操作之前,允许当前的线程一直阻塞。在本例程中,我们让线程在此阻塞,防止服务端执行完成退出。

由于是异步操作,我们可以传递一个CompletionHandler<AsynchronousSocketChannel,? super A>类型的handler实例接收accept操作成功的通知消息,在本例程中我们通过AcceptCompletionHandler实例作为handler接收通知消息,下面,我们继续对AcceptCompletionHandler进行分析。

public class AcceptCompletionHandler implements  CompletionHandler<AsynchronousSocketChannel,AsyncTimeServerHandler> {
    @Override
    public void completed(AsynchronousSocketChannel result, AsyncTimeServerHandler attachment) {
    attachment.asynchronousServerSocketChannel.accept(attachment, this);
    ByteBuffer buffer = ByteBuffer.allocate(1024);
    result.read(buffer, buffer, new ReadCompletionHandler(result));
    }

    @Override
    public void failed(Throwable exc, AsyncTimeServerHandler attachment) {
    exc.printStackTrace();
    attachment.latch.countDown();
    }
}

CompletionHandler有两个方法,分别是completedfailed
首先看completed接口的实现,从attachment获取成员变量AsynchronousServerSocketChannel,然后继续调用它的accept方法。当我们调用AsynchronousServerSocketChannel的accept方法后,如果有新的客户端连接接入,系统将回调我们传入的CompletionHandler实例的completed方法,表示新的客户端已经接入成功,因为一个AsynchronousServerSocketChannel可以接收成千上万个客户端,所以我们需要继续调用它的accept方法,接收其它的客户端连接,最终形成一个循环。每当接收一个客户读连接成功之后,再异步接收新的客户端连接。

链路建立成功之后,服务端需要接收客户端的请求消息,过调用AsynchronousSocketChannel的read方法进行异步读操作。

下面是ReadCompletionHandler 的代码

public class ReadCompletionHandler implements CompletionHandler<Integer, ByteBuffer> {

    private AsynchronousSocketChannel channel;

    public ReadCompletionHandler(AsynchronousSocketChannel channel) {
    if (this.channel == null)
        this.channel = channel;
    }

    @Override
    public void completed(Integer result, ByteBuffer attachment) {
    attachment.flip();
    byte[] body = new byte[attachment.remaining()];
    attachment.get(body);
    try {
        String req = new String(body, "UTF-8");
        System.out.println("The time server receive order : " + req);
        String currentTime = "QUERY TIME ORDER".equalsIgnoreCase(req) ? new java.util.Date(
            System.currentTimeMillis()).toString() : "BAD ORDER";
        doWrite(currentTime);
    } catch (UnsupportedEncodingException e) {
        e.printStackTrace();
    }
    }

    private void doWrite(String currentTime) {
    if (currentTime != null && currentTime.trim().length() > 0) {
        byte[] bytes = (currentTime).getBytes();
        ByteBuffer writeBuffer = ByteBuffer.allocate(bytes.length);
        writeBuffer.put(bytes);
        writeBuffer.flip();
        channel.write(writeBuffer, writeBuffer,
            new CompletionHandler<Integer, ByteBuffer>() {
            @Override
            public void completed(Integer result, ByteBuffer buffer) {
                // 如果没有发送完成,继续发送
                if (buffer.hasRemaining())
                channel.write(buffer, buffer, this);
            }

            @Override
            public void failed(Throwable exc, ByteBuffer attachment) {
                try {
                channel.close();
                } catch (IOException e) {
                // ingnore on close
                }
            }
            });
    }
    }

    @Override
    public void failed(Throwable exc, ByteBuffer attachment) {
    try {
        this.channel.close();
    } catch (IOException e) {
        e.printStackTrace();
    }
    }
}

首先看构造方法,我们将AsynchronousSocketChannel通过参数传递到ReadCompletionHandler中当作成员变量来使用,主要用于读取半包消息和发送应答。然后看对消息的处理,读取到消息后的处理,首先对attachment进行flip操作,为后续从缓冲区读取数据做准备。根据缓冲区的可读字节数创建byte数组,然后通过new String方法创建请求消息,对请求消息进行判断,如果是”QUERY TIME ORDER”则获取当前系统服务器的时间,调用doWrite方法发送给客户端。

之后是发送消息给客户端,这里还是个异步的操作,调用AsynchronousSocketChannel的异步write方法。和前面的异步read方法一样,它也有三个与read方法相同的参数,将CompletionHandler的实现类传进去,write完毕会回调我们实现类的completed方法,继续执行。

最后是failed方法,它的实现很简单,就是当发生异常的时候,对异常Throwable进行判断,如果是IO异常,就关闭链路,释放资源,如果是其它异常,按照业务自己的逻辑进行处理。

赞(0) 打赏
未经允许不得转载:IDEA激活码 » 使用AIO 完成网络通信

相关推荐

  • 暂无文章

一个分享Java & Python知识的社区