登录 注册

3. Netty 初认识 --NIO 入门 --- 示例 (完整代码)

本贴最后更新于 1518 天前，其中的信息可能已经事过景迁

首先启动 TimeServer 服务端开始监听
其次启动 TimeClient 连接服务端,并发送固定指令 QUERY TIME ORDER 查询当前时间
最后服务端返回当前时间

具体代码如下(可运行)

TimeServer.java

package club.wujingjian.com.wujingjian.nio.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();
            }
        }
        MultiplexerTimeServer timeServer = new MultiplexerTimeServer(port);

        new Thread(timeServer, "NIO-MultiplexerTimeServer-001").start();
    }
}

MultiplexerTimeServer.java

package club.wujingjian.com.wujingjian.nio.server;

import club.wujingjian.com.wujingjian.nio.Util;

import java.io.IOError;
import java.io.IOException;
import java.net.InetSocketAddress;
import java.nio.ByteBuffer;
import java.nio.channels.SelectionKey;
import java.nio.channels.Selector;
import java.nio.channels.ServerSocketChannel;
import java.nio.channels.SocketChannel;
import java.util.Date;
import java.util.Iterator;
import java.util.Set;

public class MultiplexerTimeServer implements  Runnable {

    private Selector selector;

    private ServerSocketChannel servChannel;

    private volatile  boolean stop;

    /**
     * 初始化多路复用器、绑定监听端口
     * 进行资源初始化,创建多路复用器Selector、ServerSocketChannel,
     * 对Channel和TCP参数进行配置,（例如,将ServerSocketChannel设置为异步非阻塞模式,它的backlog设置为1024）
     *
     */
    public MultiplexerTimeServer(int port) {
        try {
            selector = Selector.open();
            servChannel = ServerSocketChannel.open();
            servChannel.configureBlocking(false);
            servChannel.socket().bind(new InetSocketAddress(port), 1024);
            //将ServerSocketChannel注册到Selector,监听SelectionKey.OP_ACCEPT操作位
            servChannel.register(selector, SelectionKey.OP_ACCEPT);
            System.out.println(Util.nowDate() + "The time server is start in port : " + port);
        } catch (IOException e) {
            e.printStackTrace();;
            System.exit(1);
        }
    }

    public void stop(){
        this.stop = true;
    }
    @Override
    public void run() {
        while (!stop) {
            /**
             * 循环遍历selector,休眠时间为1s
             */
            try {
                selector.select(1000);
                Set<SelectionKey> selectedKeys = selector.selectedKeys();
                Iterator<SelectionKey> it = selectedKeys.iterator();
                SelectionKey key = null;
                while (it.hasNext()) {
                    key = it.next();
                    it.remove();
                    try {
                        handleInput(key);
                    } catch (Exception e) {
                        key.cancel();
                        if (key.channel() != null) {
                            key.channel().close();
                        }
                    }
                }
            } catch (Throwable throwable) {
                throwable.printStackTrace();
            }
        }

        //多路复用器关闭后,所有注册在上面的channel 和pipe 等资源都会被自动去注册并关闭,所以不需要重复释放资源
        if (selector != null) {
            try {
                selector.close();
            } catch (IOException e) {
                e.printStackTrace();
            }
        }
    }



    private void handleInput(SelectionKey key) throws  IOException {
        if (key.isValid()) {
            /**
             *处理新接入的客户端请求信息,根据SelectionKey的操作位进行判断即可获知网络事件的类型
             */
            if (key.isAcceptable()) {
                //接收新连接
                ServerSocketChannel ssc = (ServerSocketChannel) key.channel();
                //通过ServerSocketChannel的accept()接收客户端的连接请求并创建SocketChannel实例.相当于三次握手
                SocketChannel sc = ssc.accept();
                sc.configureBlocking(false);
                //添加新连接到多路复用器Selector
                sc.register(selector, SelectionKey.OP_READ);
            }
            if (key.isReadable()) {
                /**
                 * 读取客户端的请求消息,首先创建一个ByteBuffer,由于事先无法得知客户端发送的码流大小,作为例程，我们开辟一个1K的缓冲区，
                 * 然后调用SocketChannel的read方法读取请求码流.
                 * 注意: 由于我们已经将SocketChannel设置为异步非阻塞模式,因此他的read是非阻塞的,  socketChannel.read()的返回值有以下三种情况
                 *     返回值大于0:  读取到了字节,对字节进行编解码.
                 *     返回值等于0: 没有读取到字节,属于正常场景,忽略
                 *     返回值为-1 : 链路已经关闭,需要关闭SocketChannel,释放资源
                 */
                SocketChannel sc = (SocketChannel) key.channel();
                ByteBuffer readBuffer = ByteBuffer.allocate(1024);
                int readBytes = sc.read(readBuffer);
                if (readBytes > 0) {
                    //读取到码流以后,我们进行解码,首先对readBuffer进行flip操作,作用是将缓冲区当前的limit设置为position,position设置为0,用于后续对缓冲区的读取操作
                    readBuffer.flip();
                    //根据缓冲区可读的字节个数创建字节数组
                    byte[] bytes = new byte[readBuffer.remaining()];
                    //将缓冲区可读的字节数组复制到新创建的字节数组bytes中.
                    readBuffer.get(bytes);
                    String body = new String(bytes, "UTF-8");
                    System.out.println(Util.nowDate() + "The time server receive order : " + body);
                    //如果请求指令是"QUERY TIME ORDER" 则把服务器的当前时间编码后返回给客户端
                    String currentTime = "QUERY TIME ORDER".equalsIgnoreCase(body) ? new Date(System.currentTimeMillis()).toString() : "BAD ORDER";
                    System.out.println(Util.nowDate() + "服务器准备返回时间为 :" + currentTime);
                    doWrite(sc, currentTime);
                } else if (readBytes < 0) {
                    //对端链路关闭
                    key.cancel();
                    sc.close();
                } else{
                    // 读取到0字节,忽略
                }

            }
        }
    }

    private void doWrite(SocketChannel channel, String response) throws  IOException{
        if (response != null && response.trim().length() > 0) {
            //将字符串编码成字节数组,
            byte [] bytes = response.getBytes();
            //根据字节数组容量创建ByteBuffer
            ByteBuffer writeBuffer = ByteBuffer.allocate(bytes.length);
            //将字节数组复制到缓冲区中
            writeBuffer.put(bytes);
            //将缓冲区的position设置为0,以便准备输出
            writeBuffer.flip();
            /**
             *将缓冲区中的字节数组发送出去
             * 注意:    由于SocketChannel是异步非阻塞的,它并不能保证一次性把需要发送的字节数组发送完毕,此时会出现"写半包"的问题,
             *      我们需要注册写操作,不断轮询Selector将没有发送完的ByteBuffer发送完毕,可以通过ByteBuffer的hasRemaining()方法
             *      判断消息是否发送完成。此处仅仅是个入门示例,没有处理"写半包"的情况.
             */
            channel.write(writeBuffer);
            if (!writeBuffer.hasRemaining()) {
                System.out.println(Util.nowDate() + "发送成功");
            }
        }
    }
}

下面是客户端代码

TimeClient.java

package club.wujingjian.com.wujingjian.nio.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 TimeClientHandle("127.0.0.1",port), "TimeClient-001").start();
    }
}

TimeClientHandle.java

package club.wujingjian.com.wujingjian.nio.client;

import club.wujingjian.com.wujingjian.nio.Util;

import java.io.IOException;
import java.net.InetSocketAddress;
import java.nio.ByteBuffer;
import java.nio.channels.SelectionKey;
import java.nio.channels.Selector;
import java.nio.channels.SocketChannel;
import java.util.Iterator;
import java.util.Set;

public class TimeClientHandle implements Runnable {

    private String host;
    private int port ;
    private Selector selector;
    private SocketChannel socketChannel;
    private volatile boolean stop;

    /**
     * 构造函数初始化NIO多路复用器和SocketChannel对象，并将socketChannel设置为异步非阻塞模式
     * @param host
     * @param port
     */
    public TimeClientHandle(String host, int port) {
        this.host = host == null ?"127.0.0.1": host;
        this.port = port;
        try {
            selector = Selector.open();
            socketChannel = SocketChannel.open();
            socketChannel.configureBlocking(false);
        } catch (IOException e) {
            e.printStackTrace();
            System.exit(1);
        }
    }

    @Override
    public void run() {
        try {
            //发送连接请求,作为示例,连接是成功的,所以不需要做重连操作,实际代码中,可能要考虑重连,不能放到while循环外面
            doConnect();
        } catch (IOException e) {
            e.printStackTrace();
            System.exit(1);
        }
        while (!stop) {
            try {
                selector.select(1000);
                Set<SelectionKey> selectedKeys = selector.selectedKeys();
                Iterator<SelectionKey> it = selectedKeys.iterator();
                SelectionKey key = null;
                //当有就绪的Channel时候执行handleInput方法
                while (it.hasNext()) {
                    key = it.next();
                    it.remove();
                    try {
                        handleInput(key);
                    } catch (Exception e) {
                        if (key != null) {
                            key.cancel();
                            if (key.channel() != null) {
                                key.channel().close();
                            }
                        }
                    }
                }
            } catch (Exception e) {
                e.printStackTrace();
                System.exit(1);
            }
        }

        //多路复用器关闭后,所有注册在上面的channel和pipe等资源都会被自动去注册并关闭,所以不需要重复释放资源
        if (selector != null) {
            try {
                selector.close();
            } catch (IOException e) {
                e.printStackTrace();
            }
        }
    }

    private void handleInput(SelectionKey key) throws  IOException {
        if (key.isValid()) {
            //判断是否连接成功
            SocketChannel sc = (SocketChannel) key.channel();
            if (key.isConnectable()) { //如果SelectionKey处于连接状态,说明服务端已经返回ACK应答消息,这是我们需要对结果进行判断
                if (sc.finishConnect()) { //如果sc.finishConnect()返回true,说明客户端连接成功,
                    System.out.println(Util.nowDate() + "客户端连接成功");
                    sc.register(selector, SelectionKey.OP_READ); //将SocketChannel注册到多路复用器上,注册SelectionKey.OP_READ操作位,监听网络读操作,
                    doWrite(sc); //然后发送请求消息给服务端
                } else {
                    System.out.println("连接失败,进程退出");
                    System.exit(1); //连接失败,进程退出
                }
            }

            if (key.isReadable()) {
                ByteBuffer readBuffer = ByteBuffer.allocate(1024);
                int readBytes = sc.read(readBuffer);
                if (readBytes > 0) {
                    readBuffer.flip();
                    byte[] bytes = new byte[readBuffer.remaining()];
                    readBuffer.get(bytes);
                    String body = new String(bytes, "UTF-8");
                    System.out.println(Util.nowDate() + "Now is :" + body);
                    this.stop = true;
                } else if (readBytes < 0) {
                    //对端链路关闭
                    key.cancel();
                    sc.close();
                } else {
                    //读到0 字节,忽略
                }
            }
        }
    }

    private void doConnect() throws  IOException {
        //如果直连成功,则注册到多路复用器上,发送请求消息,读应答
        if (socketChannel.connect(new InetSocketAddress(host, port))) {
            socketChannel.register(selector, SelectionKey.OP_READ);
            doWrite(socketChannel);
        } else {
            /**
             * 如果没有直接连接成功,则说明服务端没有返回TPC握手应答消息,担这并不代表连接失败.
             * 我们需要将SocketChannel注册到多路复用器Selector上,注册SelectionKey.OP_CONNECT，
             * 当服务端返回TCP syn-ack消息后,Selector就能够轮询到这个SocketChannel处于连接就绪状态
             */
            socketChannel.register(selector, SelectionKey.OP_CONNECT);
        }
    }

    private void doWrite(SocketChannel sc) throws IOException {
        byte[] req = "QUERY TIME ORDER".getBytes();
        ByteBuffer writeBuffer = ByteBuffer.allocate(req.length);
        writeBuffer.put(req);
        writeBuffer.flip();
        sc.write(writeBuffer); //此时可能会存在"半包写"问题,实际工作中要考虑
        if (!writeBuffer.hasRemaining()) { //此时如果缓冲区中消息全部发送完成，打印如下消息Send order 2 server succeed.
            System.out.println(Util.nowDate() + "Send order 2 server succeed.");
        }
    }
}

先允许服务器端，后允许客户端代码最后输出分别为:

image.png

image.png

  • Netty

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

    49 引用 • 33 回帖 • 21 关注
北京 位置
393 12 143 31 38 14 19 131

相关帖子

回帖
3. Netty 初认识 --NIO 入门 --- 示例 (完整代码)

欢迎来到这里！

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

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

近期热议

推荐标签 标签

  • InfluxDB

    InfluxDB 是一个开源的没有外部依赖的时间序列数据库。适用于记录度量，事件及实时分析。

    2 引用 • 55 关注
  • 房星科技

    房星网，我们不和没有钱的程序员谈理想，我们要让程序员又有理想又有钱。我们有雄厚的房地产行业线下资源，遍布昆明全城的 100 家门店、四千地产经纪人是我们坚实的后盾。

    6 引用 • 141 回帖 • 559 关注
  • Sandbox

    如果帖子标签含有 Sandbox ，则该帖子会被视为“测试帖”，主要用于测试社区功能，排查 bug 等，该标签下内容不定期进行清理。

    370 引用 • 1215 回帖 • 581 关注
  • SMTP

    SMTP（Simple Mail Transfer Protocol）即简单邮件传输协议，它是一组用于由源地址到目的地址传送邮件的规则，由它来控制信件的中转方式。SMTP 协议属于 TCP/IP 协议簇，它帮助每台计算机在发送或中转信件时找到下一个目的地。

    4 引用 • 18 回帖 • 588 关注
  • Vditor

    Vditor 是一款浏览器端的 Markdown 编辑器，支持所见即所得、即时渲染（类似 Typora）和分屏预览模式。它使用 TypeScript 实现，支持原生 JavaScript、Vue、React 和 Angular。

    313 引用 • 1666 回帖 • 1 关注
  • H2

    H2 是一个开源的嵌入式数据库引擎，采用 Java 语言编写，不受平台的限制，同时 H2 提供了一个十分方便的 web 控制台用于操作和管理数据库内容。H2 还提供兼容模式，可以兼容一些主流的数据库，因此采用 H2 作为开发期的数据库非常方便。

    11 引用 • 54 回帖 • 641 关注
  • SOHO

    为成为自由职业者在家办公而努力吧！

    7 引用 • 55 回帖 • 93 关注
  • IDEA

    IDEA 全称 IntelliJ IDEA，是一款 Java 语言开发的集成环境，在业界被公认为最好的 Java 开发工具之一。IDEA 是 JetBrains 公司的产品，这家公司总部位于捷克共和国的首都布拉格，开发人员以严谨著称的东欧程序员为主。

    180 引用 • 400 回帖
  • Ant-Design

    Ant Design 是服务于企业级产品的设计体系，基于确定和自然的设计价值观上的模块化解决方案，让设计者和开发者专注于更好的用户体验。

    17 引用 • 23 回帖 • 1 关注
  • CSDN

    CSDN (Chinese Software Developer Network) 创立于 1999 年，是中国的 IT 社区和服务平台，为中国的软件开发者和 IT 从业者提供知识传播、职业发展、软件开发等全生命周期服务，满足他们在职业发展中学习及共享知识和信息、建立职业发展社交圈、通过软件开发实现技术商业化等刚性需求。

    14 引用 • 155 回帖
  • Bug

    Bug 本意是指臭虫、缺陷、损坏、犯贫、窃听器、小虫等。现在人们把在程序中一些缺陷或问题统称为 bug（漏洞）。

    77 引用 • 1741 回帖 • 1 关注
  • Google

    Google（Google Inc.，NASDAQ：GOOG）是一家美国上市公司（公有股份公司），于 1998 年 9 月 7 日以私有股份公司的形式创立，设计并管理一个互联网搜索引擎。Google 公司的总部称作“Googleplex”，它位于加利福尼亚山景城。Google 目前被公认为是全球规模最大的搜索引擎，它提供了简单易用的免费服务。不作恶（Don't be evil）是谷歌公司的一项非正式的公司口号。

    49 引用 • 192 回帖
  • 以太坊

    以太坊（Ethereum）并不是一个机构，而是一款能够在区块链上实现智能合约、开源的底层系统。以太坊是一个平台和一种编程语言 Solidity，使开发人员能够建立和发布下一代去中心化应用。 以太坊可以用来编程、分散、担保和交易任何事物：投票、域名、金融交易所、众筹、公司管理、合同和知识产权等等。

    34 引用 • 367 回帖 • 2 关注
  • jsoup

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

    6 引用 • 1 回帖 • 461 关注
  • 安全

    安全永远都不是一个小问题。

    189 引用 • 813 回帖
  • iOS

    iOS 是由苹果公司开发的移动操作系统，最早于 2007 年 1 月 9 日的 Macworld 大会上公布这个系统，最初是设计给 iPhone 使用的，后来陆续套用到 iPod touch、iPad 以及 Apple TV 等产品上。iOS 与苹果的 Mac OS X 操作系统一样，属于类 Unix 的商业操作系统。

    84 引用 • 139 回帖
  • Openfire

    Openfire 是开源的、基于可拓展通讯和表示协议 (XMPP)、采用 Java 编程语言开发的实时协作服务器。Openfire 的效率很高，单台服务器可支持上万并发用户。

    6 引用 • 7 回帖 • 89 关注
  • BAE

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

    19 引用 • 75 回帖 • 619 关注
  • Sublime

    Sublime Text 是一款可以用来写代码、写文章的文本编辑器。支持代码高亮、自动完成，还支持通过插件进行扩展。

    10 引用 • 5 回帖 • 2 关注
  • Sillot

    Sillot （汐洛）孵化自思源笔记，致力于服务智慧新彖乄，具有彖乄驱动、极致优雅、开发者友好的特点
    Github 地址：https://github.com/Hi-Windom/Sillot

    15 引用 • 6 回帖 • 28 关注
  • 笔记

    好记性不如烂笔头。

    303 引用 • 777 回帖
  • Netty

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

    49 引用 • 33 回帖 • 21 关注
  • PWL

    组织简介

    用爱发电 (Programming With Love) 是一个以开源精神为核心的民间开源爱好者技术组织，“用爱发电”象征开源与贡献精神，加入组织，代表你将遵守组织的“个人开源爱好者”的各项条款。申请加入：用爱发电组织邀请帖。
    用爱发电组织官网：https://programmingwithlove.stackoverflow.wiki/

    用爱发电组织的核心驱动力：

    • 遵守开源守则，体现开源&贡献精神：以分享为目的，拒绝非法牟利。
    • 自我保护：使用适当的 License 保护自己的原创作品。
    • 尊重他人：不以各种理由、各种漏洞进行未经允许的抄袭、散播、洩露；以礼相待，尊重所有对社区做出贡献的开发者；通过他人的分享习得知识，要留下足迹，表示感谢。
    • 热爱编程、热爱学习：加入组织，热爱编程是首当其要的。我们欢迎热爱讨论、分享、提问的朋友，也同样欢迎默默成就的朋友。
    • 倾听：正确并恳切对待、处理问题与建议，及时修复开源项目的 Bug ，及时与反馈者沟通。不抬杠、不无视、不辱骂。
    • 平视：不诋毁、轻视、嘲讽其他开发者，主动提出建议、施以帮助，以和谐为本。只要他人肯努力，你也可能会被昔日小看的人所超越，所以请保持谦虚。
    • 乐观且活跃：你的努力决定了你的高度。不要放弃，多年后回头俯瞰，才会发现自己已经成就往日所仰望的水平。积极地将项目开源，帮助他人学习、改进，自己也会获得相应的提升、成就与成就感。
    1 引用 • 487 回帖 • 7 关注
  • 单点登录

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

    9 引用 • 25 回帖 • 2 关注
  • HHKB

    HHKB 是富士通的 Happy Hacking 系列电容键盘。电容键盘即无接点静电电容式键盘（Capacitive Keyboard）。

    5 引用 • 74 回帖 • 407 关注
  • GitLab

    GitLab 是利用 Ruby 一个开源的版本管理系统，实现一个自托管的 Git 项目仓库，可通过 Web 界面操作公开或私有项目。

    46 引用 • 72 回帖
  • ZeroNet

    ZeroNet 是一个基于比特币加密技术和 BT 网络技术的去中心化的、开放开源的网络和交流系统。

    1 引用 • 21 回帖 • 591 关注

最新标签