登录 注册

4. Netty 初认识 --AIO 编程

本贴最后更新于 1587 天前，其中的信息可能已经时移世异

通过上篇博文我们已能发现 NIO 编程难度确实比同步阻塞 BIO 大很多,而且之前的 NIO 并没有考虑"半包读","半包写"，如果加上这些,会更加复杂,那为什么 NIO 使用越来越广泛，它的优点如下:

  1. 客户端发起的连接操作是异步的,可以通过在多路复用器注册 OP_CONNECT 等待后续结果,不需要像之前的客户端那样被同步阻塞。

  2. SocketChannel 的读写操作都是异步的，如果没有可读写的数据它不会同步等待，直接返回，这样 I/O 通信线程就可以处理其他的链路,不需要同步等待这个链路可用。

  3. 线程模型的优化:由于 JDK 的 Selector 在 Linux 等主流操作系统上通过 epoll 实现,它没有连接句柄数的限制(只受限于操作系统的最大句柄数或者对单个进程的句柄限制),这意味着一个 Selector 线程可用同时处理成千上万个客户端连接，而且性能不会随着客户端的增加而线性下降，一会那次，它非常适合做高性能、高负载的网络服务器。

    JDK1.7 升级了 NIO 类库,升级后的 NIO 类库被称为 NIO2.0,引人注目的是,java 正式提供了异步文件 I/O 操作，同时提供了与 UNIX 网络编程事件驱动 I/O 对应的 AIO; 它不需要通过多路复用器 Selector 对注册的通道进行轮询操作即可实现异步读写。下面以 AIO 举例上篇文章的客户端连接服务器获取当前时间。

NIO2.0 提供了异步文件通道和异步套接字通道的实现，异步通道提供两种方式后去操作结果:

通过 java.util.concurrent.Future 类来标识异步操作的结果;
在执行异步操作的时候传入一个 java.nio.channels.

CompletionHandler 接口的实现类作为操作完成的回调。

服务端
TimeServer.java

package club.wujingjian.com.wujingjian.aio.server;


public class TimeServer {

    public static void main(String[] args) {
        int port = 9090;

        if (args != null && args.length > 0) {
            try {
                port = Integer.parseInt(args[0]);
            } catch (NumberFormatException e) {
                e.printStackTrace();
            }
        }
        //创建异步的时间服务器处理类
        AsyncTimeServerHandler timeServer = new AsyncTimeServerHandler(port);
        //启动异步时间服务器处理类的线程
        new Thread(timeServer, "AIO-AsyncTimeServerHandler-001").start();
    }
}

AsyncTimeServerHandler.java

package club.wujingjian.com.wujingjian.aio.server;

import java.io.IOException;
import java.net.InetSocketAddress;
import java.nio.channels.AsynchronousServerSocketChannel;
import java.util.concurrent.CountDownLatch;

public class AsyncTimeServerHandler implements Runnable {

    private int port;
    CountDownLatch latch;
    AsynchronousServerSocketChannel asynchronousServerSocketChannel;

    public AsyncTimeServerHandler(int port) {
        this.port = port;
        try {
            //创建一个异步服务端通道 AsynchronousServerSocketChannel
            asynchronousServerSocketChannel = AsynchronousServerSocketChannel.open();
            //调用bind方法绑定监听端口,如果端口合法且没被占用,绑定成功
            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作用是在完成一组正在执行的操作之前,允许当前的线程一直阻塞,在本例中我们让线程在此阻塞,防止服务端执行完成退出
        //在实际项目中不需要启动独立线程来处理AsynchronousServerSocketChannel，这里仅仅做个示例展示
        latch = new CountDownLatch(1);
        doAccept(); //用于接收客户端的连接
        try {
            latch.await();
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }

    public void doAccept(){
        //AcceptCompletionHandler 用来接收通知消息
        asynchronousServerSocketChannel.accept(this, new AcceptCompletionHandler());
    }
}

AcceptCompletionHandler.java

package club.wujingjian.com.wujingjian.aio.server;

import java.nio.ByteBuffer;
import java.nio.channels.AsynchronousSocketChannel;
import java.nio.channels.CompletionHandler;

public class AcceptCompletionHandler implements CompletionHandler<AsynchronousSocketChannel, AsyncTimeServerHandler> {


    @Override
    public void completed(AsynchronousSocketChannel result, AsyncTimeServerHandler attachment) {
        //从attachment获取成员变量AsynchronousServerSocketChannel,然后继续调用它的accept方法.
        //当我们调用AsynchronousServerSocketChannel的accept方法后,如果有新的客户端连接接入,系统将回调我们传入的CompletionHandler示例的completed方法,表示新
        //的客户端已经接入成功,因为一个AsynchronousSocketChannel可以接收成千上万个客户端,搜易我们这里要继续调用它的accept方法,接收其他的客户端连接,最终形成
        //一个循环.每当接收一个客户读连接成功之后,再一步接收新的客户端连接
        attachment.asynchronousServerSocketChannel.accept(attachment, this);

        //链路建立成功后,服务端需要接收客户端的请求,创建ByteBuffer
        ByteBuffer buffer = ByteBuffer.allocate(1024);

        /**   AsynchronousSocketChannel的read方法参数解释如下:
         *  第一个参数ByteBuffer dst: 接收缓冲区,用于从异步Chanenl中读取数据包;
         *  第二个参数A attachemnt:  异步Channel携带的附件，通知回调的时候作为入参使用;
         *  第三个参数CompletionHandler: 接收通知回调的业务handler,此处用ReadCompletionHandler
         */
        result.read(buffer, buffer, new ReadCompletionHandler(result));
    }

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

ReadCompletionHandler.java

package club.wujingjian.com.wujingjian.aio.server;

import java.io.IOException;
import java.io.UnsupportedEncodingException;
import java.nio.ByteBuffer;
import java.nio.channels.AsynchronousSocketChannel;
import java.nio.channels.CompletionHandler;
import java.util.Date;

public class ReadCompletionHandler implements CompletionHandler<Integer, ByteBuffer> {

    private AsynchronousSocketChannel channel;

    //将AsynchronousSocketChannel 传入构造参数中,主要用于读取半包消息和发送应答，本示例不对半包读写进行具体说明
    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 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) {
                        //ignore on close
                    }
                }
            });
        }
    }

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

客户端

TimeClient.java

package club.wujingjian.com.wujingjian.aio.client;


public class TimeClient {
    public static void main(String[] args) {
        int port = 9090;
        if (args != null && args.length > 0) {
            try {
                port = Integer.parseInt(args[0]);
            } catch (NumberFormatException e) {
                e.printStackTrace();
            }
        }
        new Thread(new AsyncTimeClientHandler("127.0.0.1",port), "AIO-AsyncTimeClientHandler-001").start();
    }
}

AsyncTimeClientHandler.java

package club.wujingjian.com.wujingjian.aio.client;


import java.io.IOException;
import java.io.UnsupportedEncodingException;
import java.net.InetSocketAddress;
import java.nio.ByteBuffer;
import java.nio.channels.AsynchronousSocketChannel;
import java.nio.channels.CompletionHandler;
import java.util.concurrent.CountDownLatch;

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);//创建CountDownLatch进行等待,防止异步操作没有执行完线程就退出
        //client.connect第二个参数A attach :  AsynchronousSocketChannel的附件,用于回调通知时作为入参被传递,调用者可以自定义
        //client.connect第三个参数CompletionHandler: 异步操作回调通知接口,由调用者实现
        //本例中我们第二，三个参数都是用了AsyncTimeClientHandler类本身,因为它实现了CompletionHandler接口
        client.connect(new InetSocketAddress(host, port), this, this);
        try {
            latch.await();
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        try {
            client.close();
        } catch (IOException e1) {
            e1.printStackTrace();
        }
    }

    @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);
                            } catch (UnsupportedEncodingException e) {
                                e.printStackTrace();
                            }
                        }

                        @Override
                        public void failed(Throwable exc, ByteBuffer attachment) {
                            try {
                                client.close();
                                latch.countDown();
                            } catch (IOException e) {
                                //ignore on close
                            }
                        }
                    });
                }
            }

            @Override
            public void failed(Throwable exc, ByteBuffer attachment) {
                try {
                    client.close();
                    latch.countDown();
                } catch (IOException e) {
                    // ignore on close
                }
            }
        });
    }

    @Override
    public void failed(Throwable exc, AsyncTimeClientHandler attachment) {
        try {
            client.close();
            latch.countDown();
        } catch (IOException e) {
            // ignore on close
        }
    }
}

运行结果 Server 端
image.png
客户端:

image.png

  • Netty

    Netty 是一个基于 NIO 的客户端-服务器编程框架，使用 Netty 可以让你快速、简单地开发出一个可维护、高性能的网络应用，例如实现了某种协议的客户、服务端应用。

    49 引用 • 33 回帖 • 27 关注
北京 位置
289 9 67 34 33 12 17 112

相关帖子

回帖
4. Netty 初认识 --AIO 编程

欢迎来到这里！

我们正在构建一个小众社区，大家在这里相互信任，以平等 • 自由 • 奔放的价值观进行分享交流。最终，希望大家能够找到与自己志同道合的伙伴，共同成长。

注册 关于
请输入回帖内容 ...
289306290
北京
回帖
2
 帖子
33
 积分
3

近期热议

推荐标签 标签

  • jQuery

    jQuery 是一套跨浏览器的 JavaScript 库，强化 HTML 与 JavaScript 之间的操作。由 John Resig 在 2006 年 1 月的 BarCamp NYC 上释出第一个版本。全球约有 28% 的网站使用 jQuery，是非常受欢迎的 JavaScript 库。

    63 引用 • 134 回帖 • 733 关注
  • 招聘

    哪里都缺人，哪里都不缺人。

    189 引用 • 1056 回帖 • 2 关注
  • Sphinx

    Sphinx 是一个基于 SQL 的全文检索引擎，可以结合 MySQL、PostgreSQL 做全文搜索，它可以提供比数据库本身更专业的搜索功能，使得应用程序更容易实现专业化的全文检索。

    1 引用 • 191 关注
  • BAE

    百度应用引擎（Baidu App Engine）提供了 PHP、Java、Python 的执行环境，以及云存储、消息服务、云数据库等全面的云服务。它可以让开发者实现自动地部署和管理应用，并且提供动态扩容和负载均衡的运行环境，让开发者不用考虑高成本的运维工作，只需专注于业务逻辑，大大降低了开发者学习和迁移的成本。

    19 引用 • 75 回帖 • 618 关注
  • API

    应用程序编程接口（Application Programming Interface）是一些预先定义的函数，目的是提供应用程序与开发人员基于某软件或硬件得以访问一组例程的能力，而又无需访问源码，或理解内部工作机制的细节。

    76 引用 • 429 回帖
  • jsoup

    jsoup 是一款 Java 的 HTML 解析器，可直接解析某个 URL 地址、HTML 文本内容。它提供了一套非常省力的 API，可通过 DOM，CSS 以及类似于 jQuery 的操作方法来取出和操作数据。

    6 引用 • 1 回帖 • 476 关注
  • Spring

    Spring 是一个开源框架，是于 2003 年兴起的一个轻量级的 Java 开发框架，由 Rod Johnson 在其著作《Expert One-On-One J2EE Development and Design》中阐述的部分理念和原型衍生而来。它是为了解决企业应用开发的复杂性而创建的。框架的主要优势之一就是其分层架构，分层架构允许使用者选择使用哪一个组件，同时为 JavaEE 应用程序开发提供集成的框架。

    942 引用 • 1458 回帖 • 118 关注
  • 前端

    前端技术一般分为前端设计和前端开发，前端设计可以理解为网站的视觉设计，前端开发则是网站的前台代码实现，包括 HTML、CSS 以及 JavaScript 等。

    247 引用 • 1347 回帖
  • Hprose

    Hprose 是一款先进的轻量级、跨语言、跨平台、无侵入式、高性能动态远程对象调用引擎库。它不仅简单易用，而且功能强大。你无需专门学习，只需看上几眼，就能用它轻松构建分布式应用系统。

    9 引用 • 17 回帖 • 610 关注
  • 锤子科技

    锤子科技（Smartisan）成立于 2012 年 5 月，是一家制造移动互联网终端设备的公司，公司的使命是用完美主义的工匠精神，打造用户体验一流的数码消费类产品（智能手机为主），改善人们的生活质量。

    4 引用 • 31 回帖 • 4 关注
  • 又拍云

    又拍云是国内领先的 CDN 服务提供商,国家工信部认证通过的“可信云”，乌云众测平台认证的“安全云”，为移动时代的创业者提供新一代的 CDN 加速服务。

    21 引用 • 37 回帖 • 519 关注
  • 单点登录

    单点登录（Single Sign On）是目前比较流行的企业业务整合的解决方案之一。SSO 的定义是在多个应用系统中，用户只需要登录一次就可以访问所有相互信任的应用系统。

    9 引用 • 25 回帖
  • FreeMarker

    FreeMarker 是一款好用且功能强大的 Java 模版引擎。

    23 引用 • 20 回帖 • 437 关注
  • 服务

    提供一个服务绝不仅仅是简单的把硬件和软件累加在一起，它包括了服务的可靠性、服务的标准化、以及对服务的监控、维护、技术支持等。

    41 引用 • 24 回帖 • 10 关注
  • HBase

    HBase 是一个分布式的、面向列的开源数据库，该技术来源于 Fay Chang 所撰写的 Google 论文 “Bigtable：一个结构化数据的分布式存储系统”。就像 Bigtable 利用了 Google 文件系统所提供的分布式数据存储一样，HBase 在 Hadoop 之上提供了类似于 Bigtable 的能力。

    17 引用 • 6 回帖 • 58 关注
  • Swagger

    Swagger 是一款非常流行的 API 开发工具，它遵循 OpenAPI Specification（这是一种通用的、和编程语言无关的 API 描述规范）。Swagger 贯穿整个 API 生命周期，如 API 的设计、编写文档、测试和部署。

    26 引用 • 35 回帖 • 12 关注
  • BookxNote

    BookxNote 是一款全新的电子书学习工具，助力您的学习与思考，让您的大脑更高效的记忆。

    笔记整理交给我，一心只读圣贤书。

    1 引用 • 1 回帖 • 2 关注
  • Spark

    Spark 是 UC Berkeley AMP lab 所开源的类 Hadoop MapReduce 的通用并行框架。Spark 拥有 Hadoop MapReduce 所具有的优点；但不同于 MapReduce 的是 Job 中间输出结果可以保存在内存中，从而不再需要读写 HDFS，因此 Spark 能更好地适用于数据挖掘与机器学习等需要迭代的 MapReduce 的算法。

    74 引用 • 46 回帖 • 561 关注
  • Android

    Android 是一种以 Linux 为基础的开放源码操作系统，主要使用于便携设备。2005 年由 Google 收购注资，并拉拢多家制造商组成开放手机联盟开发改良，逐渐扩展到到平板电脑及其他领域上。

    334 引用 • 323 回帖 • 25 关注
  • QQ

    1999 年 2 月腾讯正式推出“腾讯 QQ”，在线用户由 1999 年的 2 人（马化腾和张志东）到现在已经发展到上亿用户了，在线人数超过一亿，是目前使用最广泛的聊天软件之一。

    45 引用 • 557 回帖 • 181 关注
  • LaTeX

    LaTeX（音译“拉泰赫”）是一种基于 ΤΕΧ 的排版系统，由美国计算机学家莱斯利·兰伯特（Leslie Lamport）在 20 世纪 80 年代初期开发，利用这种格式，即使使用者没有排版和程序设计的知识也可以充分发挥由 TeX 所提供的强大功能，能在几天，甚至几小时内生成很多具有书籍质量的印刷品。对于生成复杂表格和数学公式，这一点表现得尤为突出。因此它非常适用于生成高印刷质量的科技和数学类文档。

    9 引用 • 32 回帖 • 152 关注
  • Telegram

    Telegram 是一个非盈利性、基于云端的即时消息服务。它提供了支持各大操作系统平台的开源的客户端，也提供了很多强大的 APIs 给开发者创建自己的客户端和机器人。

    5 引用 • 35 回帖 • 1 关注
  • V2Ray
    1 引用 • 15 回帖 • 1 关注
  • Kubernetes

    Kubernetes 是 Google 开源的一个容器编排引擎，它支持自动化部署、大规模可伸缩、应用容器化管理。

    109 引用 • 54 回帖
  • 坑

    一些有用的避坑指南。

    69 引用 • 93 回帖
  • App

    App（应用程序，Application 的缩写）一般指手机软件。

    90 引用 • 383 回帖
  • 尊园地产

    昆明尊园房地产经纪有限公司，即：Kunming Zunyuan Property Agency Company Limited（简称“尊园地产”）于 2007 年 6 月开始筹备，2007 年 8 月 18 日正式成立，注册资本 200 万元，公司性质为股份经纪有限公司，主营业务为：代租、代售、代办产权过户、办理银行按揭、担保、抵押、评估等。

    1 引用 • 22 回帖 • 703 关注

最新标签