当前位置：
  1. 首页
  2. 市民
  3. 正文

前言

　　之前已经整理过了bio、nio两种i/o的相关博文，每一种i/o都有其特点，但相对开发而言，肯定是要又高效又简单的i/o编程才是真正需要的，在之前的nio博文（深入学习netty（2）——传统nio编程）中就已经介绍过nio编程的缺点（相比较而言的缺点：同步非阻塞，需要单独开启线程不断轮询），所以才会有真正的异步非阻塞i/o出现，这就是此篇博文需要介绍的aio编程。

　　参考资料《netty in action》、《netty权威指南》（有需要的小伙伴可以评论或者私信我）

　　博文中所有的代码都已上传到github，欢迎star、fork

　　感兴趣可以先学习相关博文：

  • 深入学习netty（1）——传统bio编程
  • 深入学习netty（2）——传统nio编程

 

 

一、nio 2.0与aio编程

　　jdk 1.7升级了nio类库，升级后的nio类库称之为nio 2.0，java提供了异步文件i/o操作，同时提供了与unix网络编程事件驱动i/o对应的aio。

　　nio 2.0的异步套接字通道是真正的异步非阻塞i/o，对应有unix网络编程中的事件驱动i/o（aio），相比较nio，它不需要通过selector对注册的通道进行轮询操作即可实现异步读写，简化了nio的编程模型。

　　nio 2.0提供了新的异步通道的概念，异步通道提供了以下两种方式获取操作结果：

  • 通过juc.futrue类来表示异步操作的结果。
asynchronoussocketchannel socketchannel = asynchronoussocketchannel.open();
inetsocketaddress inetsocketaddress = new inetsocketaddress("localhost", 8080);
future<void> connect = socketchannel.connect(inetsocketaddress);
while (!connect.isdone()) {
　　thread.sleep(10);
}
  • 在异步操作的时候传入java.nio.channels。实现completionhandler接口complete()的方法作为操作完成回调。
private class mycompletionhandler implements completionhandler<integer, bytebuffer> {
 
　　@override
　　public void completed(integer result, bytebuffer attachment) {
　　　　// todo 回调后业务操作
　　}
　　@override
　　public void failed(throwable t, bytebuffer attachment) {
　　　　t.printstacktrace();
　　}

二、aio服务端

（1）服务端aio异步处理任务asynctimeserverhandler：

  • 创建异步服务通道并监听端口
  • 异步监听客户端连接
/**
 * 服务端aio异步处理任务
 * -创建异步服务通道监听端口
 * -监听客户端连接
 */
public class asynctimeserverhandler implements runnable{

    private int port;

    countdownlatch latch;
    asynchronousserversocketchannel asynchronousserversocketchannel;

    public asynctimeserverhandler(int port) {
        this.port = port;
        try {
            // 创建异步的服务通道asynchronousserversocketchannel, 并bind监听端口
            asynchronousserversocketchannel = asynchronousserversocketchannel.open();
            asynchronousserversocketchannel.bind(new inetsocketaddress(port));
            system.out.println("the time server is start in port : " + port);
        } catch (ioexception e) {
            e.printstacktrace();
        }
    }

    @override
    public void run() {
        // countdownlatch没有count减一，所以导致一直阻塞
        latch = new countdownlatch(1);
        doaccept();
        try {
            // 防止执行操作线程还未结束，服务端线程就退出，程序不退出的前提下，才能够让accept继续可以回调接受来自客户端的连接
            // 实际开发过程中不需要单独开启线程去处理asynchronousserversocketchannel
            latch.await();
        } catch (interruptedexception e) {
            e.printstacktrace();
        }
    }

    /**
     * 接收客户端的连接
     * 参数completionhandler类型的handler实例来接收accept操作成功的通知消息
     */
    public void doaccept() {
        asynchronousserversocketchannel.accept(this, new acceptcompletionhandler());
    }
}

（2）服务端连接异步回调处理器acceptcompletionhandler：异步处理客户端连接完成后的操作

/**
 * 客户端连接异步处理器
 * completed()方法完成回调logic
 * failed()方法完成失败回调logic
 */
public class acceptcompletionhandler implements completionhandler<asynchronoussocketchannel, asynctimeserverhandler> {

    /**
     * 调用该方法表示客户端已经介接入成功
     * 同时再accept接收新的客户端连接
     * @param result
     * @param attachment
     */
    @override
    public void completed(asynchronoussocketchannel result,
                          asynctimeserverhandler attachment) {
        // 此时还要继续调用accept方法是因为，completed方法表示上一个客户端连接完成，而下一个新的客户端需要连接
        // 如此形成新的循环：每接收一个客户端的成功连接之后，再异步接收新的客户端连接
        attachment.asynchronousserversocketchannel.accept(attachment, this);
        // 预分配1m的缓冲区
        bytebuffer buffer = bytebuffer.allocate(1024);
        // 调用read方法异步读，传入completionhandler类型参数异步回调读事件
        result.read(buffer, buffer, new readcompletionhandler(result));
    }

    @override
    public void failed(throwable exc, asynctimeserverhandler attachment) {
        exc.printstacktrace();
        // 让服务线程不再阻塞
        attachment.latch.countdown();
    }
}

（3）服务端read事件异步回调处理器readcompletionhandler：异步回调处理客户端请求数据

/**
 * 服务端read事件异步处理器
 *  completed异步回调处理客户端请求数据
 */
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数组
        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();
            // write异步回调，传入completionhandler类型参数
            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) {
                                // todo 只要是i/o异常就需要关闭链路，释放资源

                            }
                        }
                    });
        }
    }

    @override
    public void failed(throwable exc, bytebuffer attachment) {
        try {
            this.channel.close();
        } catch (ioexception e) {
            e.printstacktrace();
            // todo 只要是i/o异常就需要关闭链路，释放资源
        }
    }
}

（4）服务端启动timeserver

/**
 * aio 异步非阻塞服务端
 * 不需要单独开线程去处理read、write等事件
 * 只需要关注complete-handlers中的回调completed方法
 */
public class timeserver {

    public static void main(string[] args) throws ioexception {
        int port = 8086;
        asynctimeserverhandler timeserver = new asynctimeserverhandler(port);
        new thread(timeserver, "aio-asynctimeserverhandler").start();
    }
}

（5）启动服务端

服务端console：

使用命令netstat查看8086端口是否监听

三、aio客户端

（1）客户端aio异步回调处理任务：

  • 打开asynchronoussocketchannel通道，连接服务端
  • 发送服务端指令
  • 回调处理服务端应答
/**
 * 客户端aio异步回调处理任务
 * -打开asynchronoussocketchannel通道，连接服务端
 * -发送服务端指令
 * -回调处理服务端应答
 */
public class asynctimeclienthandler implements completionhandler<void, asynctimeclienthandler>, runnable {

    private asynchronoussocketchannel client;
    private string host;
    private int port;
    private countdownlatch latch;

    public asynctimeclienthandler(string host, int port) {
        this.host = host;
        this.port = port;
        try {
            client = asynchronoussocketchannel.open();
        } catch (ioexception e) {
            e.printstacktrace();
        }
    }

    @override
    public void run() {
        latch = new countdownlatch(1);
        client.connect(new inetsocketaddress(host, port), this, this);
        try {
            // 防止异步操作都没完成，连接线程就结束退出
            latch.await();
        } catch (interruptedexception e1) {
            e1.printstacktrace();
        }
        try {
            client.close();
        } catch (ioexception e) {
            e.printstacktrace();
        }
    }

    /**
     * 发送请求完成异步回调
     * @param result
     * @param attachment
     */
    @override
    public void completed(void result, asynctimeclienthandler attachment) {
        byte[] req = "query time order".getbytes();
        bytebuffer writebuffer = bytebuffer.allocate(req.length);
        writebuffer.put(req);
        writebuffer.flip();
        client.write(writebuffer, writebuffer,
                new completionhandler<integer, bytebuffer>() {
                    @override
                    public void completed(integer result, bytebuffer buffer) {
                        if (buffer.hasremaining()) {
                            client.write(buffer, buffer, this);
                        } else {
                            bytebuffer readbuffer = bytebuffer.allocate(1024);
                            // 回调服务端应答消息
                            client.read(readbuffer, readbuffer,
                                    new completionhandler<integer, bytebuffer>() {
                                        @override
                                        public void completed(integer result, bytebuffer buffer) {
                                            buffer.flip();
                                            byte[] bytes = new byte[buffer.remaining()];
                                            buffer.get(bytes);
                                            string body;
                                            try {
                                                body = new string(bytes, "utf-8");
                                                system.out.println("now is : " + body);
                                                // 服务端应答完成后，连接线程退出
                                                latch.countdown();
                                            } catch (unsupportedencodingexception e) {
                                                e.printstacktrace();
                                            }
                                        }

                                        @override
                                        public void failed(throwable exc, bytebuffer attachment) {
                                            try {
                                                client.close();
                                                // 防止线程一直阻塞
                                                latch.countdown();
                                            } catch (ioexception e) {
                                                // ingnore on close
                                            }
                                        }
                                    });
                        }
                    }

                    @override
                    public void failed(throwable exc, bytebuffer attachment) {
                        try {
                            client.close();
                            latch.countdown();
                        } catch (ioexception e) {
                            // ingnore on close
                        }
                    }
                });
    }

    @override
    public void failed(throwable exc, asynctimeclienthandler attachment) {
        exc.printstacktrace();
        try {
            client.close();
            latch.countdown();
        } catch (ioexception e) {
            e.printstacktrace();
        }
    }
}

（2）客户端timeclient

/**
 * aio 异步非阻塞 客户端
 * 不需要单独开线程去处理read、write等事件
 * 只需要关注complete-handlers中的回调completed方法
 */
public class timeclient {
    public static void main(string[] args) {
        int port = 8086;
        new thread(new asynctimeclienthandler("127.0.0.1", port), "aio-asynctimeclienthandler").start();

    }
}

（3）启动客户端

客户端console：

服务端console：

四、总结

服务端通过countdownlatch一直阻塞

由代码实践我们可知：

　　jdk底层通过threadpoolexecutor执行回调通知，异步回调通知类由sun.nio.ch.asynchronouschannelgroupimpl实现，然后将任务提交到该线程池以处理i/o事件，并分派给completion-handlers ，该队列消耗对组中通道执行的异步操作的结果。

　　异步socketchannel是被动执行，不需要单独像nio编程那样单独创建一个独立的i/o线程处理读写操作，都是由jdk底层的线程池负责回调并驱动读写操作的。所以基于nio 2.0的新的异步非阻塞相比较nio编程要简单，这两区别在于：

  • 在nio中等待io事件由我们注册的selector来完成，在感兴趣的事情来了，我们的线程来accept.read.write.connect…解析，解析完后再交由业务逻辑处理。
  • 而在在异步io（aio、nio 2.0）中等待io事件同样为accept，read，write，connect，但数据处理交由系统完成，我们需要做的就是在completionhandlers中处理业务逻辑回调即可。
上一篇 精尽Spring Boot源码分析 – Condition 接口的扩展
下一篇 1976年属龙的今年几岁，1976出生的今年多少岁

相关文章