Java NIO 和 Netty快速入门-Toy模板网

这篇具有很好参考价值的文章主要介绍了Java NIO 和 Netty快速入门。希望对大家有所帮助。如果存在错误或未考虑完全的地方，请大家不吝赐教，您也可以点击"举报违法"按钮提交疑问。

三大组件

1.Channel & Buffer

channel 是读写数据的双向通道，可以从 channel将数据读入buffer，也可以将buffer数据写入channel(较之前的stream要么是输入，要么是输出更为底层)
- 四种常见Channel:
  - FileChannel
  - DatagramChannel
  - SocketChannel
  - ServerSocketChannel
buffer用来缓冲读写数据
- 常见buffer：
  - ByteBuffer
    - MappedByteBuffer
    - DirectByteBuffer
    - HeapByteBuffer
  - ShortBuffer
  - IntBuffer
  - LongBuffer
  - FloatBuffer
  - DoubleBuffer
  - CharBuffer
Selector
- 多线程版本缺点：
  1. 内存占用高
  2. 线程上下文切换成本高
  3. 只适合连接少数场景
- 线程池版设计：
  1. 缺点：
    - 阻塞模式下，线程仅能处理一个 socket连接
    - 仅适合短链接场景
- selector 的作用就是配合一个线程来管理多个channel，获取这些 channel 上发生的事件，这些channel 工作在非阻塞模式下，不会让线程吊死在一个 channel上。适合连接数特别多，但是流量低的场景。
- 调用selector 的select() 会阻塞直到 channel 发生了读写就绪事件，这些事件发生，select方法就会返回这些事件，交给thread来处理

ByteBuffer 正确使用方式

向 buffer 写入数据，例如调用 channel.read(buffer)
调用 flip() 切换至读模式
从 buffer 读取数据，例如调用 buffer.get()
调用clear 或compact() 切换至写模式
重复 1 ~ 4 步骤

package org.example;


import lombok.extern.slf4j.Slf4j;

import java.io.FileInputStream;
import java.io.IOException;
import java.nio.ByteBuffer;
import java.nio.channels.FileChannel;

/**
 * Hello world!
 *
 */
 @Slf4j
public class App
{
    public static void main( String[] args )
    {
        // fileChannel
        try (FileChannel channel = new FileInputStream("data.txt").getChannel()) {
            // 准备缓冲区
            ByteBuffer buffer = ByteBuffer.allocate(10);
            while (true){
                // 读取数据，向缓冲区写入
                int len = channel.read(buffer);
                log.debug("读取到的字节数: {}", len);
                if (len == -1) break;
                // 打印buffer内容
                buffer.flip(); // 切换至读模式
                while (buffer.hasRemaining()){
                    // 检查是否有剩余
                    byte b = buffer.get();
                    log.debug("实际的字节: {}", (char) b);
                }
                // 切换为写模式
                buffer.clear();
            }
        } catch (IOException e) {
        }
    }
}

ByteBuffer 结构

主要由以下组成

capacity：容量
position：写入位置指针
limit：读取限制

Java NIO 和 Netty快速入门,nio,netty,java,nio,python,netty

使用flip()后，position切换为读取位置，limit切换为读取限制

Java NIO 和 Netty快速入门,nio,netty,java,nio,python,netty

clear，读模式切换到写模式

Java NIO 和 Netty快速入门,nio,netty,java,nio,python,netty

compact方法是把未读完的部分向前压缩，然后切换至写模式

Java NIO 和 Netty快速入门,nio,netty,java,nio,python,netty

package org.example;

import java.nio.ByteBuffer;

public class TestBytebuffer {
    public static void main(String[] args) {
        ByteBuffer allocate = ByteBuffer.allocate(10);
        allocate.put((byte) 0x61);
        allocate.flip();
        while (allocate.hasRemaining()) {
            byte b = allocate.get();
            System.out.print(b + " ");
        }
        System.out.println();
    }
}

三种String转化为ByteBuffer的方式

ByteBuffer buffer = ByteBuffer.allocate(16);
buffer.put("hello".getBytes());

ByteBuffer buffer2 = StandardCharesets.UTF_8.encode("hello");

ByteBuffer buffer3 = ByteBuffer.wrap("hello".getBytes);

对于后两者，会自动切换为读模式，所以可以通过decode()进行解码

String str1 = StandardCharsets.UTF_8.decode(buffer2).toString();

而对于第一个方法，需要先使用filp()先将其转化为读模式。

黏包半包问题：

hello\n
world\n
java\n

会转化成

hello\nworl // 黏包
d\njava\n //半包

FileChannel文件编程

Path

.代表当前路径
..表示上一级路径

Files

nio

阻塞模式：

服务端

package org.example.nio;

import java.io.IOException;
import java.net.InetSocketAddress;
import java.net.ServerSocket;
import java.nio.ByteBuffer;
import java.nio.channels.ServerSocketChannel;
import java.nio.channels.SocketChannel;
import java.util.ArrayList;

public class Server {
    public static void main(String[] args) throws IOException {
        ByteBuffer buffer = ByteBuffer.allocate(16);
        // 创建服务器
        ServerSocketChannel ssc = ServerSocketChannel.open();
        // 绑定监听端口
        ssc.bind(new InetSocketAddress(8080));
        ArrayList<SocketChannel> socketChannels = new ArrayList<>();
        while (true){
            // accept
            SocketChannel sc = ssc.accept();
            socketChannels.add(sc);
            // 接收客户端发送数据
            for (SocketChannel channel: socketChannels){
                channel.read(buffer);
                buffer.flip();
                while (buffer.hasRemaining()){
                    System.out.print((char) buffer.get());
                }
                buffer.clear();
            }
        }
    }
}

客户端

package org.example.nio;

import java.io.IOException;
import java.net.InetSocketAddress;
import java.nio.channels.SocketChannel;
import java.nio.charset.Charset;

public class Client {
    public static void main(String[] args) throws IOException {
        SocketChannel open = SocketChannel.open();
        open.connect(new InetSocketAddress("localhost", 8080));
        System.out.println("waiting...");
        open.write(Charset.defaultCharset().encode("hello nio!"));
    }
}

非阻塞

服务端

package org.example.nio;

import java.io.IOException;
import java.net.InetSocketAddress;
import java.net.ServerSocket;
import java.nio.ByteBuffer;
import java.nio.channels.ServerSocketChannel;
import java.nio.channels.SocketChannel;
import java.util.ArrayList;

public class Server {
    public static void main(String[] args) throws IOException {
        ByteBuffer buffer = ByteBuffer.allocate(16);
        // 创建服务器
        ServerSocketChannel ssc = ServerSocketChannel.open();
        ssc.configureBlocking(false);
        // 绑定监听端口
        ssc.bind(new InetSocketAddress(8080));
        ArrayList<SocketChannel> socketChannels = new ArrayList<>();
        while (true){
            // accept
            SocketChannel sc = ssc.accept();// 非阻塞没有拿到连接返回null
            if (sc != null){
                System.out.println("connected...");
                sc.configureBlocking(false);
                socketChannels.add(sc);
            }
            // 接收客户端发送数据
            if (!socketChannels.isEmpty()){
                for (SocketChannel channel: socketChannels){
                    int read = channel.read(buffer);
                    if (read > 0){
                        buffer.flip();
                        while (buffer.hasRemaining()){
                            System.out.print((char) buffer.get());
                        }
                        buffer.clear();
                    }
                }
            }
        }
    }
}

客户端

package org.example.nio;

import java.io.IOException;
import java.net.InetSocketAddress;
import java.nio.channels.SocketChannel;
import java.nio.charset.Charset;

public class Client {
    public static void main(String[] args) throws IOException {
        SocketChannel open = SocketChannel.open();
        open.connect(new InetSocketAddress("localhost", 8080));
        System.out.println("waiting...");
        open.write(Charset.defaultCharset().encode("hello nio!"));
    }
}

四种事件：

accept - 会在有连接请求时触发
connect - 是客户端，连接建立后触发
read - 可读事件
write- 可写事件

Selector

package org.example.nio;

import lombok.extern.slf4j.Slf4j;

import java.io.IOException;
import java.net.InetSocketAddress;
import java.net.ServerSocket;
import java.nio.ByteBuffer;
import java.nio.channels.*;
import java.util.ArrayList;
import java.util.Iterator;

@Slf4j
public class Server {
    public static void main(String[] args) throws IOException {

        Selector selector = Selector.open();


        ByteBuffer buffer = ByteBuffer.allocate(16);
        // 创建服务器
        ServerSocketChannel ssc = ServerSocketChannel.open();
        ssc.configureBlocking(false);

        // 注册
        SelectionKey sscKey = ssc.register(selector, 0, null);
        sscKey.interestOps(SelectionKey.OP_ACCEPT);// 只关注accept事件
        log.debug("register key {}", sscKey);
        // 绑定监听端口
        ssc.bind(new InetSocketAddress(8080));
        ArrayList<SocketChannel> socketChannels = new ArrayList<>();
        while (true){
            // select 阻塞直到四种线程之一发生
            selector.select();
            Iterator<SelectionKey> iterator = selector.selectedKeys().iterator();
            while (iterator.hasNext()) {
                SelectionKey key = iterator.next();
                log.debug("key: {}", key);
                ServerSocketChannel channel = (ServerSocketChannel) key.channel();
                SocketChannel sc = channel.accept();
                log.debug("{}", sc);
            }
        }
    }
}

由于selector的任务需要处理完成后才进入到新的select阻塞，所以如果没有处理任务，selector会重复迭代当前key，此时使用cancel()方法就可以完成。

Java NIO 和 Netty快速入门,nio,netty,java,nio,python,netty

处理完一个key，就移除一个key

处理消息边界

三种思路：

固定消息长度，客户端和服务端约定一个长度，数据包大小一样，服务器按照预定长度读取。浪费带宽
按分隔符拆分，效率低下
TLV模式，type，length，value

网络编程

多路复用：单线程可以配合 Selector 完成对多个 Channel 可读写事件的监控，称之为多路复用

阻塞：相关方法都会导致线程暂停（线程闲置）

非阻塞：相关方法不会让线程暂停，但是及时没有建立连接和可读数据，线程仍然在运行，浪费cpu

监听channel事件：

阻塞直到绑定事件发生
```
int count = selector.select();
```
阻塞直到绑定事件发生，或是超时(单位是ms)
```
int count = selector.select(long timeout);
```
不会阻塞，也就是不管有没有事件，立刻返回，自己根据返回值检查是否有事件
```
int count = selector.selectNow();
```

Select何时不阻塞

事件发生时
调用 selector.wakeup();
调用 selector.close();
selector 所在线程 interrupt

NIO vs BIO

stream vs channel

stream 不会自动缓冲数据，channel会利用系统提供的发送缓冲区、接收缓冲区
stream仅支持阻塞api，channel同时支持阻塞、非阻塞api，网络channel可配合selector实现多路复用
二者均为全双工，即读写可以同时进行

五种IO模型：

阻塞IO
非阻塞IO
多路复用
信号驱动
异步IO

Java NIO 和 Netty快速入门,nio,netty,java,nio,python,netty

同步：线程自己获取结果（一个线程）

异步：线程自己不去获取结果，而是由其他线程送结果（至少2个线程）

零拷贝

一次调用读写内部工作流程：

Java NIO 和 Netty快速入门,nio,netty,java,nio,python,netty

nio优化

ByteBuffer.allocate(10) HeapByteBuffer 使用的还是java内存。
ByteBuffer.allocate(10) DirectByteBuffer 使用的是操作系统内存。

如图：

Java NIO 和 Netty快速入门,nio,netty,java,nio,python,netty

最终优化：

Java NIO 和 Netty快速入门,nio,netty,java,nio,python,netty

实现了0拷贝，即不用在java内存中进行拷贝。

更少的用户态切换
不利于cpu计算，减少cpu缓存伪共享
零拷贝只适合小文件传输

AIO

用aio来解决数据复制阶段阻塞问题：

同步意味着，在进行读写操作时，线程需要等待结果，还是相当于闲置
异步意味着，在进行读写操作时，线程不必等待结果，而是将来由操作系统来通过回调方式由另外的线程来获得结果。

Netty入门

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

Netty和NIO区别：

NIO：

需要自己构建协议
解决TCP 传输问题，入粘包半包
epoll空轮询导致cpu 100%
读 API 进行增强，使之更加易用

快速入门

依赖：

<!-- https://mvnrepository.com/artifact/io.netty/netty-all -->
<dependency>
    <groupId>io.netty</groupId>
    <artifactId>netty-all</artifactId>
    <version>4.1.77.Final</version>
</dependency>

服务端：

package org.example.netty_quickStart;

import io.netty.bootstrap.ServerBootstrap;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.ChannelInboundHandlerAdapter;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.SimpleChannelInboundHandler;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.sctp.nio.NioSctpChannel;
import io.netty.channel.sctp.nio.NioSctpServerChannel;
import io.netty.channel.socket.nio.NioServerSocketChannel;
import io.netty.channel.socket.nio.NioSocketChannel;
import io.netty.handler.codec.string.StringDecoder;

public class HelloServer {
    public static void main(String[] args) throws InterruptedException {
        // 启动器，负责组装 netty组件，启动服务器
        new ServerBootstrap()
                // 选择服务器的 ServerSocketChannel实现
                .group(new NioEventLoopGroup()) // 1
                .channel(NioServerSocketChannel.class) // 2
                // boss负责处理连接，worker(child)负责处理读写，决定了 worker(child)能执行哪些操作(handler)
                .childHandler(
                        // channel代表和客户端进行数据读写的通道 ，此处为初始化。负责添加其他handler的handler
                        new ChannelInitializer<NioSocketChannel>() { // 3
                    protected void initChannel(NioSocketChannel ch) {
                        // 添加具体handler
                        ch.pipeline().addLast(new StringDecoder()); // 将ByteBuf转化为字符
                        ch.pipeline().addLast(new SimpleChannelInboundHandler<String>() { // 6
                            @Override
                            protected void channelRead0(ChannelHandlerContext ctx, String msg) {
                                System.out.println(msg);
                            }
                        });
                    }
                })
                .bind(8080); // 4
    }
}

客户端：

package org.example.netty_quickStart;

import io.netty.bootstrap.Bootstrap;
import io.netty.channel.Channel;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.nio.NioSocketChannel;
import io.netty.handler.codec.string.StringEncoder;

import java.util.Date;

public class HelloClient {
    public static void main(String[] args) throws InterruptedException {
        // 创建启动器
        new Bootstrap()
                // 添加EventLoop
                .group(new NioEventLoopGroup())
                // 选择客户端channel实现
                .channel(NioSocketChannel.class)
                // 添加处理器
                .handler(new ChannelInitializer<NioSocketChannel>() {
                    @Override
                    protected void initChannel(NioSocketChannel ch) throws Exception {
                        ch.pipeline().addLast(new StringEncoder());
                    }
                })
                .connect("localhost", 8080)
                .sync()
                .channel()
                .writeAndFlush(new Date() + "hello world!");

    }
}

Java NIO 和 Netty快速入门,nio,netty,java,nio,python,netty

组件：

EventLoop

EventLoop本质是一个单线程执行器(同时维护了一个Selector)，里面有run方法处理Channel上源源不断的io事件

EventLoopGorup是一组EventLoop，Channel一般会调用EventGroup的register方法来绑定其中一个EventLoop，后续这个Channel上的io事件都由这个EventLoop来处理

Channel

close() 可以用来关闭 channel
closeFuture() 用来处理 channel 的关闭
- sync 方法作用是同步等待 channel 关闭
- 而 addListener 方法是异步等待 channel 关闭
pipeline() 方法添加处理器
write() 方法将数据写入
writeAndFlush() 方法将数据写入并刷出

ChannelFuture

其实这一部分：

Channel localhost = new Bootstrap()
                // 添加EventLoop
                .group(new NioEventLoopGroup())
                // 选择客户端channel实现
                .channel(NioSocketChannel.class)
                // 添加处理器
                .handler(new ChannelInitializer<NioSocketChannel>() {
                    @Override
                    protected void initChannel(NioSocketChannel ch) throws Exception {
                        ch.pipeline().addLast(new StringEncoder());
                    }
                })
                .connect("localhost", 8080)

是一个ChannelFuture对象

其存在一个sync异步阻塞方法，会等待连接建立完成再进行后续写入操作。(同步等待)

也可以使用异步方式

channelFuture.addListener((ChannelFutureListener) future -> {
    System.out.println(future.channel()); // 2
});

Netty异步提高的是单位时间内对数据的吞吐量。

Future & Promise

jdk Future 只能同步等待任务结束（或成功、或失败）才能得到结果
netty Future 可以同步等待任务结束得到结果，也可以异步方式得到结果，但都是要等任务结束
netty Promise 不仅有 netty Future 的功能，而且脱离了任务独立存在，只作为两个线程间传递结果的容器

Handler & Pipeline

ChannelHandler 用来处理 Channel 上的各种事件，分为入站、出站两种。所有 ChannelHandler 被连成一串，就是 Pipeline

入站处理器通常是 ChannelInboundHandlerAdapter 的子类，主要用来读取客户端数据，写回结果
出站处理器通常是 ChannelOutboundHandlerAdapter 的子类，主要对写回结果进行加工

打个比喻，每个 Channel 是一个产品的加工车间，Pipeline 是车间中的流水线，ChannelHandler 就是流水线上的各道工序，而后面要讲的 ByteBuf 是原材料，经过很多工序的加工：先经过一道道入站工序，再经过一道道出站工序最终变成产品

注意：在使用addlast方法时，我们的handler并不是加载最后而是加在tail之前。入站按从head开始方向进行，出站需要写方法才可以使用，从tail向前出站

区分ctx和channel的writeAndFlush方法，ctx只会从当前handler向前找，而channel是从tail向前找。

测试embedded-channel

ByteBuf

大多数时候netty采用直接内存作为ByteBuf内存

可以使用下面的代码来创建池化基于堆的 ByteBuf

ByteBuf buffer = ByteBufAllocator.DEFAULT.heapBuffer(10);

也可以使用下面的代码来创建池化基于直接内存的 ByteBuf

ByteBuf buffer = ByteBufAllocator.DEFAULT.directBuffer(10);

直接内存创建和销毁的代价昂贵，但读写性能高（少一次内存复制），适合配合池化功能一起用
直接内存对 GC 压力小，因为这部分内存不受 JVM 垃圾回收的管理，但也要注意及时主动释放

池化

池化的最大意义在于可以重用 ByteBuf，优点有

没有池化，则每次都得创建新的 ByteBuf 实例，这个操作对直接内存代价昂贵，就算是堆内存，也会增加 GC 压力
有了池化，则可以重用池中 ByteBuf 实例，并且采用了与 jemalloc 类似的内存分配算法提升分配效率
高并发时，池化功能更节约内存，减少内存溢出的可能

池化功能是否开启，可以通过下面的系统环境变量来设置

-Dio.netty.allocator.type={unpooled|pooled}

ByteBuf组成：

ByteBuf 由四部分组成

Java NIO 和 Netty快速入门,nio,netty,java,nio,python,netty

最开始读写指针都在 0 位置

扩容

再写入一个 int 整数时，容量不够了（初始容量是 10），这时会引发扩容

buffer.writeInt(6);
log(buffer);

扩容规则是

如何写入后数据大小未超过 512，则选择下一个 16 的整数倍，例如写入后大小为 12 ，则扩容后 capacity 是 16
如果写入后数据大小超过 512，则选择下一个 2^n，例如写入后大小为 513，则扩容后 capacity 是 210=1024（29=512 已经不够了）
扩容不能超过 max capacity 会报错

我们的Pipeline存在的head和tail会进行释放内存，但是尽量在最后使用资源的handler中进行释放操作。

slice

【零拷贝】的体现之一，对原始 ByteBuf 进行切片成多个 ByteBuf，切片后的 ByteBuf 并没有发生内存复制，还是使用原始 ByteBuf 的内存，切片后的 ByteBuf 维护独立的 read，write 指针

Java NIO 和 Netty快速入门,nio,netty,java,nio,python,netty

duplicate

【零拷贝】的体现之一，就好比截取了原始 ByteBuf 所有内容，并且没有 max capacity 的限制，也是与原始 ByteBuf 使用同一块底层内存，只是读写指针是独立的

Java NIO 和 Netty快速入门,nio,netty,java,nio,python,netty

copy

会将底层内存数据进行深拷贝，因此无论读写，都与原始 ByteBuf 无关

CompositeByteBuf

可以将多个 ByteBuf 合并为一个逻辑上的 ByteBuf，避免拷贝

Unpooled

Unpooled 是一个工具类，类如其名，提供了非池化的 ByteBuf 创建、组合、复制等操作

这里仅介绍其跟【零拷贝】相关的 wrappedBuffer 方法，可以用来包装 ByteBuf

ByteBuf 优势

池化 - 可以重用池中 ByteBuf 实例，更节约内存，减少内存溢出的可能
读写指针分离，不需要像 ByteBuffer 一样切换读写模式
可以自动扩容
支持链式调用，使用更流畅
很多地方体现零拷贝，例如 slice、duplicate、CompositeByteBuf文章来源地址https://www.toymoban.com/news/detail-835844.html