Netty入门基础知识
简介
Netty是一款高性能java网络编程框架，被广泛应用在中间件、直播、社交、游戏等领域。Netty对java NIO进行高级封装，简化了网络应用的开发过程。
stream与channel
stream不会自动缓冲数据，channel会利用系统提供的发送缓冲区，接收缓冲区（更为底层）
stream仅支持阻塞API,channel同时支持阻塞、非阻塞API，网络channel可配置selector实现多路复用
两者均为全双工，即读写可以同时进行。
IO模型
IO请求：IO调用阶段:用户进程向内核发起系统调用。IO执行阶段:内核等待IO请求处理完成返回
同步阻塞IO(BIO)
应用向内核发起IO请求，发起调用的线程一直等待IO请求结果，直到数据处理完并返回阻塞才释放
同步非阻塞IO (NIO)
应用先内核发起IO请求，不断的轮询其请求结果，当内核处理好结果后返回给用户态的缓冲区中，在操作系统内核复制数据到用户态程序的时候还是会阻塞。
IO多路复用
多路复用其实就是java的selector,selector会检测操作系统内核是否会有io事件发生（会阻塞），如果有io事件发生就会从内核中向用户态程序发送通知。当用户线程接收到通知后，就会发起read或write操作，内核就会复制数据，当复制数据的过程中还是会阻塞。与BIO相比selector可以检测多个channel的不同事件（BIO在处理了一个channel的事件时不能处理另一个channel的事件需要等待第一个事件的处理完成）有多个channel的时候内核会将多个事件一次性交个selector进行处理后续就不再需要selector在等待事件的发生了。
信号驱动IO
异步IO
同步：线程自己去获取结果（一个线程）
异步：线程自己不去获取结果，而是由其他线程发送结果（至少两个线程）
总结
当调用一次channel.read或者stream.read后，会切换至操作系统内核态来完成真正的数据读取。而读取有分为两个阶段：等待数据阶段、复制数据阶段。
Netty实现的IO模型
基于非阻塞IO实现，底层依赖的是JDK NIO框架的多路复用器Selector，一个多路复用器Selector可以同时轮询多个Channel，当数据达到就绪状态后需要一个
事件分发器，事件分发器负责将读写事件分发给对应的读写事件处理器。事件分发器采用两种设计模式，Reactor和Proactor
NIO三大组件
Channel & Buffer
channel类似与stream,他就是读写数据的双向通道，可以从channel将数据读入buffer，也可以将buffer的数据写入channel。
常见的channel
FileChannel、DatagramChannel、SocketChannel、ServerSocketChannel。
常见的Buffer
ByteBuffer: MappedByteBuffer、DirectByteBuffer、HeapByteBuffer
ShortBuffer
IntBuffer
LongBuffer
FloatBuffer
DoubleBuffer
CharBuffer
Buffer基本使用
// FileChannel
// 通过输入输入输出流，或者RandomAccessFile
try (FileChannel channel = new FileInputStream("src/test/resources/data.txt").getChannel()) {
// 准备10大小的缓冲区
ByteBuffer buffer = ByteBuffer.allocate(10);
// 从channel读取数据，向buffer写入
while (true) {
int read = channel.read(buffer);
log.debug("读取到的字节: {}", read);
if (read == -1) {
break;
}
// 打印buffer内容
buffer.flip(); // 切换读模式
while (buffer.hasRemaining()) { // 有下一个字节
byte b = buffer.get();// 读一个字节
System.out.print((char) b);
}
// 切换为写模式
buffer.clear();
}
} catch (Exception e) {
e.printStackTrace();
}
ByteBuffer正确使用:
1、向buffer写入数据，例如调用channel.read(buffer);
2、调用flip()切换为读模式
3、从buffer读取数据，例如调用buffer.get()
4、调用clear()或者compact()切换为写模式，重复1~4步骤
Buffer内部结构
ByteBuffer有以下属性
1、capacity: ByteBuffer容量
2、position:读写指针
3、limit: 写入限制的容量
写模式下，position是写入位置，limit对于容量大小
调用flip函数后，position切换为读取位置，limit切换为读取限制
读取后
调用clear后
调用comact函数后,是把未读完的部分向前压缩，然后切换为写模式
allocate与allocateDirect
ByteBuffer.allocate(16)的Class为java.nio.HeapByteBuffer。使用的是java堆内存进行分配，读写效率较低，受到GC影响
ByteBuffer.allocateDirect(16)的Class为java.nio.DirectByteBuffer。使用的是直接内存进行分配（分配内存的效率比较低），读写效率较高（少一次数据的拷贝），不受到GC影响
写入Buffer
调用Channel的read方法
channel.read(buffer);
调用buffer自己的put方法
channel.put((byte) 127);
从Buffer读取
调用channel的write方法
int wirteBytes = channel.write(buffer);
使用buffer自己的get方法
byte b = buffer.get();
调用get方法会使position读指针向后移动，如果想重复读取数据，可以调用rewind方法将position重置为0，或者调用get(int i)的方法获取索引
i的内容,他不会移动position指针。
示例代码
@Test
public void testRewind() {
ByteBuffer buffer = ByteBuffer.allocate(10);
buffer.put(new byte[] { 'a', 'b', 'c', 'd' });
// 切换读模式
buffer.flip();
// 从头开始读
buffer.get(new byte[4]);
System.out.println(buffer); // 输出结果为: java.nio.HeapByteBuffer[pos=4 lim=4 cap=10]
buffer.rewind();
System.out.println(buffer); // 输出结果为: java.nio.HeapByteBuffer[pos=0 lim=4 cap=10]
}
Mark与Reset
mark作为记录position的位置，reset是将position重置到mark的位置
示例代码
@Test
public void testRewind() {
ByteBuffer buffer = ByteBuffer.allocate(10);
buffer.put(new byte[] { 'a', 'b', 'c', 'd' });
// 切换读模式
buffer.flip();
System.out.println((char) buffer.get()); // 输出结果为: a
buffer.mark(); // 记录‘b’的位置
System.out.println((char) buffer.get()); // 输出结果为: b
System.out.println((char) buffer.get()); // 输出结果为: c
buffer.reset(); // 重置到'b'的位置
System.out.println((char) buffer.get()); // 输出结果为: b
}
Buffer与字符串互转
字符串转Buffer
ByteBuffer buffer = ByteBuffer.allocate(16);
buffer.put("hello world".getBytes());
由于使用buffer的时候没有调用
flip所以一直处于写模式
使用Charset方式
// Charset转换方式
ByteBuffer buffer = StandardCharsets.UTF_8.encode("hello");
System.out.println(buffer);
使用StandardCharsets会自动切换到读模式
使用Wrap方式
// wrap
ByteBuffer buffer = ByteBuffer.wrap("hello".getBytes());
System.out.println(buffer);
会自动切换到读模式
Buffer转字符串
ByteBuffer buffer = StandardCharsets.UTF_8.encode("hello");
System.out.println(buffer);
String s = StandardCharsets.UTF_8.decode(buffer).toString();
System.out.println(s);
由于使用
Charset和Wrap的方式会自动切换为读模式，所以可以直接读取。如果还在处于写模式，会任何数据都读取不到。
ByteBuffer buffer = ByteBuffer.allocate(16);
buffer.put("hello world".getBytes());
System.out.println(buffer);
buffer.flip(); // 切换到读模式
String s = StandardCharsets.UTF_8.decode(buffer).toString();
System.out.println(s);
必须切换到读模式
分散读集中写
Buffer调试工具类
依赖
<dependency>
<groupId>io.netty</groupId>
<artifactId>netty-all</artifactId>
<version>4.1.51.Final</version>
</dependency>
package org.example;
import java.nio.ByteBuffer;
import io.netty.util.internal.MathUtil;
import io.netty.util.internal.StringUtil;
/**
* @author Panwen Chen
* @date 2021/4/12 15:59
*/
public class ByteBufferUtil {
private static final char[] BYTE2CHAR = new char[256];
private static final char[] HEXDUMP_TABLE = new char[256 * 4];
private static final String[] HEXPADDING = new String[16];
private static final String[] HEXDUMP_ROWPREFIXES = new String[65536 >>> 4];
private static final String[] BYTE2HEX = new String[256];
private static final String[] BYTEPADDING = new String[16];
static {
final char[] DIGITS = "0123456789abcdef".toCharArray();
for (int i = 0; i < 256; i++) {
HEXDUMP_TABLE[i << 1] = DIGITS[i >>> 4 & 0x0F];
HEXDUMP_TABLE[(i << 1) + 1] = DIGITS[i & 0x0F];
}
int i;
// Generate the lookup table for hex dump paddings
for (i = 0; i < HEXPADDING.length; i++) {
int padding = HEXPADDING.length - i;
StringBuilder buf = new StringBuilder(padding * 3);
for (int j = 0; j < padding; j++) {
buf.append(" ");
}
HEXPADDING[i] = buf.toString();
}
// Generate the lookup table for the start-offset header in each row (up to 64KiB).
for (i = 0; i < HEXDUMP_ROWPREFIXES.length; i++) {
StringBuilder buf = new StringBuilder(12);
buf.append(StringUtil.NEWLINE);
buf.append(Long.toHexString(i << 4 & 0xFFFFFFFFL | 0x100000000L));
buf.setCharAt(buf.length() - 9, '|');
buf.append('|');
HEXDUMP_ROWPREFIXES[i] = buf.toString();
}
// Generate the lookup table for byte-to-hex-dump conversion
for (i = 0; i < BYTE2HEX.length; i++) {
BYTE2HEX[i] = ' ' + StringUtil.byteToHexStringPadded(i);
}
// Generate the lookup table for byte dump paddings
for (i = 0; i < BYTEPADDING.length; i++) {
int padding = BYTEPADDING.length - i;
StringBuilder buf = new StringBuilder(padding);
for (int j = 0; j < padding; j++) {
buf.append(' ');
}
BYTEPADDING[i] = buf.toString();
}
// Generate the lookup table for byte-to-char conversion
for (i = 0; i < BYTE2CHAR.length; i++) {
if (i <= 0x1f || i >= 0x7f) {
BYTE2CHAR[i] = '.';
} else {
BYTE2CHAR[i] = (char) i;
}
}
}
/**
* 打印所有内容
* @param buffer
*/
public static void debugAll(ByteBuffer buffer) {
int oldlimit = buffer.limit();
buffer.limit(buffer.capacity());
StringBuilder origin = new StringBuilder(256);
appendPrettyHexDump(origin, buffer, 0, buffer.capacity());
System.out.println("+--------+-------------------- all ------------------------+----------------+");
System.out.printf("position: [%d], limit: [%d]\n", buffer.position(), oldlimit);
System.out.println(origin);
buffer.limit(oldlimit);
}
/**
* 打印可读取内容
* @param buffer
*/
public static void debugRead(ByteBuffer buffer) {
StringBuilder builder = new StringBuilder(256);
appendPrettyHexDump(builder, buffer, buffer.position(), buffer.limit() - buffer.position());
System.out.println("+--------+-------------------- read -----------------------+----------------+");
System.out.printf("position: [%d], limit: [%d]\n", buffer.position(), buffer.limit());
System.out.println(builder);
}
private static void appendPrettyHexDump(StringBuilder dump, ByteBuffer buf, int offset, int length) {
if (MathUtil.isOutOfBounds(offset, length, buf.capacity())) {
throw new IndexOutOfBoundsException(
"expected: " + "0 <= offset(" + offset + ") <= offset + length(" + length
+ ") <= " + "buf.capacity(" + buf.capacity() + ')');
}
if (length == 0) {
return;
}
dump.append(
" +-------------------------------------------------+" +
StringUtil.NEWLINE + " | 0 1 2 3 4 5 6 7 8 9 a b c d e f |" +
StringUtil.NEWLINE + "+--------+-------------------------------------------------+----------------+");
final int startIndex = offset;
final int fullRows = length >>> 4;
final int remainder = length & 0xF;
// Dump the rows which have 16 bytes.
for (int row = 0; row < fullRows; row++) {
int rowStartIndex = (row << 4) + startIndex;
// Per-row prefix.
appendHexDumpRowPrefix(dump, row, rowStartIndex);
// Hex dump
int rowEndIndex = rowStartIndex + 16;
for (int j = rowStartIndex; j < rowEndIndex; j++) {
dump.append(BYTE2HEX[getUnsignedByte(buf, j)]);
}
dump.append(" |");
// ASCII dump
for (int j = rowStartIndex; j < rowEndIndex; j++) {
dump.append(BYTE2CHAR[getUnsignedByte(buf, j)]);
}
dump.append('|');
}
// Dump the last row which has less than 16 bytes.
if (remainder != 0) {
int rowStartIndex = (fullRows << 4) + startIndex;
appendHexDumpRowPrefix(dump, fullRows, rowStartIndex);
// Hex dump
int rowEndIndex = rowStartIndex + remainder;
for (int j = rowStartIndex; j < rowEndIndex; j++) {
dump.append(BYTE2HEX[getUnsignedByte(buf, j)]);
}
dump.append(HEXPADDING[remainder]);
dump.append(" |");
// Ascii dump
for (int j = rowStartIndex; j < rowEndIndex; j++) {
dump.append(BYTE2CHAR[getUnsignedByte(buf, j)]);
}
dump.append(BYTEPADDING[remainder]);
dump.append('|');
}
dump.append(StringUtil.NEWLINE +
"+--------+-------------------------------------------------+----------------+");
}
private static void appendHexDumpRowPrefix(StringBuilder dump, int row, int rowStartIndex) {
if (row < HEXDUMP_ROWPREFIXES.length) {
dump.append(HEXDUMP_ROWPREFIXES[row]);
} else {
dump.append(StringUtil.NEWLINE);
dump.append(Long.toHexString(rowStartIndex & 0xFFFFFFFFL | 0x100000000L));
dump.setCharAt(dump.length() - 9, '|');
dump.append('|');
}
}
public static short getUnsignedByte(ByteBuffer buffer, int index) {
return (short) (buffer.get(index) & 0xFF);
}
}
分散读
try (FileChannel channel = new RandomAccessFile("src/test/resources/data.txt", "r").getChannel()) {
ByteBuffer b1 = ByteBuffer.allocate(10);
ByteBuffer b2 = ByteBuffer.allocate(4);
channel.read(new ByteBuffer[]{ b1, b2 });
b1.flip();
b2.flip();
ByteBufferUtil.debugAll(b1);
ByteBufferUtil.debugAll(b2);
} catch (IOException e) {
}
data.txt的内容如下:
1234567890abcd
集中写入
ByteBuffer b1 = StandardCharsets.UTF_8.encode("hello");
ByteBuffer b2 = StandardCharsets.UTF_8.encode("world");
try (FileChannel channel = new RandomAccessFile("src/test/resources/data1.txt", "rw").getChannel()) {
channel.write(new ByteBuffer[]{ b1, b2 });
} catch (Exception e) {
}
data1.txt的内容如下:
helloworld
黏包半包解析
黏包
粘包问题是指数据在传输时，在一条消息中读取到了另一条消息的部分数据，这种现象就叫做粘包。 比如发送了两条消息，分别为“ABC”和“DEF”，那么正常情况下接收端也应该收到两条消息“ABC”和“DEF”，但接收端却收到的是“ABCD”，像这种情况就叫做粘包
半包
半包问题是指接收端只收到了部分数据，而非完整的数据的情况就叫做半包。比如发送了一条消息是“ABC”，而接收端却收到的是“AB”和“C”两条信息，这种情况就叫做半包
示例代码
@Test
public void sendMessage() {
ByteBuffer buffer = ByteBuffer.allocate(32);
buffer.put("Hello,world\nI'm tang\nHo".getBytes());
splitMessage(buffer);
buffer.put("w are you?\n".getBytes());
splitMessage(buffer);
}
public void splitMessage(ByteBuffer buffer) {
buffer.flip();
for (int i = 0; i < buffer.limit(); i++) {
// 找到一条信息的结尾
if (buffer.get(i) == '\n') {
// 存储到长度为一条信息的结尾索引加一减去buffer起始索引的ByteBuffer中
int len = i + 1 - buffer.position();
ByteBuffer target = ByteBuffer.allocate(len);
for (int j = 0; j < len; j++) {
target.put(buffer.get());
}
ByteBufferUtil.debugAll(target);
}
}
buffer.compact();
}
FileChannel初步了解
FileChannel只能工作在阻塞模式下。FileChannel不能直接打开，必须通过
FileInputStream、FileOutputStream、RandomAccessFile来获取FileChannel，它们都有getChannel方法通过
FileInputStream获取的channel只能读通过
FileOutputStream获取的channel只能写通过
RandomAccessFile是否能读写根据RandomAccessFile的构造函数的参数来决定的。
transferTo
transferTo方法的作用是从一个channel传输到另一个channel, FileChannel效率高，底层会使用零拷贝进行优化，最多能传输2G的数据
try (FileChannel from = new FileInputStream("src/test/resources/data.txt").getChannel();
FileChannel to = new FileOutputStream("src/test/resources/to.txt").getChannel()) {
from.transferTo(0, from.size(), to);
} catch (Exception e) {
e.printStackTrace();
}
FileChannel传输大于2G的数据
try (FileChannel from = new FileInputStream("src/test/resources/data.txt").getChannel();
FileChannel to = new FileOutputStream("src/test/resources/to.txt").getChannel()) {
long size = from.size();
for (long left = size; left > 0; ) {
System.out.println("position: " + (size - left) + " left: " + left);
left -= from.transferTo((size - left), left, to);
}
} catch (Exception e) {
e.printStackTrace();
}
Path对象
jdk7 引入了Path和Paths类。Path用来表示文件路径，Paths是工具类，用来获取Path实例
Path source = Paths.get("test.txt"); // 相对路径，使用user.dir 环境变量来定位test.txt
Path source = Paths.get("d:\\test.txt"); // 绝对路径，代表了d:\test.txt
Path source = Paths.get("d:/test.txt"); // 绝对路径
Path projects = Paths.get("d:\\data", "test"); // 代表了d:\data\test
Path path = Paths.get("d:\\data\\test\\a\\..\\b");
System.out.println(path.normalize()); // 正常化路径 输出d:\data\test\b
Files遍历文件夹
@Test
public void WalkFileTreeTest() throws IOException {
AtomicInteger dirCount = new AtomicInteger();
AtomicInteger fileCount = new AtomicInteger();
Files.walkFileTree(Paths.get("/Users/tang"), new SimpleFileVisitor<Path>() {
// 进入文件夹之前
@Override
public FileVisitResult preVisitDirectory(Path dir, BasicFileAttributes attrs) throws IOException {
System.out.println("===> " + dir);
// dirCount++
dirCount.incrementAndGet();
return super.preVisitDirectory(dir, attrs);
}
// 进入文件夹时校验文件
@Override
public FileVisitResult visitFile(Path file, BasicFileAttributes attrs) throws IOException {
System.out.println("file: " + file);
// fileCount++
fileCount.incrementAndGet();
return super.visitFile(file, attrs);
}
});
System.out.println("dirCount: " + dirCount.get() + " fileCount: " + fileCount.get());
}
Files遍历删除
@Test
public void walkFileTreeDeleteTest() throws IOException {
Files.walkFileTree(Paths.get("/Users/tang/Downloads/柏拉图 - 搜索_files"), new SimpleFileVisitor<Path>() {
@Override
public FileVisitResult preVisitDirectory(Path dir, BasicFileAttributes attrs) throws IOException {
System.out.println("==> 进入: " + dir);
return super.preVisitDirectory(dir, attrs);
}
@Override
public FileVisitResult visitFile(Path file, BasicFileAttributes attrs) throws IOException {
System.out.println("delete file: " + file);
Files.delete(file);
return super.visitFile(file, attrs);
}
@Override
public FileVisitResult postVisitDirectory(Path dir, IOException exc) throws IOException {
System.out.println("delete: " + dir);
Files.delete(dir);
return super.postVisitDirectory(dir, exc);
}
});
}
Files遍历拷贝
@Test
public void walkFileCopyTest() throws IOException {
String source = "/Users/tang/Downloads/xc-ui-pc-static-portal";
String target = "/Users/tang/Downloads/xc-ui-pc-static-portal-副本";
Files.walk(Paths.get(source)).forEach(path -> {
String targetName = path.toString().replace(source, target);
// 是目录
if (Files.isDirectory(path)) {
try {
Files.createDirectory(Paths.get(targetName));
} catch (IOException e) {
throw new RuntimeException(e);
}
} else if (Files.isRegularFile(path)) {
// 是文件
try {
Files.copy(path, Paths.get(targetName));
} catch (IOException e) {
throw new RuntimeException(e);
}
}
});
}
阻塞与非阻塞
阻塞
1、在没有数据可读时，包括数据复制过程中，线程必须阻塞等待，不会占用CPU,但线程相当于闲置状态
2、32位JVM一个线程320k，64位JVM一个线程1024k，为了减少线程数，需要采用线程池技术
3、但即使用了线程池，如果有多个连接建立，但很长时间inactive,会阻塞线程池中的所有线程
非阻塞
1、在某个Channel没有可读事件是，线程不必阻塞，它可以去处理其他有可读事件的Channel
2、数据复制过程中，线程实际还是阻塞的
3、写数据时，线程只能等待数据写入的Channel即可，无需等Channel通过网络把数据发送出去
代码示例
@Test
public void nonBlockTest() throws IOException {
ByteBuffer buffer = ByteBuffer.allocate(16);
ServerSocketChannel channel = ServerSocketChannel.open();
// 设置非阻塞模式，默认为true,如果设置为true，则accept方法会一直阻塞，直到有客户端连接
channel.configureBlocking(false);
channel.bind(new InetSocketAddress(8080));
List<SocketChannel> channels = new ArrayList<>();
while (true) {
log.debug("connecting.....");
SocketChannel socketChannel = channel.accept();
if (socketChannel != null) {
log.debug("connected......{}", socketChannel);
// 设置非阻塞模式，默认true,如果设置为true，则read方法会一直阻塞，直到有数据可读
socketChannel.configureBlocking(false);
channels.add(socketChannel);
}
for (SocketChannel sc : channels) {
log.debug("before read....{}", sc);
int read = sc.read(buffer);
if (read > 0) {
buffer.flip();
buffer.clear();
log.debug("after read....{}", sc);
}
}
}
}
使用异步操作虽然解决了线程堵塞的问题，但当没有连接请求发送到服务器的时候while(true)循环会不断的循环且不处理任何事情，这会导致CPU资源大量浪费了。
Selector的使用
selector通过管理检测channel中的事件有无发生，channel中有事件发生了，selector就会获得这些事件并使用线程进行处理，如果没有事件发生selector就会阻塞线程。
事件有accept（在有连接请求时触发）、connect（客户端连接建立后触发）、read（客户端发送数据，服务端通过read事件读取）、write（可写事件）这四种类型。
selectKey移除问题
示例代码
@Test
public void selectorTest() throws IOException {
// 创建selector 可以管理多个channel
Selector selector = Selector.open();
ServerSocketChannel serverSocketChannel = ServerSocketChannel.open();
// 设置非阻塞模式，默认为true,如果设置为true，则accept方法会一直阻塞，直到有客户端连接
serverSocketChannel.configureBlocking(false);
serverSocketChannel.bind(new InetSocketAddress(8080));
// 将channel注册到selector上，并指定事件为accept
// serverKey就是将来事件发生后，通过它可以找到事件和哪个channel发生的事件
SelectionKey serverKey = serverSocketChannel.register(selector, 0, null);
// 关注的是该channel的accept事件
serverKey.interestOps(SelectionKey.OP_ACCEPT);
log.debug("register key: {}", serverKey);
while (true) {
// select方法会一直阻塞，直到有事件发生
// select在事件未处理时是不会阻塞
selector.select();
// 获取所有发生的事件
Iterator<SelectionKey> keys = selector.selectedKeys().iterator();
while (keys.hasNext()) {
// key与serverKey一致
SelectionKey key = keys.next();
log.debug("key: {}", key);
// 移除事件
keys.remove();
// 处理事件
if (key.isAcceptable()) {
// 处理accept事件
ServerSocketChannel server = (ServerSocketChannel) key.channel();
log.debug("server socket: {}", server);
SocketChannel socketChannel = server.accept();
socketChannel.configureBlocking(false);
SelectionKey socketKey = socketChannel.register(selector, 0, null);
socketKey.interestOps(SelectionKey.OP_READ);
log.debug("{}", socketKey);
} else if (key.isReadable()) {
// 处理read事件 获取触发事件的channel
SocketChannel socketChannel = (SocketChannel) key.channel();
ByteBuffer buffer = ByteBuffer.allocate(16);
socketChannel.read(buffer);
buffer.flip();
ByteBufferUtil.debugAll(buffer);
}
}
}
}
Selector selector = Selector.open();
当执行该代码selector会创建一个集合存储SelectionKey，SelectionKey用来管理channel。
SelectionKey serverKey = serverSocketChannel.register(selector, 0, null);
// 关注的是该channel的accept事件
serverKey.interestOps(SelectionKey.OP_ACCEPT);
当执行该代码会往selector集合里存储一个SelectionKey，并且它检测的事件类型是accept
selector.select();
当有客户端来进行连接时，select不会堵塞会创建selectedKeys集合（该集合和selector集合不一致，并且SelectionKeys集合不会主动删除内容，只会添加内容）中会存储发生这个事件的SelectionKey对象
SocketChannel socketChannel = server.accept();
SelectionKey socketKey = socketChannel.register(selector, 0, null);
socketKey.interestOps(SelectionKey.OP_READ);
当该事件被处理后，socketChannel.register(selector, 0, null)会将处理该事件的channel注册到selector集合中，并且检测的是read事件。本次while循环结束，当客户端有数据发送给服务器的时候。
如果没有调用keys.remove();那么在selectedKeys集合中既有上次处理的事件的selectedKey对象也有本次处理read事件的selectedKey对象。
所以key.isAcceptable()条件满足，但这时客户端只是向服务器写数据，并没有发起连接请求，所以server.accept();返回的空对象从而引发异常。
当客户端强制强制断开，需要使用key.cancel();从selector集合中移除。
@Test
public void selectorTest() {
// 创建selector 可以管理多个channel
Selector selector = Selector.open();
//.. 与示例代码一致，代码逻辑省略
while (true) {
//.. 与示例代码一致，代码逻辑省略
while (keys.hasNext()) {
//.. 与示例代码一致，代码逻辑省略
if (key.isAcceptable()) {
//.. 与示例代码一致，代码逻辑省略
} else if (key.isReadable()) {
try {
// 处理read事件 获取触发事件的channel
SocketChannel socketChannel = (SocketChannel) key.channel();
ByteBuffer buffer = ByteBuffer.allocate(16);
int read = socketChannel.read(buffer);
if (read > 0) {
splitMessage(buffer);
} else if (read == -1) {
key.channel();
}
} catch (Exception e) {
e.printStackTrace();
// 由于客户端强制断开所以需要key取消，从selector的key集合中移除
key.cancel();
}
}
}
}
}
客户端主动断开
InetSocketAddress localhost = new InetSocketAddress("localhost", 8080);
SocketChannel channel = SocketChannel.open();
channel.connect(localhost);
channel.close();
客户端的SocketChannel调用close，则客户端主动断开。
服务端代码
SocketChannel socketChannel = (SocketChannel) key.channel();
ByteBuffer buffer = ByteBuffer.allocate(16);
int read = socketChannel.read(buffer);
当客户端主动断开时，服务端的
socketChannel.read(buffer);返回-1,可以根据该返回值取消key。
消息边界问题
解决方法：
1、客户端和服务器都约定好的消息长度，但缺点是浪费带宽
2、使用分隔符拆分消息，缺点就是效率低，需要判断每个字符是否是分隔符
3、TLV格式，即Type类型，Length长度，Value数据，类型和长度已知的情况下，就可以方便获取消息大小，配合适合的buffer，缺点是buffer需要提前分配，通过内容过大，则影响server吞吐量
http 1.1使用的是TLV格式
http 2.0使用的是LTV格式
消息分割
public void splitMessage(ByteBuffer buffer) {
buffer.flip();
for (int i = 0; i < buffer.limit(); i++) {
// 找到一条信息的结尾
if (((char) buffer.get(i)) == '\n') {
// 存储到长度为一条信息的结尾索引加一减去buffer起始索引的ByteBuffer中
int len = i + 1 - buffer.position();
ByteBuffer target = ByteBuffer.allocate(len);
for (int j = 0; j < len; j++) {
target.put(buffer.get());
}
ByteBufferUtil.debugAll(target);
}
}
buffer.compact();
}
ByteBuffer扩容
当接收到的消息大于ByteBuffer的指定容量时，我们需要对ByteBuffer进行扩容。
示例代码
public void splitMessage(ByteBuffer buffer) {
buffer.flip();
for (int i = 0; i < buffer.limit(); i++) {
// 找到一条信息的结尾
if (((char) buffer.get(i)) == '\n') {
// 存储到长度为一条信息的结尾索引加一减去buffer起始索引的ByteBuffer中
int len = i + 1 - buffer.position();
ByteBuffer target = ByteBuffer.allocate(len);
for (int j = 0; j < len; j++) {
target.put(buffer.get());
}
ByteBufferUtil.debugAll(target);
}
}
buffer.compact();
}
@Test
public void clientSendMessageTest() throws IOException {
SocketChannel channel = SocketChannel.open();
InetSocketAddress localhost = new InetSocketAddress("localhost", 8080);
channel.connect(localhost);
channel.write(Charset.defaultCharset().encode("0123456789abcdef3333\n"));
System.in.read();
}
@SneakyThrows
@Test
public void selectorTest() {
// 创建selector 可以管理多个channel
Selector selector = Selector.open();
ServerSocketChannel serverSocketChannel = ServerSocketChannel.open();
// 设置非阻塞模式，默认为true,如果设置为true，则accept方法会一直阻塞，直到有客户端连接
serverSocketChannel.configureBlocking(false);
serverSocketChannel.bind(new InetSocketAddress(8080));
// 将channel注册到selector上，并指定事件为accept
// serverKey就是将来事件发生后，通过它可以找到事件和哪个channel发生的事件
SelectionKey serverKey = serverSocketChannel.register(selector, 0, null);
// 关注的是该channel的accept事件
serverKey.interestOps(SelectionKey.OP_ACCEPT);
log.debug("register key: {}", serverKey);
while (true) {
// select方法会一直阻塞，直到有事件发生
// select在事件未处理时是不会阻塞
selector.select();
// 获取所有发生的事件
Iterator<SelectionKey> keys = selector.selectedKeys().iterator();
while (keys.hasNext()) {
// key与serverKey一致
SelectionKey key = keys.next();
log.debug("key: {}", key);
// 取消关注事件
keys.remove();
// 处理事件
if (key.isAcceptable()) {
// 处理accept事件
ServerSocketChannel server = (ServerSocketChannel) key.channel();
log.debug("server socket: {}", server);
SocketChannel socketChannel = server.accept();
socketChannel.configureBlocking(false);
// ByteBuffer作为SocketChannel的附件
ByteBuffer buffer = ByteBuffer.allocate(16);
SelectionKey socketKey = socketChannel.register(selector, 0, buffer);
socketKey.interestOps(SelectionKey.OP_READ);
log.debug("{}", socketKey);
} else if (key.isReadable()) {
try {
// 处理read事件 获取触发事件的channel
SocketChannel socketChannel = (SocketChannel) key.channel();
// 获取触发事件的channel的附件
ByteBuffer buffer = (ByteBuffer) key.attachment();
int read = socketChannel.read(buffer);
if (read == -1) {
key.cancel();
}else {
splitMessage(buffer);
// 当position等于limit时，说明buffer已经读完，需要扩容
if (buffer.position() == buffer.limit()) {
ByteBuffer newBuffer = ByteBuffer.allocate(buffer.capacity() * 2);
// 由于splitMessage()方法已经将buffer
// 修改为写模式所以需要切换为读模式
buffer.flip();
newBuffer.put(buffer);
key.attach(newBuffer);
}
}
} catch (Exception e) {
e.printStackTrace();
// 由于客户端强制断开所以需要key取消，从selector的key集合中移除
key.cancel();
}
}
}
}
}
ByteBuffer大小分配
每个channel都需要记录可能被切分的消息，因为ByteBuffer不是线程安全的，ByteBuffer不能被多个channel使用，因为需要每个channel维护一个独立的ByteBuffer
ByteBuffer不能太大，比如一个ByteBuffer 1Mb的话，要支持百万连接就要1Tb内存，因此需要设计大小可变的ByteBuffer。设计思路如下:
1、先分配一个较小的buffer，如4Kb,发现数据不够，就在分配8Kb的buffer，将4Kb的buffer内容拷贝到8Kb的buffer中。优点就是消息连续容易处理，缺点就是数据拷贝耗费性能
2、用多个数组组成buffer，如果数组不够，把多出来的内容写入新的数组中，与1的区别就是消息不连续解析复杂，优点是避免了拷贝性能损耗。
服务器写大量数据的问题
当服务器向客户端写入大量的数据的时候，由于ByteBuffer的限制导致服务器不能一次性写入大量的数据，会出现当ByteBuffer不可写入的时候服务器会出现阻塞状态。这就导致CPU的资源没有得到充分的利用。解决方法：当ByteBuffer不可写入的时候，那么服务器可以采取读操作，当ByteBuffer可写入的时候就写入数据，从而大大提高CPU资源利用率。
示例代码
@Test
public void writeTest() throws IOException {
Selector selector = Selector.open();
ServerSocketChannel serverSocketChannel = ServerSocketChannel.open();
serverSocketChannel.configureBlocking(false);
serverSocketChannel.bind(new InetSocketAddress(8080));
serverSocketChannel.register(selector, SelectionKey.OP_ACCEPT, null);
while (true) {
selector.select();
Iterator<SelectionKey> keys = selector.selectedKeys().iterator();
while (keys.hasNext()) {
SelectionKey key = keys.next();
keys.remove();
if (key.isAcceptable()) {
SocketChannel socketChannel = serverSocketChannel.accept();
socketChannel.configureBlocking(false);
SelectionKey selectionKey = socketChannel.register(selector, SelectionKey.OP_READ, null);
// 向客户端发送大量数据
StringBuilder sb = new StringBuilder();
for (int i = 0; i < 300000000; i++) {
sb.append("h");
}
ByteBuffer buffer = Charset.defaultCharset().encode(sb.toString());
socketChannel.write(buffer);
// 判断buffer是否有剩余内容
if (buffer.hasRemaining()) {
// 关注可写事件和保留原事件
selectionKey.interestOps(selectionKey.interestOps() + SelectionKey.OP_WRITE);
// 将剩余的buffer的数据作为key的附件
selectionKey.attach(buffer);
}
} else if (key.isWritable()) {
ByteBuffer buffer = (ByteBuffer) key.attachment();
SocketChannel socketChannel = (SocketChannel) key.channel();
int write = socketChannel.write(buffer);
System.out.println(write);
// buffer没有可写内容就将key的附件置空
if (!buffer.hasRemaining()) {
key.attach(null);
// 不需要再关注可写事件
key.interestOps(key.interestOps() - SelectionKey.OP_WRITE);
}
}
}
}
}
多线程优化问题分析
由于之前的代码都是使用单线程来处理读写操作，没有充分多核CPU资源，所以需要使用多线程来进行优化。优化思路如下:
1、单线程配一个选择器（selector）,专门处理accept事件。
2、创建CPU核心数的线程，每个线程一个选择器，轮流处理read事件。
使用一个Boss线程专门处理select事件，负责将task建立连接给worker线程进行处理，worker线程专门处理读写操作。
创建BOSS线程并关联worker线程
@Test
public void bossThreadedTest() throws IOException {
// 当前线程为Boss
Thread.currentThread().setName("Boss");
// selector
Selector boss = Selector.open();
ServerSocketChannel serverSocketChannel = ServerSocketChannel.open();
serverSocketChannel.configureBlocking(false);
serverSocketChannel.bind(new InetSocketAddress(8080));
// 将channel注册到selector中, 并关注accept事件
serverSocketChannel.register(boss, SelectionKey.OP_ACCEPT, null);
// 创建指定数量的worker
Worker worker = new Worker("worker-01");
while (true) {
boss.select();
Iterator<SelectionKey> keys = boss.selectedKeys().iterator();
while (keys.hasNext()) {
SelectionKey key = keys.next();
keys.remove();
if (key.isAcceptable()) {
SocketChannel socketChannel = serverSocketChannel.accept();
socketChannel.configureBlocking(false);
log.debug("connection....{}", socketChannel.getRemoteAddress());
log.debug("before register....{}", socketChannel.getRemoteAddress());
// 初始化worker的selector，并启动worker-01线程
worker.register(socketChannel);
log.debug("after register....");
}
}
}
}
创建worker线程
/***
* 专门检测读写事件
* @author Tang
* @date 2023/12/15 22:49:28
* */
@NoArgsConstructor
static class Worker implements Runnable {
private Thread worker;
private Selector selector;
private String workerName;
private volatile boolean isStarter = false;
private ConcurrentLinkedQueue<Runnable> queue;
public Worker(String workerName) {
this.workerName = workerName;
this.queue = new ConcurrentLinkedQueue<>();
}
/***
* 初始化线程和selector
* @return void
* @author Tang
* @date 2023/12/15 22:47:32
*/
public void register(final SocketChannel socketChannel) throws IOException {
if (!isStarter) {
this.worker = new Thread(this, this.workerName);
this.worker.start();
selector = Selector.open();
isStarter = true;
}
// 由于channel的注册任务是在boss线程中，所以将
// 任务放进队列中，然后需要再worker线程执行该任务
queue.add(() -> {
// 将channel和worker进行关联
try {
socketChannel.register(selector, SelectionKey.OP_READ, null);
} catch (ClosedChannelException e) {
e.printStackTrace();
}
});
// 主动唤醒worker线程的select方法
selector.wakeup();
}
@SneakyThrows
@Override
public void run() {
while (true) {
selector.select();
// 执行队列中的任务
if (!queue.isEmpty()) {
queue.poll().run();
}
Iterator<SelectionKey> events = selector.selectedKeys().iterator();
while (events.hasNext()) {
SelectionKey event = events.next();
events.remove();
if (event.isReadable()) {
try {
SocketChannel socketChannel = (SocketChannel) event.channel();
ByteBuffer buffer = ByteBuffer.allocate(16);
log.debug("read.....");
int read = socketChannel.read(buffer);
if (read == -1) {
event.cancel();
} else {
buffer.flip();
ByteBufferUtil.debugAll(buffer);
}
} catch (Exception e) {
e.printStackTrace();
event.cancel();
}
}
}
}
}
}
由于boss线程和worker线程公用一个selector，当worker线程的selector先执行从而导致没有绑定到事件，这个时候selector.select()方法就会一直在worker线程阻塞，导致boss线程无法使用对应的selector，这就导致服务器和客户端都无法进行读写操作。所以就必须保证boss线程的selector绑定到事件后再将selector交给worker线程进行处理。
由于是多线程环境对selector进行操作，selector分别在boss线程和worker线程进行绑定事件以及等待事件的发生这两个代码逻辑先后循序无法保障，所以boss线程的selector绑定事件逻辑就交给worker线程来进行处理（但其中的SocketChannel必须是boss线程的SocketChannel）。
获取CPU个数
Runtime.getRuntime().availableProcessors()，如果工作在docker容器下，因为容器不是物理隔离。会拿到物理的CPU个数，而不是容器申请时的个数。
jdk10修复了这个问题，使用jvm参数UseContainerSupport配置，默认开启
使用多个Worker线程
修改代码如下：
@Test
public void bossThreadedTest() throws IOException {
// 与之前代码一致
.....
// 多个worker，如果是多核CPU,将worker线程数设置为CPU核心数
Worker[] workers = new Worker[2];
for (int i = 0; i < workers.length; i++) {
workers[i] = new Worker("worker" + i);
}
// 使用计数器来实现轮询效果
AtomicInteger index = new AtomicInteger();
while (true) {
// 与之前代码一致
.....
// 轮询初始化worker的selector，并启动worker线程
workers[index.getAndIncrement() % workers.length].register(socketChannel);
}
}
零拷贝
所谓的零拷贝并不是真正意义上的无拷贝，而是不会拷贝重复数据到jvm内存中。
优点：
1、更少的用户态和内核态的切换
2、不利用CPU计算，减少CPU缓存伪共享
3、零拷贝适合小文件传输
传统io问题
传统的io将一个文件通过socket写出
File file = new File("/test/data.txt");
RandomAccessFile accessFile = new RandomAccessFile(file, "r");
byte[] buf = new byte[(int)file.length()];
accessFile.read(buf);
Socket socket = ...;
socket.getOutputStream().write(buf);
内部工作流程是这样子的：
用户态与内核态的切换发生了3次，数据拷贝发生了4次。
NIO优化
ByteBuffer.allocate(10) HeapByteBuffer使用的还是java内存，ByteBuffer.allocateDirect(10)使用的是操作系统内存。
java使用DirectByteBuffer将堆外内存映射到jvm内存中来直接访问使用，相当于内核缓冲区和用户缓冲区进行合并，从而变相的减少一次数据拷贝过程。
进一步优化
java中对应着两个channel调用transferTo/transferFrom方法（底层使用的是linux2.1提供的sendFile方法）拷贝数据
java调用transferTo方法后，要从java程序的用户态切换到内核态，使用DMA将数据读入内核缓冲区中（该过程不会使用cpu）,数据从内核缓冲区传输到socket缓冲区（cpu会参与拷贝），最后使用DMA将socket缓冲区的数据写入网卡（该过程不会使用cpu）。
整个过程只发生了一次用户态与内核态的切换，数据拷贝了3次
DMA
DMA，全称Direct Memory Access，即直接存储器访问。
DMA传输将数据从一个地址空间复制到另一个地址空间，提供在外设和存储器之间或者存储器和存储器之间的高速数据传输。当CPU初始化这个传输动作，传输动作本身是由DMA控制器来实现和完成的。DMA传输方式无需CPU直接控制传输，也没有中断处理方式那样保留现场和恢复现场过程，通过硬件为RAM和IO设备开辟一条直接传输数据的通道，使得CPU的效率大大提高。
再进一步优化（需要linux2.4提供的api）
java调用transferTo方法后，要从java程序的用户态切换到内核态，使用DMA将数据读入内核缓冲区中（该过程不会使用cpu），只会将一些
offset和length信息写入到socket缓冲区中（过程几乎无消耗）。使用DMA将内核缓冲区的数据写入到网卡中（不会使用CPU）。这个过程只发生了一次内核态和用户态的切换，数据拷贝了2次。
本文来自互联网用户投稿，该文观点仅代表作者本人，不代表本站立场。本站仅提供信息存储空间服务，不拥有所有权，不承担相关法律责任。 如若内容造成侵权/违法违规/事实不符，请联系我的编程经验分享网邮箱：chenni525@qq.com进行投诉反馈，一经查实，立即删除！