IO、NIO、Netty分别实现服务端与客户端通信

>>强大,10k+点赞的 SpringBoot 后台管理系统竟然出了详细教程!

点击关注公众号,利用碎片时间学习

简化下场景:客户端每隔两秒发送一个带有时间戳的 “hello world” 给服务端,服务端收到之后打印。

IO编程

IO服务端:
public class IOServer {
    public static void main(String[] args) throws Exception {

        ServerSocket serverSocket = new ServerSocket(8000);

        // (1) 接收新连接线程
        new Thread(() -> {
            while (true) {
                try {
                    // (1) 阻塞方法获取新的连接
                    Socket socket = serverSocket.accept();

                    // (2) 每一个新的连接都创建一个线程,负责读取数据
                    new Thread(() -> {
                        try {
                            int len;
                            byte[] data = new byte[1024];
                            InputStream inputStream = socket.getInputStream();
                            // (3) 按字节流方式读取数据
                            while ((len = inputStream.read(data)) != -1) {
                                System.out.println(new String(data, 0, len));
                            }
                        } catch (IOException e) {
                        }
                    }).start();

                } catch (IOException e) {
                }

            }
        }).start();
    }
}
  • Server 端首先创建了一个serverSocket来监听 8000 端口,然后创建一个线程,线程里面不断调用阻塞方法 serversocket.accept();获取新的连接;
  • 当获取到新的连接之后,给每条连接创建一个新的线程,这个线程负责从该连接中读取数据;
  • 然后读取数据是以字节流的方式。
IO客户端:
public class IOClient {

    public static void main(String[] args) {
        new Thread(() -> {
            try {
                Socket socket = new Socket("127.0.0.1"8000);
                while (true) {
                    try {
                        socket.getOutputStream().write((new Date() + ": hello world").getBytes());
                        Thread.sleep(2000);
                    } catch (Exception e) {
                    }
                }
            } catch (IOException e) {
            }
        }).start();
    }
}
总结:

通过IO实现的服务端与客户端之间的通讯非常简单,但是存在的问题也非常多:

  • 线程资源受限:线程是操作系统中非常宝贵的资源,同一时刻有大量的线程处于阻塞状态是非常严重的资源浪费,操作系统耗不起。
  • 线程切换效率低下:单机 CPU 核数固定,线程爆炸之后操作系统频繁进行线程切换,应用性能急剧下降。
  • 数据读写是以字节流为单位。

NIO 编程

解决线程资源受限

NIO 编程模型中,新来一个连接不再创建一个新的线程,而是可以把这条连接直接绑定到某个固定的线程,然后这条连接所有的读写都由这个线程来负责。

IO、NIO、Netty分别实现服务端与客户端通信

如上图所示,IO 模型中,一个连接来了,会创建一个线程,对应一个 while 死循环,死循环的目的就是不断监测这条连接上是否有数据可以读,大多数情况下,1w 个连接里面同一时刻只有少量的连接有数据可读,因此,很多个 while 死循环都白白浪费掉了,因为读不出啥数据。

而在 NIO 模型中,他把这么多 while 死循环变成一个死循环,这个死循环由一个线程控制。

也就是在IO中,每一个连接都需要有专门的线程进行服务,而NIO是通过一个线程进行批量轮询获取连接数据得,消耗的线程资源大幅减少。NIO 模型中线程数量大大降低,线程切换效率因此也大幅度提高。

面向Buffer读写替代面向流操作

IO 读写是面向流的,一次性只能从流中读取一个或者多个字节,并且读完之后流无法再读取,你需要自己缓存数据。而 NIO 的读写是面向 Buffer 的,你可以随意读取里面任何一个字节数据,不需要你自己缓存数据,这一切只需要移动读写指针即可。

NIO服务端实现
public class NIOServer {
    public static void main(String[] args) throws IOException {
        Selector serverSelector = Selector.open();
        Selector clientSelector = Selector.open();

        new Thread(() -> {
            try {
                // 对应IO编程中服务端启动
                ServerSocketChannel listenerChannel = ServerSocketChannel.open();
                listenerChannel.socket().bind(new InetSocketAddress(8000));
                listenerChannel.configureBlocking(false);
                listenerChannel.register(serverSelector, SelectionKey.OP_ACCEPT);

                while (true) {
                    // 监测是否有新的连接,这里的1指的是阻塞的时间为 1ms
                    if (serverSelector.select(1) > 0) {
                        Set<SelectionKey> set = serverSelector.selectedKeys();
                        Iterator<SelectionKey> keyIterator = set.iterator();

                        while (keyIterator.hasNext()) {
                            SelectionKey key = keyIterator.next();

                            if (key.isAcceptable()) {
                                try {
                                    // (1) 每来一个新连接,不需要创建一个线程,而是直接注册到clientSelector
                                    SocketChannel clientChannel = ((ServerSocketChannel) key.channel()).accept();
                                    clientChannel.configureBlocking(false);
                                    clientChannel.register(clientSelector, SelectionKey.OP_READ);
                                } finally {
                                    keyIterator.remove();
                                }
                            }

                        }
                    }
                }
            } catch (IOException ignored) {
            }

        }).start();


        new Thread(() -> {
            try {
                while (true) {
                    // (2) 批量轮询是否有哪些连接有数据可读,这里的1指的是阻塞的时间为 1ms
                    if (clientSelector.select(1) > 0) {
                        Set<SelectionKey> set = clientSelector.selectedKeys();
                        Iterator<SelectionKey> keyIterator = set.iterator();

                        while (keyIterator.hasNext()) {
                            SelectionKey key = keyIterator.next();

                            if (key.isReadable()) {
                                try {
                                    SocketChannel clientChannel = (SocketChannel) key.channel();
                                    ByteBuffer byteBuffer = ByteBuffer.allocate(1024);
                                    // (3) 面向 Buffer
                                    clientChannel.read(byteBuffer);
                                    byteBuffer.flip();
                                    System.out.println(Charset.defaultCharset().newDecoder().decode(byteBuffer)
                                            .toString());
                                } finally {
                                    keyIterator.remove();
                                    key.interestOps(SelectionKey.OP_READ);
                                }
                            }

                        }
                    }
                }
            } catch (IOException ignored) {
            }
        }).start();


    }
}
  • NIO 模型中通常会有两个线程,每个线程绑定一个轮询器 selector ,在我们这个例子中serverSelector负责轮询是否有新的连接,clientSelector负责轮询连接是否有数据可读
  • 服务端监测到新的连接之后,不再创建一个新的线程,而是直接将新连接绑定到clientSelector上;
  • clientSelector被一个 while 死循环包裹着,如果在某一时刻有多条连接有数据可读,那么通过 clientSelector.select(1)方法可以轮询出来,进而批量处理;
  • 数据的读写面向 Buffer。
总结:

强烈不建议直接基于JDK原生NIO来进行网络开发:

  • JDK 的 NIO 编程需要了解很多的概念,编程复杂,对 NIO 入门非常不友好,编程模型不友好,ByteBuffer 的 Api 简直反人类;
  • 对 NIO 编程来说,一个比较合适的线程模型能充分发挥它的优势,而 JDK 没有给你实现,你需要自己实现,就连简单的自定义协议拆包都要你自己实现;
  • JDK 的 NIO 底层由 epoll 实现,该实现饱受诟病的空轮询 bug 会导致 cpu 飙升 100%;
  • 项目庞大之后,自行实现的 NIO 很容易出现各类 bug,维护成本较高,上面这一坨代码我都不能保证没有 bug。

正因为如此,NIO的客户端代码就不实现了,你可以直接使用IOClient.java与NIOServer.java通信进行测试。

Netty编程

Netty 封装了 JDK 的 NIO,让你用得更爽,你不用再写一大堆复杂的代码了。

Netty 是一个异步事件驱动的网络应用框架,用于快速开发可维护的高性能服务器和客户端。

使用 Netty 而不使用 JDK 原生 NIO 的原因

  • 使用 JDK 自带的NIO需要了解太多的概念,编程复杂,一不小心 bug 横飞;
  • Netty 底层 IO 模型随意切换,而这一切只需要做微小的改动,改改参数,Netty可以直接从 NIO 模型变身为 IO 模型;
  • Netty 自带的拆包解包,异常检测等机制让你从NIO的繁重细节中脱离出来,让你只需要关心业务逻辑;
  • Netty 解决了 JDK 的很多包括空轮询在内的 Bug;
  • Netty 底层对线程,selector 做了很多细小的优化,精心设计的 reactor 线程模型做到非常高效的并发处理;
  • 自带各种协议栈让你处理任何一种通用协议都几乎不用亲自动手;
  • Netty 社区活跃,遇到问题随时邮件列表或者 issue;
  • Netty 已经历各大 RPC 框架,消息中间件,分布式通信中间件线上的广泛验证,健壮性无比强大;
服务端实现

引入依赖:

 <dependency>
        <groupId>io.netty</groupId>
        <artifactId>netty-all</artifactId>
        <version>4.1.6.Final</version>
 </dependency>

NettyServer.java

public class NettyServer {
    public static void main(String[] args) {
        ServerBootstrap serverBootstrap = new ServerBootstrap();

        NioEventLoopGroup boss = new NioEventLoopGroup();
        NioEventLoopGroup worker = new NioEventLoopGroup();
        serverBootstrap
                .group(boss, worker)
                .channel(NioServerSocketChannel.class)
                .childHandler(new ChannelInitializer<NioSocketChannel>() 
{
                    @Override
                    protected void initChannel(NioSocketChannel ch) {
                        ch.pipeline().addLast(new StringDecoder());
                        ch.pipeline().addLast(new SimpleChannelInboundHandler<String>() {
                            @Override
                            protected void channelRead0(ChannelHandlerContext ctx, String msg) {
                                System.out.println(msg);
                            }
                        });
                    }
                })
                .bind(8000);
    }
  • 首先看到,我们创建了两个NioEventLoopGroup,这两个对象可以看做是传统IO编程模型的两大线程组,bossGroup表示监听端口,accept 新连接的线程组,workerGroup表示处理每一条连接的数据读写的线程组。
  • 我们创建了一个引导类 ServerBootstrap,这个类将引导我们进行服务端的启动工作。
  • 通过.group(bossGroup, workerGroup)给引导类配置两大线程组,这个引导类的线程模型也就定型了。
  • 通过.channel(NioServerSocketChannel.class)来指定 IO 模型为NIO,当然,这里也有其他的选择,如果你想指定 IO 模型为 BIO,那么这里配置上OioServerSocketChannel.class类型即可,当然通常我们也不会这么做,因为Netty的优势就在于NIO。
  • 调用childHandler()方法,给这个引导类创建一个ChannelInitializer,这里主要就是定义后续每条连接的数据读写,业务处理逻辑。

总结一下就是,要启动一个Netty服务端,必须要指定三类属性,分别是线程模型、IO 模型、连接读写处理逻辑,有了这三者,之后在调用bind(8000),我们就可以在本地绑定一个 8000 端口启动起来。

客户端实现

对于客户端,,依然需要线程模型、IO 模型,以及 IO 业务处理逻辑三大参数,如下:

public class NettyClient {
    public static void main(String[] args) throws InterruptedException {
        Bootstrap bootstrap = new Bootstrap();
        NioEventLoopGroup group = new NioEventLoopGroup();

        bootstrap.group(group)// 指定线程模型
                .channel(NioSocketChannel.class) // 指定 IO 类型为 NIO
                .handler(new ChannelInitializer<Channel>() 
// IO 处理逻辑
                    @Override
                    protected void initChannel(Channel ch) {
                        ch.pipeline().addLast(new StringEncoder());
                    }
                });

        //建立连接
        Channel channel = bootstrap.connect("127.0.0.1"8000).addListener(future -> {
            if (future.isSuccess()) {
                System.out.println("连接成功!");
            } else {
                System.err.println("连接失败!");
                //重新连接
                bootstrap.connect("127.0.0.1"8000);

            }

        }).channel();

        while (true) {
            channel.writeAndFlush(new Date() + ": hello world!");
            Thread.sleep(2000);
        }
    }
}
  • 客户端启动的引导类是 Bootstrap,负责启动客户端以及连接服务端;而上一小节我们在描述服务端的启动的时候,这个辅导类是 ServerBootstrap
  • 通过bootstrap.group(group)指定线程模型;
  • 我们指定 IO 模型为 NioSocketChannel,表示 IO 模型为 NIO。也可以设置 IO 模型为 OioSocketChannel,但是通常不会这么做,因为 Netty 的优势在于 NIO;
  • 给引导类指定一个 handler,这里主要就是定义连接的业务处理逻辑;
  • 配置完线程模型、IO 模型、业务处理逻辑之后,调用 connect 方法进行连接,可以看到 connect 方法有两个参数,第一个参数可以填写 IP 或者域名,第二个参数填写的是端口号,由于 connect 方法返回的是一个 Future,也就是说这个方是异步的,我们通过 addListener 方法可以监听到连接是否成功,进而打印出连接信息。
总结

创建一个引导类,然后给他指定线程模型,IO 模型,连接读写处理逻辑,连接上特定主机和端口,客户端就启动起来了。connect 方法是异步的,我们可以通过这个异步回调机制来实现指数退避重连逻辑。

来源:wangcw.blog.csdn.net/article/details/108243295

推荐:

主流Java进阶技术(学习资料分享)

IO、NIO、Netty分别实现服务端与客户端通信
PS:因为公众号平台更改了推送规则,如果不想错过内容,记得读完点一下“在看”,加个“星标”,这样每次新文章推送才会第一时间出现在你的订阅列表里。“在看”支持我们吧!

原文始发于微信公众号(Java笔记虾):IO、NIO、Netty分别实现服务端与客户端通信