Netty

分布式根基于网络编程，Netty恰是java网络编程的王者，致力于高性能编程。

前置

适用于网络开发，服务器开发。多线程，线程池，maven。

大纲

NIO编程（Selector，ByteBuffer和Channel）
Netty入门：EventLoop，Channel，Future，Pipeline，Handler，ByteBuf
Netty进阶：粘包半包，协议，序列化
Netty调优：参数优化
Netty源码

NIO基础

non-clocking io：非阻塞IO

1 三大组件

1.1 Channel&Buffer

Channel是读写数据的双向通道。常见通道的有File，Datagram，Socket，ServerSocket。
Buffer是用来缓冲读写数据的。常见的有Byte(Mapped,Direct,Heap)，Int，Float，Double，Char，

1.2 Selector

多线程版本：早期的服务器基于多线程实现。内存占用高，线程上下文切换成本高，只适合连接少的场景。
线程池版本：阻塞模式下，线程仅能处理一个socket连接，仅适合短链接场景。
selector版本：selector作用就是配合一个线程来管理多个channel，获取channel上发生的事件，这些channel工作在非阻塞下，不会让线程吊死在一个channel上适合连接多但流量低的场景。
调用selector的select()会阻塞直到channel发生了读写就绪事件，select方法会返回这些事件交给thread来处理。

2 ByteBuffer

向buffer写入数据：channel.read(buffer);
调用filp切换读模式
从buffer读数据：buffer.get();
调用clear或compact切换写模式
重复1-4

2.1 ByteBuffer结构

Buffer/ByteBuffer/ByteBuf详解
ByteBuffer有以下重要属性：capacity容量，position读写指针，limit限制。
flip：position切换读取位置，limit切换为读取限制
compact：把未读完的部分向前压缩，然后切换写模式
ByteBuffer写模式结构teBuffer
ByteBuffer读模式结构

2.2 ByteBuffer常见方法

分配空间：ByteBuffer buf = ByteBuffer.allocate(16);（堆，低效，会GC）ByteBufferDirect(16); （直接内存，高效，不会GC，分配低效）
写入数据：
* 调用channel的read方法：channel.read();
* 调用buffer自己的put方法：buf.put();
读取数据：
* 调用channel的write方法：channel.write();
* 调用buffer自己的get方法：buf.get();
* 注：get方法会将position指针向后走，想重复读可调用rewind方法将position置0或get(i);
* 标记position：mark
* 回到标记位置：reset
字符串与ByteBuffer转换：
* 直转方法：buffer.put(“hello”.getBytes());
* Charset方法：ByteBuffer buf = StandardCharsets.UTF_8.encode(“hello”);
* wrap方法：ByteBuffer buf = ByteBuffer.wrap(“hello”.getBytes());
* 回转String：StandardCharsets.UTF_8.decode(buf).toString();

2.3 组合练习

public class TestByteBufferExam {public static void main(String[] args) {/*** 网络上多条数据发送客户端使用/n进行分割，但由于某种原因，被进行重新组合，例如* Hello,world\n* I'm aric\n* How are you?\n* 变成下面的两个 byteBuffer(粘包，半包)* Hello,world\nI'm aric\nHo* w are you?\n* 现要求将错乱的数据恢复按\n分割数据*/ByteBuffer source = ByteBuffer.allocate(32);source.put("Hello,world\nI'm aric\nHo".getBytes());split(source);source.put("w are you?\n".getBytes());split(source);}private static void split(ByteBuffer source) {source.flip();for (int i = 0; i < source.limit(); i++) {if (source.get(i) == '\n') {  //找到一条完整消息int length = i + 1 - source.position();  //消息长度//把这条完整的消息存入新的ByteBufferByteBuffer target = ByteBuffer.allocate(length);//从source读，向target写for (int j = 0; j < length; j++) {byte b = source.get();target.put(b);}debugAll(target);}}source.compact();}
}

3 文件编程

3.1 FileChannel

注：FileChannel只能工作在阻塞模式下。
获取
不能直接打开FIleChannel，必须通过FileInputStream，FileOutputStream或RandomAccessFile来获取FileChannel，他们都有getChannel();

FileInputStream：此channel只能读
FileOutputStream：此channel只能写
RandomAccessFile：根据其读写模式决定

读取
会从channel读取数据填充ByteBuffer，返回值表示读到多少字节，-1表示达到了文件的末尾。

int readBytes = channel.read(buffer);

写入

ByteBuffer buffer = ...;
buffer.put();  //存入数据
buffer.filp();  //切换读模式
while(buffer.hasRemaining()){  //用while因为buffer无法保证一次读取channel中全部内容。channel.write(buffer);
}

关闭
channel必须关闭。
位置
获取当前位置：long pos = hannel.position();
设置当前位置：channel.position(pos);
大小
size方法
强制写入
数据先会缓存，调用force(true)方法可将文件内容和元数据立刻写入磁盘。

3.2 两个Channel传输数据

        try (FileChannel from = new FileInputStream("data.txt").getChannel();FileChannel to = new FileOutputStream("to.txt").getChannel();) {//transferTo底层采用零拷贝，效率高//from.transferTo(0, from.size(), to);  最大只能传输2Glong size = from.size();for (long left = size; left > 0; ) {  //left 表示剩余多少字节left -= from.transferTo((size - left), left, to);}} catch (IOException e) {e.printStackTrace();}

3.3 Path

jdk7引入Path和Paths类，Path表示文件路径，Paths是工具类，用来获取path实例。

3.4 Files

检查文件是否存在
Path path = Paths.get(“data.txt”);
System.out.println(Files.ex);
拷贝文件
Files.copy(source, target);
移动文件
Files.move(source, target, StandardCopyOption.ATOMIC_MOVE);
删除文件
Files.delete(target);

3.5 遍历文件(访问者模式)

    public static void main(String[] args) throws IOException {walkFile();  //遍历文件deleteFile();  //删除文件String source = "I:\\BaiduNetdiskDownload";String target = "I:\\BaiduNetdisk";copyFile(source, target);  //拷贝文件}private static void walkFile() throws IOException {AtomicInteger dirCount = new AtomicInteger();AtomicInteger fileCount = new AtomicInteger();AtomicInteger jarCount = new AtomicInteger();Files.walkFileTree(Paths.get("I:\\BaiduNetdiskDownload"), new SimpleFileVisitor<Path>() {@Overridepublic FileVisitResult preVisitDirectory(Path dir, BasicFileAttributes attrs) throws IOException {System.out.println(">" + dir);dirCount.incrementAndGet();return super.preVisitDirectory(dir, attrs);}@Overridepublic FileVisitResult visitFile(Path file, BasicFileAttributes attrs) throws IOException {if (file.toString().endsWith(".jar")) {System.out.println(file);jarCount.incrementAndGet();}System.out.println(file);fileCount.incrementAndGet();return super.visitFile(file, attrs);}});System.out.println(dirCount);System.out.println(fileCount);System.out.println(jarCount);}private static void deleteFile() throws IOException {Files.walkFileTree(Paths.get("I:\\BaiduNetdisk"), new SimpleFileVisitor<Path>(){@Overridepublic FileVisitResult visitFile(Path file, BasicFileAttributes attrs) throws IOException {Files.delete(file);return super.visitFile(file, attrs);}@Overridepublic FileVisitResult postVisitDirectory(Path dir, IOException exc) throws IOException {Files.delete(dir);return super.postVisitDirectory(dir, exc);}});}private static void copyFile(String source, String target) throws IOException {Files.walk(Paths.get(source)).forEach(path -> {try {String targetName = path.toString().replace(source, target);//是目录if (Files.isDirectory(path)) {Files.createDirectories(Paths.get(targetName))}//是普通文件else if (Files.isRegularFile(path)) {Files.copy(path, Paths.get(targetName));}} catch (IOException e) {e.printStackTrace();}});}

4 网络编程

4.1 阻塞&非阻塞

阻塞

阻塞模式下，相关方法都会导致线程暂停
- ServerSocketChannel.accept 会在没有连接建立时让线程暂停
- SocketChannel.read 会在没有数据可读时让线程暂停
- 没有数据和数据复制过程中，线程阻塞等待，不占CPU，空闲
单线程下，阻塞方法之间相互影响，几乎不能工作，需多线程支持
多线程下需考虑问题：
- 32位jvm一个线程320K，64位jvm一个线程1024K，为减少线程数，需采用线程池技术。
- 即使用线程池，多链接长时间inactive，会阻塞线程池中所有线程。

非阻塞

非阻塞下，相关方法都不会让线程暂停
- 在 ServerSocketChannel.accept 在没有连接建立时，会返回 null，继续运行
- SocketChannel.read 在没有数据可读时，会返回 0，但线程不必阻塞，可以去执行其它 SocketChannel 的 read 或是去执行ServerSocketChannel.accept
- 写数据时，线程只是等待数据写入 Channel 即可，无需等 Channel 通过网络把数据发送出去
在某个channel没有可读事件时，线程不必阻塞，可去处理其他有可读事件的channel
数据复制过程中，线程实际还是阻塞的（AIO改进）

多路复用

线程必须配合Selector才能完成对多个Channel可读写事件的监控，即多路复用。

多路复用仅针对网络IO，普通文件IO没法利用多路复用。
如果不用Selector的非阻塞模式，那么Channel读取到的字节很多时候都是0，而Selector保证了有可读事件才读取。
- 有可连接事件时才去连接
- 有可读事件才去读取
- 有可写事件才去写入
Channel输入的数据一旦准备好，会触发Selector的可读事件

//nio 阻塞模式&非阻塞public static void main(String[] args) throws IOException {//消息缓冲区ByteBuffer buffer = ByteBuffer.allocate(16);//创建服务器ServerSocketChannel ssc = ServerSocketChannel.open();ssc.configureBlocking(false);  //非阻塞模式开启，没有建立连接时，sc返回null//绑定监听端口ssc.bind(new InetSocketAddress(8080));ArrayList<SocketChannel> channels = new ArrayList<>();while (true) {//循环监听客户端连接  accept  socket用来与客户端通信SocketChannel sc = ssc.accept();  //阻塞方法：线程停止运行，没链接时阻塞if(sc != null){sc.configureBlocking(false);  //将socketChannel设为非阻塞模式。如果没有读到数据，read返回0channels.add(sc);}for (SocketChannel channel : channels) {channel.read(buffer);  //阻塞方法：线程停止运行，没有数据时阻塞buffer.flip();System.out.println(buffer);buffer.clear();}}}

4.2 Selector

一个线程配合 selector 就可以监控多个 channel 的事件，事件发生线程才去处理。避免非阻塞模式下所做无用功
让这个线程能够被充分利用
节约了线程的数量
减少了线程上下文切换

4.3 处理read事件

void selectorEdition() throws IOException {//1. 创建selectorSelector selector = Selector.open();ServerSocketChannel ssc = ServerSocketChannel.open();ssc.configureBlocking(false);//2. 建立selector和channel的注册,sscKey是事件的句柄，是将来事件发生后，通过它可以知道事件和哪个channel的事件SelectionKey sscKey = ssc.register(selector, 0, null);//表示sscKey只关注accept事件sscKey.interestOps(SelectionKey.OP_ACCEPT);ssc.bind(new InetSocketAddress(8080));while (true) {//3. select 方法，没有事件发生，线程阻塞，有事件，线程才会恢复运行//selector在事件未处理时，不会阻塞，事件发生后要么处理，要么取消，不能置之不理selector.select();//4. 处理事件,selectedKeys内部包含了所有发生的事件Iterator<SelectionKey> iter = selector.selectedKeys().iterator();  //selector发生事件后，selectedKeys集合只有加入，不会删除while (iter.hasNext()) {SelectionKey key = iter.next();//处理完key一定要移除调，不然下次处理时会报空指针异常iter.remove();//5. 区分事件类型if (key.isAcceptable()) {  //如果时acceptServerSocketChannel channel = (ServerSocketChannel) key.channel();SocketChannel sc = channel.accept();//读取事件sc.configureBlocking(false);SelectionKey scKey = sc.register(selector, 0, null);scKey.interestOps(SelectionKey.OP_READ);} else if (key.isReadable()) {try {SocketChannel channel = (SocketChannel) key.channel();ByteBuffer buffer = ByteBuffer.allocate(16);int read = channel.read(buffer);  //正常断开，read返回-1if (read == -1) {key.cancel();} else {buffer.flip();System.out.println(buffer);//split(buffer);  //使用分隔符方式接收消息}} catch (IOException e) {e.printStackTrace();//客户端出异常情况，需手动从selectedKeys集合取消keykey.cancel();}}}}}

绑定Channel事件

channel 必须工作在非阻塞模式
FileChannel 没有非阻塞模式，因此不能配合 selector 一起使用
绑定的事件类型可以有

connect - 客户端连接成功时触发
accept - 服务器端成功接受连接时触发
read - 数据可读入时触发，有因为接收能力弱，数据暂不能读入的情况
write - 数据可写出时触发，有因为发送能力弱，数据暂不能写出的情况

监听Channel事件

可以通过下面三种方法来监听是否有事件发生，方法的返回值代表有多少 channel 发生了事件
方法1：阻塞直到绑定事件发生-selector.select();
方法2：阻塞直到绑定事件发生，或是超时（时间单位为 ms）-selector.select(long timeout);
方法3：不会阻塞，也就是不管有没有事件，立刻返回，自己根据返回值检查是否有事件-selector.selectNow();

select 何时不阻塞

事件发生时

客户端发起连接请求，会触发 accept 事件
客户端发送数据过来，客户端正常、异常关闭时，都会触发 read 事件，另外如果发送的数据大于 buffer 缓冲区，会触发多次读取事件
channel 可写，会触发 write 事件
在 linux 下 nio bug 发生时
调用 selector.wakeup()
调用 selector.close()
selector 所在线程 interrupt

事件发生后能否不处理

事件发生后，要么处理，要么取消（cancel），不能什么都不做，否则下次该事件仍会触发，这是因为 nio 底层使用的是水平触发

为何要 iter.remove()

因为 select 在事件发生后，就会将相关的 key 放入 selectedKeys 集合，但不会在处理完后从 selectedKeys 集合中移除，需要我们自己编码删除。例如

第一次触发了 ssckey 上的 accept 事件，没有移除 ssckey
第二次触发了 sckey 上的 read 事件，但这时 selectedKeys 中还有上次的 ssckey ，在处理时因为没有真正的 serverSocket 连上了，就会导致空指针异常

cancel 的作用

cancel 会取消注册在 selector 上的 channel，并从 keys 集合中删除 key 后续不会再监听事件。

处理消息的边界

buffer示意图

一种思路是固定消息长度，数据包大小一样，服务器按预定长度读取，缺点是浪费带宽
另一种思路是按分隔符拆分，缺点是效率低
TLV 格式，即 Type 类型、Length 长度、Value 数据，类型和长度已知的情况下，就可以方便获取消息大小，分配合适的 buffer，缺点是 buffer 需要提前分配，如果内容过大，则影响 server 吞吐量
- Http 1.1 是 TLV 格式
- Http 2.0 是 LTV 格式
  
  所以ByteBuffer不能是局部变量，如果消息过长，会分两次读取，所以每个SocketChannel都需有自己独有的ByteBuffer。并做扩容优化

if (key.isAcceptable()) {  //如果时acceptServerSocketChannel channel = (ServerSocketChannel) key.channel();SocketChannel sc = channel.accept();//读取事件sc.configureBlocking(false);ByteBuffer buffer = ByteBuffer.allocate(16);  //attachment buffer和sc绑定SelectionKey scKey = sc.register(selector, 0, null, buffer);scKey.interestOps(SelectionKey.OP_READ);
} else if (key.isReadable()) {try {SocketChannel channel = (SocketChannel) key.channel();ByteBuffer buffer = (ByteBuffer) key.attachment(); //从key中获取独有的ByteBufferint read = channel.read(buffer);  //正常断开，read返回-1if (read == -1) {key.cancel();} else {//buffer.flip();//System.out.println(buffer);split(buffer);  //使用分隔符方式接收消息if(buffer.position() == buffer.limit()) {  //扩容ByteBuffer newBuffer = ByteBuffer.allocate(buffer.capacity() * 2);buffer.flip();newBuffer.put(buffer);key.attach(newBuffer);  //替换原key中的ByteBuffer}}} catch (IOException e) {e.printStackTrace();//客户端出异常情况，需手动从selectedKeys集合取消keykey.cancel();}
}

ByteBuffer大小分配

每个channel都需要记录可能被切分的消息，因为ByteBuffer不是线程安全的，因此需要为每个channel维护一个独立的ByteBuffer
ByteBuffer不能太大，比如一个ByteBuffer 1Mb的话，要支持百万连接就要1Tb内存，因此需要设计大小可变的ByteBuffer

一种思想是首先分配一个较小的buffer，不够再扩容，优点是消息连续容易处理，缺点是数据拷贝耗性能。
另一种思想是用多个数组组成buffer，一个数组不够，把多出来的内容写入新的数组，区别是消息存储不连续，解析复杂，优点是避免拷贝

4.4处理write事件

非阻塞模式下，无法保证把 buffer 中所有数据都写入 channel，因此需要追踪 write 方法的返回值（代表实际写入字节数）
用 selector 监听所有 channel 的可写事件，每个 channel 都需要一个 key 来跟踪 buffer，但这样又会导致占用内存过多，就有两阶段策略
- 当消息处理器第一次写入消息时，才将 channel 注册到 selector 上
- selector 检查 channel 上的可写事件，如果所有的数据写完了，就取消 channel 的注册
- 如果不取消，会每次可写均会触发 write 事件
  example：https://gitee.com/xuyu294636185/netty-demo.git

write 为何要取消

只要向 channel 发送数据时，socket 缓冲可写，这个事件会频繁触发，因此应当只在 socket 缓冲区写不下时再关注可写事件，数据写完之后再取消关注

4.5多线程版

单线程配一个选择器，专门处理 accept 事件
创建 cpu 核心数的线程，每个线程配一个选择器，轮流处理 read 事件
代码：https://gitee.com/xuyu294636185/netty-demo.git

5. NIO&BIO

5.1 stream与channel

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

5.2 IO模型

当调用一次channel.read或stream.read后，会切换至操作系统内核态完成真正的数据读取，而读取又分为：等待数据阶段、复制数据阶段。
读取过程
阻塞IO：用户态调用内核态阻塞，等待内核态数据就绪复制完成后才能返回。期间用户和内核都阻塞
非阻塞IO：用户态调用内核态阻塞会立刻返回并循环直到有数据。用户态只有在等待数据时非阻塞，复制数据时还是阻塞。但是内核和用户切换很频繁。
阻塞IO的问题
多路复用：先调用select方法，用户态调用内核阻塞，有事件才返回用户态，用户态再读取内核态阻塞等待复制数据完后才返回。期间用户和内核都阻塞。
多路复用一次可以注册多个事件

同步：线程自己去获取结果（单线程）【同步阻塞，同步非阻塞，同步多路复用】
异步：线程自己不去获取结果，而是由其他线程送结果（至少两个线程）【异步阻塞（不存在），异步非阻塞】

5.3 零拷贝

传统IO

传统IO：将文件先通过accessfile读入byte数组中，再通过socket输出流i写出客户端。

File f = new File("data.txt");
RandomAccessFile file = new RandomAccessFile(file, "r");byte[] buf = new byte[(int)f.length()];
file.read(buf);Socket socket = ...;
socket.getOutputStream().write(buf);

内部工作流程

java本身不具备IO读写能力，read调用从用户态切换到内核态，操作kernel系统，将数据读到内核缓冲区，期间用户线程阻塞，系统使用DMA实现文件读，期间也不会使用CPU
从内核态切回用户态，将数据从内核缓冲区读入用户缓冲区，期间cpu参与拷贝，无法利用DMA
调用write方法，将数据从用户缓冲区写入socket缓冲区，cpu参与拷贝
接下来要向网卡写入数据，这项能力java又不具备，因此又得到从用户态切换至内核态，调用操作系统的写能力，使用DMA将socket缓冲区的数据写入网卡，不会使用cpu。
可以看到中间环节较多，java的IO实际不是物理设备级别的读写，而是缓存的复制，底层的真正读写是操作系统来完成的。

用户态与内核态的切换发生了3次，这个操作比较重量级。
数据拷贝了4次。

NIO优化

通过DirectByteBuf

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

不同：java使用DurectByteBuf将堆外内存映射到jvm内存中来直接访问使用
这块内存不受jvm垃圾回收的影响，因此内存地址固定，有助于IO读写
java中的DirectByteBuf对象仅维护了此内存的虚引用，内存回收分成两步
1. DirectByteBuf对象被垃圾回收，将虚引用加入引用队列
2. 通过专门线程访问引用队列，根据虚引用释放堆外内存
减少一次数据拷贝，用户态与内核态的切换次数没有减少