Java IO之 Netty与NIO服务器-NIO中的零拷贝
1.什么是零拷贝
一种避免 CPU 将数据从一块存储拷贝到另外一块存储的技术。针对操作系统中的设备驱动程序、文件系统以及网络协议堆栈而出现的各种零拷贝技术极大地提升了特定应用程序的性能，并且使得这些应用程序可以更加有效地利用系统资源。这种性能的提升就是通过在数据拷贝进行的同时，允许 CPU 执行其他的任务来实现的。
2.数据拷贝
2.1. 传统方式下的数据拷贝原理
①一个read系统调用后，DMA执行了一次数据拷贝，从磁盘到内核空间
②read结束后，发生第二次数据拷贝，由cpu将数据从内核空间拷贝至用户空间
③send系统调用，cpu发生第三次数据拷贝，由cpu将数据从用户空间拷贝至内核空间(socket缓冲区)
④send系统调用结束后，DMA执行第四次数据拷贝，将数据从内核拷贝至协议引擎
2.1. 基于NIO的数据零拷贝(sendfile)
①DMA从拷贝至内核缓冲区 ②cpu将数据从内核缓冲区拷贝至内核空间(socket缓冲区) ③DMA将数据从内核拷贝至协议引擎 3.代码实现基于NIO数据零拷贝 3.1. 服务端
public class IOServer {
public static void main(String[] args) throws IOException {
InetSocketAddress inetSocketAddress = new InetSocketAddress(7001);
ServerSocketChannel serverSocketChannel=ServerSocketChannel.open();
//绑定
serverSocketChannel.socket().bind(inetSocketAddress);
//创建buffer
ByteBuffer byteBuffer=ByteBuffer.allocate(1024);
while (true){
//获得socketChannel
SocketChannel socketChannel = serverSocketChannel.accept();
int readcount=0;
while (-1!=readcount){
try{
readcount = socketChannel.read(byteBuffer);
}catch (Exception e){
break;
}
byteBuffer.rewind();//倒带,让position=0,mark作废
}
}
}
}
3.2. 客户端
public class IOClient {
public static void main(String[] args) throws IOException {
SocketChannel socketChannel = SocketChannel.open();
// socketChannel.connect(new InetSocketAddress("localhost", 7001));
socketChannel.connect(new InetSocketAddress("127.0.0.1", 7001));
String fileName = "1.txt";
//获取文件通道
FileChannel fileChannel = new FileInputStream(fileName).getChannel();
//发送开始计时
Long startTime = System.currentTimeMillis();
/** transferTo实现零拷贝
* 无论文件大小
*在Linux系统下运行transferTo一次性可以传输完成
* 在windows下一次只能发送8M,超出范围文件需要分段传输
* transferTo(long position, long count,WritableByteChannel target)
*参数1:文件传输时的位置,作为分段传输的标注点
* 参数2:文件大小
* 参数3.通道
*/
long l = fileChannel.transferTo(0, fileChannel.size(), socketChannel);
System.out.println("发送的字节总数="+l+",耗时:"+(System.currentTimeMillis()-startTime));
//关闭
fileChannel.close();
}
}
NIO的零拷贝的0指的是，0次需要cpu的拷贝，DMA拷贝不算在里面。零拷贝可以大大提高我们数据传输的效率。
传统IO
磁盘 -> 内核空间的缓存(DMA)
内核空间的缓存 -> 用户态中的程序缓存
对数据做一系列操作
用户态中的程序缓存 -> 内核空间的的socket缓存
内核空间的socket缓存->网卡缓存（DMA）
网络发送(或拷贝到磁盘)！
在一次传统的IO中，操作系统一共进行了2次拷贝，4次操作系统状态转换
2次拷贝：注意，磁盘到内核态的读写是通过DMA拷贝，外部设备(磁盘,U盘)不通过CPU直接与系统内存交换数据。所以是2次拷贝
4次系统状态切换：详情如下
tips: 为什么数据不直接从磁盘拷贝到用户程序空间呢？通过局部性原理我们知道，当我们从磁盘上取一个数据的时候，有很大可能下一次读取的数据就是本次读取数据周围的数据。因此操作系统为了提高性能，在读取数据的时候，会先把该目标数据周围的数据（NTFS下是4KB为单位）也一并读到操作系统的read buffer中，下次读取时，就有很大可能命中read buffer，减少了磁盘IO
NIO 零拷贝
不需要对数据进行操作
windows Linux2.4以下
sendFile()
该版本下的sendFile，只需要2次状态切换，即 开始: 用户态->内核态（read buffer 到 socket buffer）->用户态，但是还需要把Read buffer中的数据通过cpu完整的拷贝到Socket Buffer，所以不能算是真正的0拷贝。
该种方法是2次切换，1次拷贝
Linux 2.4以后
sendFile()
该版本下，sendFile()直接将内核态的Read buffer发送到NIC的 buffer，只有部分描述信息，如缓存位移，描述符经过socket buffer，通过这些信息，DMA直接把数据拷贝到外围设备缓存中。由于数据描述信息量很少，基本可以视作0，所以这也是0拷贝
需要对数据进行操作
当我们需要在用户空间对数据进行处理时，上述方法都不能解决问题
mmap，把内核态的缓存空间和用户态的缓存空间映射到一块物理地址，实现共享，减少了内核态缓存到用户态缓存的一次拷贝。(因为使用mmap后，内存共享了，进程直接对内核空间映射到进程的虚拟地址进行读写就行，无需拷贝，但系统状态还是需要切换的)
该种方法是4次上下文切换，3次拷贝（2次DMA拷贝）。
3次上下文切换：
发出mmap系统调用，用户态->内核态，通过DMA将磁盘文件拷贝到read buffer中
mmap系统调用返回，内核态->用户态，此时用户空间和内核空间中的read buffer共享一块缓冲区内存空间
用户修改数据，并发出write调用，用户态->内核态，read buffer中的数据通过cpu拷贝到socket buffer
write调用返回，socker buffer通过DMA拷贝到外围设备。内核态->用户态
零拷贝的再次理解
我们说零拷贝，是从cpu的角度来说的,减少对cpu占用可以大大提升效率，而DMA拷贝无法避免。
零拷贝不仅仅带来更少的数据复制，还能带来其他的性能优势，例如更少的上下文切换，更少的CPU缓存伪共享以及无CPU校验和计算。
mmap和sendFile的区别
mmap适合小数据读写， sendFile适合大文件传输
mmap需要4次上下文切换，1次数据拷贝(还有);sendFile需要3次上下文切换，最小号2次数据拷贝
sendFile可以利用DMA，减少CPU拷贝，mmap则不能(必须从内核拷贝到Socket缓冲区)
ava中零拷贝有2种（零拷贝是指没有CPU拷贝）
1，mmap(内存映射)
2,sendfile
传统IO数据读写；
File file = new File("test.txt");
RandomAccessFile raf = new RandomAccessFile(file, "rw");
byte[] arr = new byte[(int) file.length()]; raf.read(arr);
Socket socket = new ServerSocket(8080).accept();
socket.getOutputStream().write(arr);
过程如下;
传统IO发现copy了4次，cpu状态切换三次，
所以要优化；
MMAP优化：
mmap 通过内存映射，将文件映射到内核缓冲区，同时，用户空间可以共享内核空间的数据。这样，在进行网络传输时，就可以减少内核空间到用户控件的拷贝次数。如下图
此时copy 3次，状态也是3次，
sendFile 优化
Linux 2.1 版本 提供了 sendFile 函数，其基本原理如下：数据根本不经过用户态，直接从内核缓冲区进入到 Socket Buffer，同时，由于和用户态完全无关，就减少了一次上下文切换 示意图和小结 提示：零拷贝从操作系统角度，是没有cpu 拷贝
此时是3次拷贝，切换变成了2次；
Linux 在 2.4 版本中，做了一些修改，避免了从内核缓冲区拷贝到 Socket buffer 的操作，直接拷贝到协议栈，从而再一次减少了数据拷贝。具体如下图和小结： 这里其实有 一次cpu 拷贝 kernel buffer -> socket buffer 但是，拷贝的信息很少，比如 lenght , offset , 消耗低，可以忽略
上图中socketbuffer变成灰色，kernel buffer可以通过DMA copy进入 protocol engine;
我们说零拷贝，是从操作系统的角度来说的。因为内核缓冲区之间，没有数据是重复的（只有 kernel buffer 有一份数据）。 零拷贝不仅仅带来更少的数据复制，还能带来其他的性能优势，例如更少的上下文切换，更少的 CPU 缓存伪共享以及无 CPU 校验和计
mmap 和 sendFile 的区别 mmap 适合小数据量读写，sendFile 适合大文件传输。 mmap 需要 4 次上下文切换，3 次数据拷贝；sendFile 需要 3 次上下文切换，最少 2 次数据拷贝。 sendFile 可以利用 DMA 方式，减少 CPU 拷贝，mmap 则不能（必须从内核拷贝到 Socket 缓冲区）。
例子：
比如下面的零拷贝
public class NewIOClient {undefined
public static void main(String[] args) throws Exception {undefined
SocketChannel socketChannel = SocketChannel.open();
socketChannel.connect(new InetSocketAddress("localhost", 6666));
String filename = "pro*****.zip";
//得到一个文件channel
FileChannel fileChannel = new FileInputStream(filename).getChannel();
//准备发送
long startTime = System.currentTimeMillis();
//在linux下一个transferTo 方法就可以完成传输
//在windows 下 一次调用 transferTo 只能发送8m , 就需要分段传输文件, 而且要主要
//传输时的位置 =》
//transferTo 底层使用到零拷贝
long transferCount = fileChannel.transferTo(0, fileChannel.size(), socketChannel);
System.out.println("发送的总的字节数 =" + transferCount + " 耗时:" + (System.currentTimeMillis() - startTime));
//关闭
fileChannel.close();
}
}
//服务器
public class NewIOServer {undefined
public static void main(String[] args) throws Exception {undefined
InetSocketAddress address = new InetSocketAddress(6666);
ServerSocketChannel serverSocketChannel = ServerSocketChannel.open();
ServerSocket serverSocket = serverSocketChannel.socket();
serverSocket.bind(address);
//创建buffer
ByteBuffer byteBuffer = ByteBuffer.allocate(4096);
while (true) {undefined
SocketChannel socketChannel = serverSocketChannel.accept();
int readcount = 0;
while (-1 != readcount) {undefined
try {undefined
readcount = socketChannel.read(byteBuffer);
}catch (Exception ex) {undefined
// ex.printStackTrace();
break;
}
//
byteBuffer.rewind(); //倒带 position = 0 mark 作废
}
}
}
}
以及老的拷贝方式：
public class OldIOClient {undefined
public static void main(String[] args) throws Exception {undefined
Socket socket = new Socket("localhost", 6666);
String fileName = "proto***.zip";
InputStream inputStream = new FileInputStream(fileName);
DataOutputStream dataOutputStream = new DataOutputStream(socket.getOutputStream());
byte[] buffer = new byte[4096];
long readCount;
long total = 0;
long startTime = System.currentTimeMillis();
while ((readCount = inputStream.read(buffer)) >= 0) {undefined
total += readCount;
dataOutputStream.write(buffer);
}
System.out.println("发送总字节数： " + total + ", 耗时： " + (System.currentTimeMillis() - startTime));
dataOutputStream.close();
socket.close();
inputStream.close();
}
}
//java IO 的服务器
public class OldIOServer {undefined
public static void main(String[] args) throws Exception {undefined
ServerSocket serverSocket = new ServerSocket(6666);
while (true) {undefined
Socket socket = serverSocket.accept();
DataInputStream dataInputStream = new DataInputStream(socket.getInputStream());
try {undefined
byte[] byteArray = new byte[4096];
while (true) {undefined
int readCount = dataInputStream.read(byteArray, 0, byteArray.length);
if (-1 == readCount) {undefined
break;
}
}
} catch (Exception ex) {undefined
ex.printStackTrace();
}
}
}
}
Java的网络编程如果不是专门搞服务器性能开发或者消息分发，几乎可能涉及不到。但是它却是面试找工作必问的一个知识点，涵盖的知识体系也非常广泛，从Java底层IO原理到操作系统内核组成，再到网络TCP、UDP、HTTP的应用实践....因此，即便是职场多年的老油条，仍然需要时刻复习，更别提我这种只有七秒钟记忆的小菜鸟了。
Java网络IO的演化，从最开始JDK1.4之前是基于阻塞的IO；发展到1.4发布后的Nio提供了selector多路复用的机制以及channel和buffer，再到1.7的NIO升级提供了真正的异步api；再发展到后来崭露头角的MINA和Netty。因此整个网络IO编程的学习整理，也会按照下面的几个步骤来进行：
- 网络IO的基本知识与概念
- 普通IO以及BIO服务器
- NIO的使用与服务器Hello world
- Netty的使用与服务器Hello world
今天就简单的了解下网络IO需要具备的基本知识与概念。
同步、异步和阻塞、非阻塞
经常听人提起，同步阻塞服务器或者异步非阻塞服务器，网上有很多的文章针对这个概念作出了讲解，每个人理解的貌似都不太一样。最容易把异步和非阻塞搞混....我这里简单的说下自己的理解：
同步synchronous、异步asynchronous，他们的区别就是发起任务后，本身的一个状态——如果是一直等待结果，那就是同步；如果立即返回，并采用其他的方式得到结果就是异步（比如，状态、通知、回调）。
举个例子：
- 在过去科技不发达的时候，银行取钱都是排队的模式。想取钱就得去排队，直到轮到自己，这就是同步；
- 现在去银行一般直接叫号，然后去休息位置休息打游戏，登到轮到自己的时候，会有通知，这就是异步；
阻塞blocking、非阻塞non-blocking，则聚焦的是CPU在等待结果的过程中的状态。比如前面的例子，排队的过程中什么也不做就是阻塞；一边排队，一遍玩王者荣耀就是非阻塞的。
用户空间与内和空间
这个概念就涉及操作系统了，为了保护操作系统的安全，会将内存分为用户空间和内核空间两个部分。如果用户想要操作内核空间的数据，需要把数据从内和空间拷贝到用户控件。
举个例子：
服务器接收客户端发过来的请求，想要进行处理，大致会经过下面几个步骤：
- 服务器的网络驱动接收到消息，去内核上申请空间；并等待完整的数据包到达（有可能分组传送，没传完...），复制到内核空间；
- 数据从内核空间拷贝到用户空间
- 用户程序进行处理
因此大致可以把接收消息理解为两个阶段：1. 等待数据到达 2. 拷贝到用户空间
了解了这个过程，就能明白为什么会出现经典的5大网络模型了.
五大网络模型
这几个网络模型还是学生时代的时候也看过，但是理解的不够透彻，也不知道到底有什么区别。最近网上也看过不少的文章，发现有一些文章引用的小例子不错，能很简单的了解这些模型的意思。所以我这边也借鉴一下：
在大连高新万达后面有一条叫做金街的小吃街，有一个露天的路边摊叫做“小红旗”，主要是做炸臭豆腐和冷面，然后用冷面把臭豆腐卷起来，刷上臭烘烘的酱料，非常好吃。每次路过都能看到不少人在排队，队伍长的有种想让人辞职加盟的感觉....基本上排个队伍都得半个小时-一个小时吧。
这个排队的过程，明显就是上面所说的同步阻塞模式....那我这边就设想下，如果小红旗的生意做大了，可以怎么发展？正好套用下网络模型的概念...
1. 同步阻塞IO
这个就不详细说了，排队的过程哪也去不了，如果你还没有带手机，排队的过程中就只能干瞪眼了。这就是很明显的同步+阻塞模式。
2. 同步非阻塞IO
如果小红旗的老板搞了一个点菜机，来点单的顾客把自己想吃的划上，然后等着老板去做，自己可以在这一个小时的时间里去周围商场溜达下。但是由于没有任何通信方式，只能不停的回来问老板，做好了没有。
回来询问的时间是由顾客自己掌控的，如果时间很短，那么可以尽量早的知道臭豆腐炸好没，但是也会影响逛街的体验；如果时间很长，有可能臭豆腐早就做好了..结果放的时间长了，反而不好吃了。
因此非阻塞IO基于状态轮训的方式，虽然能让程序在等待的过程中做点其他的事情，但是频繁的切换运行程序，反而会造成很大的压力。
3. IO多路复用/事件驱动
小红旗老板升级了系统，放弃使用点菜机，改用麦当劳那种点餐大屏。同样是点餐，但是一个大屏里面显示了很多人的臭豆腐进度，即节省了资源，也避免大家不停的询问。
其实Nio活着Netty就是基于这种模式，一个线程就可以监听很多IO操作，这样在IO等待上就高效多了。具体实现是依赖于操作系统的，windows和linux都有不同的实现方式。最初的select或者poll，都有并发数的限制，并且NIO的select还有空轮训的问题；epool则突破了连接数的限制，一个线程就可以监听大量的IO操作。这个感兴趣的朋友，可以深入了解下select、poll、epool的原理。
4. 信号驱动IO
小红旗老板又时髦了，搞了一个升级版的美味不用等。顾客基于微信小程序点菜，菜做好了自动提醒顾客取餐....这个提醒的过程，就像是发射了一个特殊的信号。
不过UNIX网络编程里面的信号驱动，可没这么简单，这个信号是依赖于操作系统底层的，捕获信号或者处理都很麻烦，所以现在应用的也不是很广泛。
5. 异步非阻塞IO
一对小情侣李雷和韩梅梅，韩梅梅口味很重，特别喜欢吃臭豆腐，但是李雷完全不感兴趣，闻到味道就想吐。于是李雷就跟韩梅梅约定，让韩梅梅自己去吃，李雷跑到旁边的咖啡厅喝咖啡。韩梅梅自己去排队买臭豆腐，买完顺便吃完，然后回来找李雷....
这个过程就是异步非阻塞的，消息的等待和处理都在服务器端完成，用户只要最后接收到消息处理完的通知就行了。
总结
总结来说，这几种网络模型：
- 同步阻塞：强调的是我要做！ —— 别的啥也别说，就是要做！
- 同步非阻塞：强调的是我想做！ —— 在想的过程中，干点其他的事情更好。
- 异步非阻塞：强调的是我做完了！—— 等得到结果通知的时候，工作已经做完了。
其中细节，还需慢慢体会...后面的文章将会挑几个模型做代码的演示，更多内容还请持续关注。
今天来复习一下基础IO，也就是最普通的IO。
- 网络IO的基本知识与概念
- 普通IO以及BIO服务器
- NIO的使用与服务器Hello world
- Netty的使用与服务器Hello world
输入流与输出流
Java的输入流和输出流，按照输入输出的单元不同，又可以分为字节流和字符流的。
JDK提供了很多输入流和输出流，比如：
字节流可以按照不同的变量类型进行读写，而字符流则是基于字符编码的。不同的字符编码包含的字节数是不一样的，因此在使用字符流时，一定要注意编码的问题。
读写
字节的输入输出流操作
// 字节输入流操作
InputStream input = new ByteArrayInputStream("abcd".getBytes());
int data = input.read();
while(data != -1){
System.out.println((char)data);
data = input.read();
}
// 字节输出流
ByteArrayOutputStream output = new ByteArrayOutputStream();
output.write("12345".getBytes());
byte[] ob = output.toByteArray();
字符的输入输出流操作
// 字符输入流操作
Reader reader = new CharArrayReader("abcd".toCharArray());
data = reader.read();
while(data != -1){
System.out.println((char)data);
data = reader.read();
}
// 字符输出流
CharArrayWriter writer = new CharArrayWriter();
writer.write("12345".toCharArray());
char[] wc = writer.toCharArray();
关闭流
流打开后，相当于占用了一个文件的资源，需要及时的释放。传统的标准关闭的方式为：
// 字节输入流操作
InputStream input = null;
try {
input = new ByteArrayInputStream("abcd".getBytes());
int data = input.read();
while (data != -1) {
System.out.println((char) data);
// todo
data = input.read();
}
}catch(Exception e){
// todo
}finally {
if(input != null) {
try {
input.close();
} catch (IOException e) {
e.printStackTrace();
}
}
}
在JDK1.7后引入了try-with-resources的语法，可以在跳出try{}的时候直接自动释放：
try(InputStream input1 = new ByteArrayInputStream("abcd".getBytes())){
//
}catch (Exception e){
//
}
IOUtils
直接使用IO的API还是很麻烦的，网上的大多数教程都是各种while循环，操作很麻烦。其实apache common已经提供了一个工具类——IOUtils，可以方便的进行IO操作。
比如IOUtils.readLines(is, Charset.forName("UTF-8"));可以方便的按照一行一行读取.
BIO阻塞服务器
基于原始的IO和Socket就可以编写一个最基本的BIO服务器。
概要： 这个模型很简单，就是主线程（Acceptor）负责接收连接，然后开启新的线程专门负责连接处理客户端的请求。
import io.netty.util.CharsetUtil;
import java.io.IOException;
import java.io.OutputStream;
import java.net.ServerSocket;
import java.net.Socket;
public class PlainOioServer {
public void serve(int port) throws IOException {
// 开启Socket服务器，并监听端口
final ServerSocket socket = new ServerSocket(port);
try{
for(;;){
// 轮训接收监听
final Socket clientSocket = socket.accept();
try {
Thread.sleep(500000);
} catch (InterruptedException e) {
e.printStackTrace();
}
System.out.println("accepted connection from "+clientSocket);
// 创建新线程处理请求
new Thread(()->{
OutputStream out;
try{
out = clientSocket.getOutputStream();
out.write("Hi\r\n".getBytes(CharsetUtil.UTF_8));
out.flush();
} catch (IOException e) {
e.printStackTrace();
} finally {
try{
clientSocket.close();
} catch (IOException e) {
e.printStackTrace();
}
}
}).start();
}
} catch (IOException e){
e.printStackTrace();
}
}
public static void main(String[] args) throws IOException {
PlainOioServer server = new PlainOioServer();
server.serve(5555);
}
}
然后执行telnet localhost 5555，就能看到返回结果了。
这种阻塞模式的服务器，原理上很简单，问题也容易就暴露出来：
- 服务端与客户端的连接相当于1:1，因此如果连接数上升，服务器的压力会很大
- 如果主线程Acceptor阻塞，那么整个服务器将会阻塞，单点问题严重
- 线程数膨胀后，整个服务器性能都会下降
改进的方式可以基于线程池或者消息队列，不过也存在一些问题：
- 线程池的数量、消息队列后端服务器并发处理数，都是并发数的限制
- 仍然存在Acceptor的单点阻塞问题
接下来，将会介绍基于Nio的非阻塞服务器模式，如果忘记什么是IO多路复用，可以回顾前面一篇分享。敬请期待吧...
前面一篇中已经介绍了基本IO的使用以及最简单的阻塞服务器的例子，本篇就来介绍下NIO的相关内容，前面的分享可以参考目录：
- 网络IO的基本知识与概念
- 普通IO以及BIO服务器
- NIO的使用与服务器Hello world
- Netty的使用与服务器Hello world
NIO，也叫做new-IO或者non-blocking-IO，就暂且理解为非阻塞IO吧。
为什么选择NIO
那么NIO相对于IO来说，有什么优势呢？总结来说：
- IO是面向流的，数据只能从一端读取到另一端，不能随意读写。NIO则是面向缓冲区的，进行数据的操作更方便了
- IO是阻塞的，既浪费服务器的性能，也增加了服务器的风险；而NIO是非阻塞的。
- NIO引入了IO多路复用器，效率上更高效了。
NIO都有什么
那么NIO都提供了什么呢？
- 基于缓冲区的双向管道，Channel和Buffer
- IO多路复用器Selector
- 更为易用的API
Buffer的使用
在NIO中提供了各种不同的Buffer，最常用的就是ByteBuffer:
可以看到，他们都有几个比较重要的变量：
- capacity——容量，这个值是一开始申请就确定好的。类似c语言申请数组的大小。
- limit——剩余，在写模式下初始的时候等于capacity；在读模式下，等于最后一次写入的位置
- mark——标记位，标记一下position的位置，可以调用reset()方法回到这个位置。
- posistion——位置，写模式下表示开始写入的位置；读模式下表示开始读的位置
总结来说，NIO的Buffer有两种模式，读模式和写模式。刚上来就是写模式，使用flip()可以切换到读模式。
关于这几个位置的使用，可以参考下面的代码：
public class ByteBufferTest {
public static void main(String[] args) {
ByteBuffer buffer = ByteBuffer.allocate(88);
System.out.println(buffer);
String value = "Netty权威指南";
buffer.put(value.getBytes());
System.out.println(buffer);
buffer.flip();
System.out.println(buffer);
byte[] v = new byte[buffer.remaining()];
buffer.get(v);
System.out.println(buffer);
System.out.println(new String(v));
}
}
得到的输出为：
java.nio.HeapByteBuffer[pos=0 lim=88 cap=88]
java.nio.HeapByteBuffer[pos=17 lim=88 cap=88]
java.nio.HeapByteBuffer[pos=0 lim=17 cap=88]
java.nio.HeapByteBuffer[pos=17 lim=17 cap=88]
Netty权威指南
读者可以自己领会一下，这几个变量的含义。另外说明一点，如果遇到自己定义POJO类，就可以像这里的Buffer重载toString()方法，这样输出的时候就很方便了。
最后关于ByteBuffer在Channel中的使用，可以参考下面的代码：
public class BufferTest {
public static void main(String[] args) throws IOException {
String file = "xxxx/test.txt";
RandomAccessFile accessFile = new RandomAccessFile(file,"rw");
FileChannel fileChannel = accessFile.getChannel();
// 20个字节
ByteBuffer buffer = ByteBuffer.allocate(20);
int bytesRead = fileChannel.read(buffer);
// buffer.put()也能写入buffer
while(bytesRead!=-1){
// 写切换到读
buffer.flip();
while(buffer.hasRemaining()){
System.out.println((char)buffer.get());
}
// buffer.rewind()重新读
// buffer.mark()标记position buffer.reset()恢复
// 清除缓冲区
buffer.clear();
// buffer.compact(); 清楚读过的数据
bytesRead = fileChannel.read(buffer);
}
}
}
这样，就熟悉了Channel和ByteBuffer的使用。接下来，看看服务器中的应用吧。
NIO服务器例子
前面BIO的服务器，是来一个连接就创建一个新的线程响应。这里基于NIO的多路复用，可以这样写：
import java.io.IOException;
import java.net.InetSocketAddress;
import java.net.ServerSocket;
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.Iterator;
import java.util.Set;
public class PlainNioServer {
public void serve(int port) throws IOException {
// 创建channel，并绑定监听端口
ServerSocketChannel serverSocketChannel = ServerSocketChannel.open();
serverSocketChannel.configureBlocking(false);
ServerSocket ssocket = serverSocketChannel.socket();
InetSocketAddress address = new InetSocketAddress(port);
ssocket.bind(address);
//创建selector,并将channel注册到selector
Selector selector = Selector.open();
serverSocketChannel.register(selector, SelectionKey.OP_ACCEPT);
final ByteBuffer msg = ByteBuffer.wrap("Hi\r\b".getBytes());
for(;;){
try{
selector.select();
}catch (IOException e){
e.printStackTrace();
break;
}
Set readyKeys = selector.selectedKeys();
Iterator iterator = readyKeys.iterator();
while(iterator.hasNext()){
SelectionKey key = iterator.next();
iterator.remove();
try{
if(key.isAcceptable()){
ServerSocketChannel server = (ServerSocketChannel)key.channel();
SocketChannel client= server.accept();
client.configureBlocking(false);
client.register(selector, SelectionKey.OP_WRITE | SelectionKey.OP_READ, msg.duplicate());
System.out.println("accepted connection from "+client);
}
if(key.isWritable()){
SocketChannel client = (SocketChannel) key.channel();
ByteBuffer buffer = (ByteBuffer) key.attachment();
while(buffer.hasRemaining()){
if(client.write(buffer)==0){
break;
}
}
client.close();
}
}catch (IOException e){
key.cancel();
try{
key.channel().close();
} catch (IOException ex){
ex.printStackTrace();
}
}
}
}
}
public static void main(String[] args) throws IOException {
PlainNioServer server = new PlainNioServer();
server.serve(5555);
}
}
这里抽象来说是下面的步骤：
- 创建ServerSocketChannel并绑定端口
- 创建Selector多路复用器，并注册Channel
- 循环监听是否有感兴趣的事件发生selector.select();
- 获得事件的句柄，并进行处理
其中Selector可以一次监听多个IO处理，效率就提高很多了。
前面介绍了基本的网络模型以及IO与NIO，那么有了NIO来开发非阻塞服务器，大家就满足了吗？有了技术支持，就回去追求效率，因此就产生了很多NIO的框架对NIO进行封装——这就是大名鼎鼎的Netty。
前几篇的内容，可以参考：
- 网络IO的基本知识与概念
- 普通IO以及BIO服务器
- NIO的使用与服务器Hello world
- Netty的使用与服务器Hello world
为什么要使用开源框架？
这个问题几乎可以当做废话，框架肯定要比一些原生的API封装了更多地功能，重复造轮子在追求效率的情况并不是明智之举。那么先来说说NIO有什么缺点吧：
- NIO的类库和API还是有点复杂，比如Buffer的使用
- Selector编写复杂，如果对某个事件注册后，业务代码过于耦合
- 需要了解很多多线程的知识，熟悉网络编程
- 面对断连重连、保丢失、粘包等，处理复杂
- NIO存在BUG，根据网上言论说是selector空轮训导致CPU飙升，具体有兴趣的可以看看JDK的官网
那么有了这些问题，就急需一些大牛们开发通用框架来方便劳苦大众了。最致命的NIO框架就是MINA和Netty了，这里不得不说个小插曲：
先来看看MINA的主要贡献者：
再来看看NETYY的主要贡献者：
总结起来，有这么几点：
- MINA和Netty的主要贡献者都是同一个人——Trustin lee，韩国Line公司的。
- MINA于2006年开发，到14、15年左右，基本停止维护
- Nety开始于2009年，目前仍由苹果公司的norman maurer在主要维护。
- Norman Maurer是《Netty in Action》一书的作者
因此，如果让你选择你应该知道选择谁了吧。另外，MINA对底层系统要求功底更深，且国内Netty的氛围更好，有李林峰等人在大力宣传（《Netty权威指南》的作者）。
讲了一大堆的废话之后，总结来说就是——Netty有前途，学它准没错。
Netty介绍
按照定义来说，Netty是一个异步、事件驱动的用来做高性能、高可靠性的网络应用框架。主要的优点有：
- 框架设计优雅，底层模型随意切换适应不同的网络协议要求
- 提供很多标准的协议、安全、编码解码的支持
- 解决了很多NIO不易用的问题
- 社区更为活跃，在很多开源框架中使用，如Dubbo、RocketMQ、Spark等
主要支持的功能或者特性有：
- 底层核心有：Zero-Copy-Capable Buffer，非常易用的灵拷贝Buffer（这个内容很有意思，稍后专门来说）；统一的API；标准可扩展的时间模型
- 传输方面的支持有：管道通信（具体不知道干啥的，还请老司机指教）；Http隧道；TCP与UDP
- 协议方面的支持有：基于原始文本和二进制的协议；解压缩；大文件传输；流媒体传输；protobuf编解码；安全认证；http和websocket
总之提供了很多现成的功能可以直接供开发者使用。
Netty服务器小例子
基于Netty的服务器编程可以看做是Reactor模型：
即包含一个接收连接的线程池（也有可能是单个线程，boss线程池）以及一个处理连接的线程池（worker线程池）。boss负责接收连接，并进行IO监听；worker负责后续的处理。为了便于理解Netty，直接看看代码:
package cn.xingoo.book.netty.chap04;
import io.netty.bootstrap.ServerBootstrap;
import io.netty.buffer.ByteBuf;
import io.netty.buffer.Unpooled;
import io.netty.channel.*;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioServerSocketChannel;
import java.net.InetSocketAddress;
import java.nio.charset.Charset;
public class NettyNioServer {
public void serve(int port) throws InterruptedException {
final ByteBuf buffer = Unpooled.unreleasableBuffer(Unpooled.copiedBuffer("Hi\r\n", Charset.forName("UTF-8")));
// 第一步，创建线程池
EventLoopGroup bossGroup = new NioEventLoopGroup(1);
EventLoopGroup workerGroup = new NioEventLoopGroup();
try{
// 第二步，创建启动类
ServerBootstrap b = new ServerBootstrap();
// 第三步，配置各组件
b.group(bossGroup, workerGroup)
.channel(NioServerSocketChannel.class)
.localAddress(new InetSocketAddress(port))
.childHandler(new ChannelInitializer() {
@Override
protected void initChannel(SocketChannel socketChannel) throws Exception {
socketChannel.pipeline().addLast(new ChannelInboundHandlerAdapter(){
@Override
public void channelActive(ChannelHandlerContext ctx) throws Exception {
ctx.writeAndFlush(buffer.duplicate()).addListener(ChannelFutureListener.CLOSE);
}
});
}
});
// 第四步，开启监听
ChannelFuture f = b.bind().sync();
f.channel().closeFuture().sync();
} finally {
bossGroup.shutdownGracefully().sync();
workerGroup.shutdownGracefully().sync();
}
}
public static void main(String[] args) throws InterruptedException {
NettyNioServer server = new NettyNioServer();
server.serve(5555);
}
}
代码非常少，而且想要换成阻塞IO，只需要替换Channel里面的工厂类即可：
public class NettyOioServer {
public void serve(int port) throws InterruptedException {
final ByteBuf buf = Unpooled.unreleasableBuffer(Unpooled.copiedBuffer("Hi\r\b", Charset.forName("UTF-8")));
EventLoopGroup bossGroup = new OioEventLoopGroup(1);
EventLoopGroup workerGroup = new OioEventLoopGroup();
try{
ServerBootstrap b = new ServerBootstrap();
b.group(bossGroup, workerGroup)//配置boss和worker
.channel(OioServerSocketChannel.class) // 使用阻塞的SocketChannel
....
概括来说，在Netty中包含下面几个主要的组件：
- Bootstrap：netty的组件容器，用于把其他各个部分连接起来；如果是TCP的Server端，则为ServerBootstrap.
- Channel：代表一个Socket的连接
- EventLoopGroup：一个Group包含多个EventLoop，可以理解为线程池
- EventLoop：处理具体的Channel，一个EventLoop可以处理多个Channel
- ChannelPipeline：每个Channel绑定一个pipeline，在上面注册处理逻辑handler
- Handler：具体的对消息或连接的处理，有两种类型，Inbound和Outbound。分别代表消息接收的处理和消息发送的处理。
- ChannelFuture：注解回调方法
了解上面的基本组件后，就看一下几个重要的内容。
Netty的Buffer和零拷贝
在Unix操作系统中，系统底层可以基于mmap实现内核空间和用户空间的内存映射。但是在Netty中并不是这个意思，它主要来自于下面几个功能：
- 通过Composite和slice实现逻辑上的Buffer的组合和拆分，重新维护索引，避免内存拷贝过程。
- 通过DirectBuffer申请堆外内存，避免用户空间的拷贝。不过堆外内存的申请和释放都很麻烦，推荐小心使用。关于堆外内存的一些研究，还可以参考执勤的分享：Java堆外内存之突破JVM枷锁 以及 Java直接内存与非直接内存性能测试
- 通过FileRegion包装FileChannel，直接实现channel到channel的传输。
另外，Netty自己封装实现了ByteBuf，相比于Nio原生的ByteBuffer，API上更易用了；同时支持容量的动态扩容；另外还支持Buffer的池化，高效复用Buffer。
public class ByteBufTest {
public static void main(String[] args) {
//创建bytebuf
ByteBuf buf = Unpooled.copiedBuffer("hello".getBytes());
System.out.println(buf);
// 读取一个字节
buf.readByte();
System.out.println(buf);
// 读取一个字节
buf.readByte();
System.out.println(buf);
// 丢弃无用数据
buf.discardReadBytes();
System.out.println(buf);
// 清空
buf.clear();
System.out.println(buf);
// 写入
buf.writeBytes("123".getBytes());
System.out.println(buf);
buf.markReaderIndex();
System.out.println("mark:"+buf);
buf.readByte();
buf.readByte();
System.out.println("read:"+buf);
buf.resetReaderIndex();
System.out.println("reset:"+buf);
}
}
输出为：
UnpooledHeapByteBuf(ridx: 0, widx: 5, cap: 5/5)
UnpooledHeapByteBuf(ridx: 1, widx: 5, cap: 5/5)
UnpooledHeapByteBuf(ridx: 2, widx: 5, cap: 5/5)
UnpooledHeapByteBuf(ridx: 0, widx: 3, cap: 5/5)
UnpooledHeapByteBuf(ridx: 0, widx: 0, cap: 5/5)
UnpooledHeapByteBuf(ridx: 0, widx: 3, cap: 5/5)
mark:UnpooledHeapByteBuf(ridx: 0, widx: 3, cap: 5/5)
read:UnpooledHeapByteBuf(ridx: 2, widx: 3, cap: 5/5)
reset:UnpooledHeapByteBuf(ridx: 0, widx: 3, cap: 5/5)
有兴趣的可以看一下上一篇分享的ByteBuffer，对比一下，就能发现在Netty中通过独立的读写索引维护，避免读写模式的切换，更加方便了。
Handler的使用
前面介绍了Handler包含了Inbound和Outbound两种，他们统一放在一个双向链表中：
当接收消息的时候，会从链表的表头开始遍历，如果是inbound就调用对应的方法；如果发送消息则从链表的尾巴开始遍历。那么上面途中的例子，接收消息就会输出：
InboundA --> InboundB --> InboundC
输出消息，则会输出：
OutboundC --> OutboundB --> OutboundA
这里有段代码，可以直接复制下来，试试看：
package cn.xingoo.book.netty.pipeline;
import io.netty.bootstrap.ServerBootstrap;
import io.netty.buffer.ByteBuf;
import io.netty.buffer.Unpooled;
import io.netty.channel.*;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioServerSocketChannel;
import java.net.InetSocketAddress;
import java.net.SocketAddress;
import java.nio.charset.Charset;
/**
* 注意：
*
* 1 ChannelOutboundHandler要在最后一个Inbound之前
*
*/
public class NettyNioServerHandlerTest {
final static ByteBuf buffer = Unpooled.unreleasableBuffer(Unpooled.copiedBuffer("Hi\r\n", Charset.forName("UTF-8")));
public void serve(int port) throws InterruptedException {
EventLoopGroup bossGroup = new NioEventLoopGroup(1);
EventLoopGroup workerGroup = new NioEventLoopGroup();
try{
ServerBootstrap b = new ServerBootstrap();
b.group(bossGroup, workerGroup)
.channel(NioServerSocketChannel.class)
.localAddress(new InetSocketAddress(port))
.childHandler(new ChannelInitializer<SocketChannel>() {
@Override
protected void initChannel(SocketChannel socketChannel) throws Exception {
ChannelPipeline pipeline = socketChannel.pipeline();
pipeline.addLast("1",new InboundA());
pipeline.addLast("2",new OutboundA());
pipeline.addLast("3",new InboundB());
pipeline.addLast("4",new OutboundB());
pipeline.addLast("5",new OutboundC());
pipeline.addLast("6",new InboundC());
}
});
ChannelFuture f = b.bind().sync();
f.channel().closeFuture().sync();
} finally {
bossGroup.shutdownGracefully().sync();
workerGroup.shutdownGracefully().sync();
}
}
public static void main(String[] args) throws InterruptedException {
NettyNioServerHandlerTest server = new NettyNioServerHandlerTest();
server.serve(5555);
}
private static class InboundA extends ChannelInboundHandlerAdapter {
@Override
public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
ByteBuf buf = (ByteBuf)msg;
System.out.println("InboundA read"+buf.toString(Charset.forName("UTF-8")));
super.channelRead(ctx, msg);
}
}
private static class InboundB extends ChannelInboundHandlerAdapter {
@Override
public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
ByteBuf buf = (ByteBuf)msg;
System.out.println("InboundB read"+buf.toString(Charset.forName("UTF-8")));
super.channelRead(ctx, msg);
// 从pipeline的尾巴开始找outbound
ctx.channel().writeAndFlush(buffer);
}
}
private static class InboundC extends ChannelInboundHandlerAdapter {
@Override
public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
ByteBuf buf = (ByteBuf)msg;
System.out.println("InboundC read"+buf.toString(Charset.forName("UTF-8")));
super.channelRead(ctx, msg);
// 这样会从当前的handler向前找outbound
//ctx.writeAndFlush(buffer);
}
}
private static class OutboundA extends ChannelOutboundHandlerAdapter {
@Override
public void write(ChannelHandlerContext ctx, Object msg, ChannelPromise promise) throws Exception {
System.out.println("OutboundA write");
super.write(ctx, msg, promise);
}
}
private static class OutboundB extends ChannelOutboundHandlerAdapter {
@Override
public void write(ChannelHandlerContext ctx, Object msg, ChannelPromise promise) throws Exception {
System.out.println("OutboundB write");
super.write(ctx, msg, promise);
}
}
private static class OutboundC extends ChannelOutboundHandlerAdapter {
@Override
public void write(ChannelHandlerContext ctx, Object msg, ChannelPromise promise) throws Exception {
System.out.println("OutboundC write");
super.write(ctx, msg, promise);
}
}
}
最后有一个TCP粘包的例子，有兴趣的也可以自己试一下，代码就不贴上来了，可以参考最后面的Github连接。