Netty初体验-1-NIO基础补漏
文章目录
- 学习netty之前先学习NIO基础
- NIO基础
- 1.三大组件（Channel，Buffer，Selector）
- 2. ByteBuffer
- 3.文件编程
- 4.网络编程
- 网络编程小结：
- 非阻塞 VS 阻塞
- 1.1 阻塞
- 1.2 非阻塞
- 1.3 多路复用
- selector何时不阻塞
- 监听channel事件
- 更进一步优化
- 5. NIO vs BIO
- 5.1 stream VS channel
- 5.2 IO模型(下面的几个IO概念可能不清晰，有个大概，如果要详细看可以百度认真了解IO概念)
- 5.3零拷贝（针对java里面不用拷贝）
- 传统IO问题
- NIO优化
- 进一步优化
- 进一步优化（linux2.4）
- 5.4 AIO
- 文件AIO
- 网络AIO
学习netty之前先学习NIO基础
NIO基础
1.三大组件（Channel，Buffer，Selector）
- 1.1 Channel （读写双向管道）和 Buffer（可以将Channel的数据读入Buffer，也可将buffer的数据写入channel）
- 常见的Channel有
- 1.FileChannel （使用）
- 2.DatagramChannel (使用UDP网络时候)
- 3.SocketChannel (使用TCP的时候，服务器和客户端都可以用)
- 4.ServerSocketChannel （使用TCP的时候，专用于服务器的时候）
- 常见的Buffer缓冲区
- ByteBuffer（常用）
- MappedByteBuffer
- DirectByteBuffer
- HeapByteBuffer
- ShortBuffer
- IntBuffer
- LongBuffer
- FloatBuffer
- DoubleBuffer
- CharBuffer
- ByteBuffer（常用）
- 常见的Channel有
- 1.2 Selector
- 多线程版本设计，一个客户端一个socket线程，多了的话，可能造成内存溢出。缺点内存占用高；线程上下文切换成本高；只适合连接数少的场景。
- 线程池版本的设计，缺点：阻塞模式下，线程仅能处理一个socket连接；仅适合短链接的场景。（适合http请求）
- Selector版本的设计。selector的作用就是配合一个线程来管理多个channel，获取这些channel上发生的事件，这些channel工作在非阻塞模式下，不会让线程吊死在一个channel上，适合连接数多，流量低的场景（low traffic）；调用selector的select()方法会阻塞直到发生了读写就绪事件，一旦这些事件发生，selector方法就会返回这些事件交给thread处理。
2. ByteBuffer
-
- ByteBuffer案例：
package com.dapeng.netty;
import lombok.extern.slf4j.Slf4j;
import java.io.FileInputStream;
import java.io.IOException;
import java.nio.ByteBuffer;
import java.nio.channels.FileChannel;
@Slf4j
public class TestByteBuffer {
public static void main(String[] args) {
// FileChannel
// 1.输入输出流 2.RandomAccessFIle
try (FileChannel channel = new FileInputStream("data.txt").getChannel()) {
// 准备缓冲区
ByteBuffer buffer = ByteBuffer.allocate(10);// 初始化容量为10byte
while (true){
// 从channel读取，向缓冲区buffer写入
int len = channel.read(buffer);// 当返回-1表示读取完成
log.debug("读取到的字节数为：{}",len);
if (len == -1){
break;
}
// 打印buffer内容
buffer.flip();// 切换读模式
while (buffer.hasRemaining()){
byte b = buffer.get();// get()默认读取一个
log.debug("读取到的实际字节：{}",(char)b);
}
// 再切入到写模式，上面可能要写入
buffer.clear(); // 切换到写
}
} catch (IOException e) {
}
}
}
- 2.ByteBuffer正确使用姿势
- 2.1向buffer写入数据的时候，例如调用channel.read(buffer)。
- 2.2 调用buffer.filp(); 切换为读模式
- 2.3 从buffer读取数据，例如调用buffer.get()；获取单个字节
- 2.4 调用buffer.clear(),或者compact() 切换至 写模式
- 2.5 重复1-4步骤。
- 3.ByteBuffer结构
- 3.1ByteBuffer有一下几个重要属性：
- Capacity（容量）
- Position（当前位置）
- LImit（写入限制）
如下图所示：
- 3.2 读写切换
- 1.刚开始为写模式：
- 2 切换读的时候，调用filp（）
- 3.读取完后，如图：
-
- 切换回来为写模式，调用clear()方法。
- 切换回来为写模式，调用clear()方法。
- 5.特殊的compact的方法，在读取一部分或者因为默写情况未读完，则需要在未读的数据后面追加写入。从未读完的那里开始，重新写入。
读写实例：
(以下所有代码需要用到该类)首先拿到工具类：
- 1.刚开始为写模式：
- 3.1ByteBuffer有一下几个重要属性：
package com.dapeng.netty.utils;
import java.nio.ByteBuffer;
import static io.netty.util.internal.StringUtil.NEWLINE;
import static io.netty.util.internal.MathUtil.isOutOfBounds;
import io.netty.util.internal.StringUtil;
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(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 (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(
" +-------------------------------------------------+" +
NEWLINE + " | 0 1 2 3 4 5 6 7 8 9 a b c d e f |" +
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(NEWLINE +
"+--------+-------------------------------------------------+----------------+");
}
private static void appendHexDumpRowPrefix(StringBuilder dump, int row, int rowStartIndex) {
if (row < HEXDUMP_ROWPREFIXES.length) {
dump.append(HEXDUMP_ROWPREFIXES[row]);
} else {
dump.append(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);
}
}
- 案例1
package com.dapeng.netty;
import java.nio.ByteBuffer;
import static com.dapeng.netty.utils.ByteBufferUtil.debugAll;
public class TestByteBufferReadWrite {
public static void main(String[] args) {
ByteBuffer buffer = ByteBuffer.allocate(10);
buffer.put((byte) 0x61); // 'a'
debugAll(buffer); // 打印当前buffer里面的内容。
buffer.put(new byte[]{ 0x62, 0x63, 0x64});
debugAll(buffer);
// 切换读模式
buffer.flip();
System.out.println("读模式" + buffer.get());
debugAll(buffer);
// 切换到compact模式
buffer.compact();
debugAll(buffer);// 切换到compact后，会往前追加未读的数据，并且会残留最后一位数据，下次写的时候直接覆盖。
buffer.put(new byte[]{ 0x65, 0x6f});// 继续写
debugAll(buffer);
}
}
- 4.ByteBuffer常见的方法
- 4.1 Allocate（）
package com.dapeng.netty;
import java.nio.ByteBuffer;
public class TestByteBufferAllocate {
public static void main(String[] args) {
System.out.println(ByteBuffer.allocate(16).getClass());
System.out.println(ByteBuffer.allocateDirect(16).getClass());
// class java.nio.HeapByteBuffer - java 堆内存，读写效率较低，受到GC影响
// class java.nio.DirectByteBuffer - 直接内存，读写效率高（少一次拷贝），不会收到GC影响，分配效率低，使用过不当会造成内存泄露。
}
}
- 4.2 从buffer写入数据
- 调用channel的read方法。
- 调用buffer的put方法。
int readBytes = channel.read(buf);//返回读取的字节数
buf.put((byte)127);
- 4.3 从buffer读取数据
- 调用channel的write方法
- 调用channel的get方法
int writeBytes = channel.write(buf);// 返回读取到的字节的ascII码。
buf.get();//获取单个字节
- 4.4 读取案例：(没有工具在上面有一个工具找一下)
package com.dapeng.netty;
import java.nio.ByteBuffer;
import static com.dapeng.netty.utils.ByteBufferUtil.debugAll;
/**
* 演示buffer的读取
*/
public class TestBufferRead {
public static void main(String[] args) {
ByteBuffer buf = ByteBuffer.allocate(10);
buf.put(new byte[]{'a','b','c','d'});
buf.flip();
// buf.get(new byte[4]);
// debugAll(buf); // 读取4个长度
// buf.rewind();// 从头开始读
// System.out.println((char)buf.get());// 获取第一个
// debugAll(buf);
// System.out.println((char)buf.get());// 获取'a'
// System.out.println((char)buf.get());// 获取'b'
// buf.mark();// 在下标为2的地方加标记
// System.out.println((char)buf.get());//获取'c'
// System.out.println((char)buf.get());// 获取'd'
// buf.reset();
// System.out.println((char)buf.get());// 获取下标为2的，'c'
// System.out.println((char)buf.get());// 获取'd'
System.out.println((char) buf.get(2));// 获取'c' 通过get并不会改索引的位置
debugAll(buf);
}
}
- 4.5 String和ByteBuffer互相转换
需要注意，使用allocate定义的bytebuffer在从字符串转byte的时候需要把buffer切换为读模式。（本次案例中先使用写模式，再切换为读模式并输出字符串）
package com.dapeng.netty;
import java.nio.ByteBuffer;
import java.nio.charset.StandardCharsets;
import static com.dapeng.netty.utils.ByteBufferUtil.debugAll;
public class TestByteBufferStringToBuffer {
public static void main(String[] args) {
// 1.字符串转buffer
ByteBuffer buffer = ByteBuffer.allocate(16);
buffer.put("hello".getBytes());
debugAll(buffer);
// 2.CharSet
ByteBuffer buffer2 = StandardCharsets.UTF_8.encode("hello");
debugAll(buffer2);
// 3.wrap
ByteBuffer buffer3 = ByteBuffer.wrap("hello".getBytes());
debugAll(buffer3);
// buffer转字符串
// 2和3已经切换为读模式了，所以可以直接decode,像第一种方式，需要进行读模式的切换，才能够正确decode
String s = StandardCharsets.UTF_8.decode(buffer2).toString();
System.out.println(s);
buffer.flip();// 切换读
String s2 = StandardCharsets.UTF_8.decode(buffer).toString();
System.out.println(s2);
}
}
- 4.6 ScatteringRead分散读（读取指定字节数）
-----不考虑分散读，需要先进行，完整的读取文件内容，再对单词进行分割，再把每个分割的放到一个ByteBuffer。
-----分散读目的为了少拷贝，提升效率。这是一种思想
案例：
package com.dapeng.netty;
import java.io.IOException;
import java.io.RandomAccessFile;
import java.nio.ByteBuffer;
import java.nio.channels.FileChannel;
import static com.dapeng.netty.utils.ByteBufferUtil.debugAll;
public class TestSactteringRead {
public static void main(String[] args) {
// 使用RandomAccessFile读取（之前使用InputStream）
try (FileChannel channel = new RandomAccessFile("words.txt","r").getChannel()) {
// 来三个ByteBuffer，分散读取已知的长度
ByteBuffer b1 = ByteBuffer.allocate(3);
ByteBuffer b2 = ByteBuffer.allocate(3);
ByteBuffer b3 = ByteBuffer.allocate(5);
channel.read(new ByteBuffer[]{b1, b2, b3});// 写入到ByteBuffer里面
// 切换ByteBuffer写模式到读模式
b1.flip();
b2.flip();
b3.flip();
debugAll(b1);
debugAll(b2);
debugAll(b3);
} catch (IOException e) {
}
}
}
- 4.7 集中写GatheringWrite
package com.dapeng.netty;
import java.io.IOException;
import java.io.RandomAccessFile;
import java.nio.ByteBuffer;
import java.nio.channels.FileChannel;
import java.nio.charset.StandardCharsets;
public class TestGatheringWrite {
public static void main(String[] args) {
ByteBuffer b1 = StandardCharsets.UTF_8.encode("hello");
ByteBuffer b2 = StandardCharsets.UTF_8.encode("world");
ByteBuffer b3 = StandardCharsets.UTF_8.encode("你好");
// 先来一个channel
try (FileChannel channel = new RandomAccessFile("write2.txt","rw").getChannel()) {
// 集中写入一个ByteBuffer
channel.write(new ByteBuffer[]{b1, b2, b3});
} catch (IOException e) {
}
}
}
- 4.8黏包，半包处理。
- 黏包：我理解的是客户端发送数据时，将含有\n或者别的字符一起发送，导致连在一起。
- 半包：当缓冲区ByteBuffer设置不足，会导致缓冲区满了，然后等下次接收导致接受的数据会断开。
案例
package com.dapeng.netty;
import io.netty.buffer.ByteBuf;
import java.nio.ByteBuffer;
import static com.dapeng.netty.utils.ByteBufferUtil.debugAll;
public class TestBufferExam {
public static void main(String[] args) {
/*
网络上有多条数据发送给服务器，数据之间使用 \n 进行分割
但由于某种原因，这些数据在接收的时候，被重新组合，例如三条数据
Hello,world\n
I'm Zhangsan\n
How are you?\n
变成了下面：
Hello,world\nI'm Zhangsan\nHo
w are you?\n
现在要求编写程序，将错乱的数据恢复成原样，按 \n 分割的数据。
*/
// 模拟接收到的数据并处理
ByteBuffer source = ByteBuffer.allocate(32);
source.put("Hello,world\nI'm Zhangsan\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') { // 字符'\n'
// 如果遇到了\n，则说明是一条完整的信息
// 重新新建一个ByteBuffer来接收
int length = i + 1 - source.position();
ByteBuffer target = ByteBuffer.allocate(length);
// 从source读，像target写
for (int j = 0; j < length; j++) {
target.put(source.get());
}
debugAll(target);
}
}
// 处理完以后切换到写模式
source.compact();
}
}
3.文件编程
注意
FileChannel只能工作在阻塞模式下
- 3.1获取FileChannel
在RandomAccessFIle里面获取的时候需要指定模式，第二个参数为读取模式，读写模式：‘rw’ - 3.2 读取：
会从Channel读取数据填充ByteBuffer，返回值代表读到了多少字节，-1表示到了文件的末尾。
int readByte = channel.read(buffer);
- 3.3 写入
正确姿势：
ByteBuffer buffer = ...;
buffer.put("...");// 给buffer填充内容
buffer.flip();// 切换读
whie(buffer.hasRemaining){
channel.write(buffer);
//在while中调用write，因为channel一次write可能写不完，所以先判断buffer是否还有剩余，后续的socketchannel也用此模式。
}
- 3.4 当前位置
long pos = channel.position();
// 设置当前位置：
long newPos = ...;
channel.position(newPos);
设置当前位置时，如果设置在文件的末尾
- 这时候读取就会返回-1，
- 这时候写入，会追加内容，注意如果position超过了文件的末尾，再写入新的内容和原来的内容会有空洞（00）
- 3.5 获取当前文件大小
channel.size();// 获取文件大小
- 3.6 强制写入
操作系统出于性能的考虑，会将FileChannel的数据存到系统的缓存当中，不是立刻写入磁盘，FileChannel可以调用force(true)，方法来将文件内容和元数据（文件的权限信息等）立刻写入磁盘。 - 3.7 TransferTo方法，将一个文件的内容复制到另一个文件里面
此方法只能传输2g内容，如果超过2g，利用循环重复传输
package com.dapeng.netty;
import java.io.FileInputStream;
import java.io.IOException;
import java.nio.channels.FileChannel;
public class TestTansferTo {
public static void main(String[] args) {
// 定义两个channel
try (
FileChannel from = new FileInputStream("data.txt").getChannel();
FileChannel to = new FileOutputStream("to.txt").getChannel();
;
) {
// 使用TransferTo方法
// 效率高，底层使用操作系统的零拷贝方法。
//--改进
long size = from.size();
// left代表剩余的大小
for(long left = size; size > 0;){
// size-left 代表，从此位置开始，传输left大小的长度，到to的channel里面。
// transferto返回转移多少大小
left -= from.transferTo((size - left),left,to);
}
} catch (IOException e) {
e.printStackTrace();
}
}
}
- 3.8 Path 和 Paths
- 3.9 Files
如果目录中含有内容，会抛出异常DirectoryNotEmptyException - 3.10 遍历目录Files.walkFileTree(Paths.get(“…”),new SimpleFileVisitor(){})
package com.dapeng.netty;
import java.io.IOException;
import java.nio.file.*;
import java.nio.file.attribute.BasicFileAttributes;
import java.util.concurrent.atomic.AtomicInteger;
public class TestWalkFile {
public static void main(String[] args) throws IOException {
// 定义两个计数器：
AtomicInteger dirCount = new AtomicInteger();
AtomicInteger filCount = new AtomicInteger();
Files.walkFileTree(Paths.get("...../jdk-1.8.jdk"),new SimpleFileVisitor<Path>(){
@Override
// 此方法是便利目录前看看
public FileVisitResult preVisitDirectory(Path dir, BasicFileAttributes attrs) throws IOException {
System.out.println("====>当前目录为：" + dir);
dirCount.incrementAndGet(); // 加1
return super.preVisitDirectory(dir, attrs);
}
@Override
// 遍历到文件后怎么做
public FileVisitResult visitFile(Path file, BasicFileAttributes attrs) throws IOException {
System.out.println("当前文件为：" + file);
filCount.incrementAndGet();// +1
return super.visitFile(file, attrs);
}
});
System.out.println("目录总数为：" + dirCount);
System.out.println("文件总数为：" + filCount);
}
}
//========
如果查看jar包，如下
AtomicInteger jarCount = new AtomicInteger();
Files.walkFileTree(Paths.get("/Library/Java/JavaVirtualMachines/jdk-1.8.jdk"),new SimpleFileVisitor<Path>(){
@Override
public FileVisitResult visitFile(Path file, BasicFileAttributes attrs) throws IOException {
if (file.toString().endsWith(".jar")) {
System.out.println("jar file ===>" + file);
jarCount.incrementAndGet();
}
return super.visitFile(file, attrs);
}
});
System.out.println("jarCount==>" + jarCount);
- 3.11 文件目录的删除
==注意删除不可逆，路径名一定谨慎填写 ==
package com.dapeng.netty;
import java.io.IOException;
import java.nio.file.*;
import java.nio.file.attribute.BasicFileAttributes;
public class TestWalkFileTreeDelete {
public static void main(String[] args) throws IOException {
Files.walkFileTree(Paths.get("路径名"),new SimpleFileVisitor<Path>(){
@Override
public FileVisitResult visitFile(Path file, BasicFileAttributes attrs) throws IOException {
// 在访问文件的时候，执行删除文件操作
Files.delete(file);
return super.visitFile(file, attrs);
}
@Override
public FileVisitResult postVisitDirectory(Path dir, IOException exc) throws IOException {
// 在退出目录的时候，执行删除目录操作
Files.delete(dir);
return super.postVisitDirectory(dir, exc);
}
});
}
}
- 3-12 文件的复制(从一个文件夹复制到另一个文件夹，所有内容)
package com.dapeng.netty;
import java.io.IOException;
import java.nio.file.Files;
import java.nio.file.Path;
import java.nio.file.Paths;
public class TestFileCopy {
public static void main(String[] args) throws IOException {
String source = "路径1";// 已存在
String target = "路径2";// 不存在
Files.walk(Paths.get(source)).forEach(path -> {
// 替换掉路径1中目录的前缀的路径
try {
String targetName = path.toString().replace(source,target);
if (Files.isDirectory(path)){
// 如果是目录，新建目录
Files.createDirectory(Paths.get(targetName));
}else if (Files.isRegularFile(path)){
// 如果是文件，则复制
Files.copy(path,Paths.get(targetName));
}
} catch (IOException e) {
e.printStackTrace();
}
});
}
}
4.网络编程
- 1.阻塞和非阻塞
- 2.阻塞案例：
服务端
package com.dapeng.netty.c4;
import lombok.extern.slf4j.Slf4j;
import java.io.IOException;
import java.net.InetSocketAddress;
import java.nio.ByteBuffer;
import java.nio.channels.ServerSocketChannel;
import java.nio.channels.SocketChannel;
import java.util.ArrayList;
import java.util.List;
import static com.dapeng.netty.utils.ByteBufferUtil.debugRead;
@Slf4j
public class Server {
public static void main(String[] args) throws IOException {
// 使用nio的阻塞模式
// 0.创建ByteBuffer
ByteBuffer buffer = ByteBuffer.allocate(16);
// 1.创建服务器
ServerSocketChannel ssc = ServerSocketChannel.open();
// 2.绑定端口
ssc.bind(new InetSocketAddress(8080));
// 3.连接集合
List<SocketChannel> channels = new ArrayList<>();
while (true){
// 4.accept 建立与客户端的连接，SocketChannel用于与客户端之间通信
log.debug("连接中.... connecting...");
SocketChannel sc =ssc.accept();// 阻塞方法，线程停止运行
log.debug("连接成功 connected {}",sc);
channels.add(sc);
for (SocketChannel channel : channels) {
// 5.接收客户端发送的消息
log.debug("before read... 读取数据之前... {}",channel);
channel.read(buffer); // 阻塞方法，线程停止运行
// 6.切换为读模式
buffer.flip();
debugRead(buffer);
//7.切换为写模式
buffer.clear();
log.debug("after read... 读取数据后...{}",channel);
}
}
}
}
客户端
通过Debug模式，然后调用sc.write(Charset.defaultCharset.encode(str));来向服务端发送消息。
package com.dapeng.netty.c4;
import java.io.IOException;
import java.net.InetSocketAddress;
import java.nio.channels.SocketChannel;
public class Client {
public static void main(String[] args) throws IOException {
SocketChannel sc =SocketChannel.open();
sc.connect(new InetSocketAddress("localhost",8080));
System.out.println("waiting ......");
}
}
小结
阻塞模式下，某个方法的执行，都会影响另一个方法的执行，比如上面案例的ssc.accept()会阻断后面代码的运行，一直等待连接，channel.read(buffer)，也会阻断线程，让线程停止运行，等待客户端发送数据。
- 非阻塞案例，使用单线程（在阻塞案例上面做了一点修改）
package com.dapeng.netty.c4;
import lombok.extern.slf4j.Slf4j;
import java.io.IOException;
import java.net.InetSocketAddress;
import java.nio.ByteBuffer;
import java.nio.channels.ServerSocketChannel;
import java.nio.channels.SocketChannel;
import java.util.ArrayList;
import java.util.List;
import static com.dapeng.netty.utils.ByteBufferUtil.debugRead;
@Slf4j
public class Server {
public static void main(String[] args) throws IOException {
// 使用nio的阻塞模式
// 0.创建ByteBuffer
ByteBuffer buffer = ByteBuffer.allocate(16);
// 1.创建服务器
ServerSocketChannel ssc = ServerSocketChannel.open();
// 设置非阻塞模式--------修改的地方
ssc.configureBlocking(false);// 默认阻塞模式，false代表非阻塞。
// 2.绑定端口
ssc.bind(new InetSocketAddress(8080));
// 3.连接集合
List<SocketChannel> channels = new ArrayList<>();
while (true){
// 4.accept 建立与客户端的连接，SocketChannel用于与客户端之间通信
// log.debug("连接中.... connecting...");
SocketChannel sc = ssc.accept();// 阻塞方法，线程停止运行，非阻塞模式下如果没有连接则为空
//--------修改的地方
if (sc != null){
log.debug("连接成功 connected {}",sc);
// 这里SocketChannel也要设置为非阻塞模式，即连接的客户端
sc.configureBlocking(false);// socketChannel设置为非阻塞模式下，read默认0
channels.add(sc);
}
for (SocketChannel channel : channels) {
// 5.接收客户端发送的消息
// log.debug("before read... 读取数据之前... {}",channel);
int read = channel.read(buffer);// 阻塞方法，线程停止运行,非阻塞模式下，默认0
if (read > 0){
// 6.切换为读模式
buffer.flip();
debugRead(buffer);
//7.切换为写模式
buffer.clear();
log.debug("after read... 读取数据后...{}",channel);
}
}
}
}
}
-
- Selector
selector在事件未处理时，不会阻塞，会一直循环，要么处理，要么取消，例如下面代码
- Selector
package com.dapeng.netty.c4;
import lombok.extern.slf4j.Slf4j;
import java.io.IOException;
import java.net.InetSocketAddress;
import java.nio.channels.*;
import java.util.Iterator;
@Slf4j
public class ServerSelector {
public static void main(String[] args) throws IOException {
// 1.创建selector，管理多个channel
Selector selector = Selector.open(); // 通过静态方法调用
ServerSocketChannel ssc = ServerSocketChannel.open();// 创建serversocketchannel
ssc.configureBlocking(false);// 设置非阻塞
// 2.建立selector和selector的联系(注册),0代表不监听任何事件
SelectionKey sscKey = ssc.register(selector, 0, null);
// key只关注监听事件
sscKey.interestOps(SelectionKey.OP_ACCEPT);
log.debug("register key {}",sscKey);
ssc.bind(new InetSocketAddress(8080));
while (true) {
// 3.selector方法，没有事件发生的时候，阻塞，有事件的时候线程才会恢复。
// 如果select事件没有被处理，例如channel.accept()；会直接把未处理事件加入到selectorKeys的set集合里面，继续循环。直到处理
selector.select();//阻塞
// 4.处理事件
Iterator<SelectionKey> iterator = selector.selectedKeys().iterator();
if (iterator.hasNext()) {
SelectionKey key = iterator.next();
log.debug("循环的key： {}",key);
ServerSocketChannel channel = (ServerSocketChannel) key.channel();
SocketChannel sc = channel.accept();
log.debug("连接的 Channel {}",sc);
// 如果不处理，必须取消掉
// key.cancel();
}
}
}
}
-
- 为了防止死循环，下面做了if判断,还有就是selectionKeys和selectedKeys的区别，注册时候把所有的channel注册到selectionKeys中，当发生事件的时候，会把对象复制到selectedKeys里面，当事件发生完后，此时的selectedkey会被标记为已处理，但是还留在selectedKeys里面，需要手动删除。在迭代器中调用iterator.remove()方法。
package com.dapeng.netty.c4;
import lombok.extern.slf4j.Slf4j;
import java.io.IOException;
import java.net.InetSocketAddress;
import java.nio.ByteBuffer;
import java.nio.channels.*;
import java.util.Iterator;
import static com.dapeng.netty.utils.ByteBufferUtil.debugRead;
@Slf4j
public class ServerSelector {
public static void main(String[] args) throws IOException {
// 1.创建selector，管理多个channel
Selector selector = Selector.open(); // 通过静态方法调用
ServerSocketChannel ssc = ServerSocketChannel.open();// 创建serversocketchannel
ssc.configureBlocking(false);// 设置非阻塞
// 2.建立selector和selector的联系(注册),0代表不监听任何事件
SelectionKey sscKey = ssc.register(selector, 0, null);
// key只关注监听事件
sscKey.interestOps(SelectionKey.OP_ACCEPT);
log.debug("register key {}",sscKey);
ssc.bind(new InetSocketAddress(8080));
while (true) {
// 3.selector方法，没有事件发生的时候，阻塞，有事件的时候线程才会恢复。
// 如果select事件没有被处理，例如channel.accept()；会直接把未处理事件加入到selectorKeys的set集合里面，继续循环。直到处理
selector.select();//阻塞
// 4.处理事件
Iterator<SelectionKey> iterator = selector.selectedKeys().iterator();
if (iterator.hasNext()) {
SelectionKey key = iterator.next();
// 处理key时候，要从selectedkeys里面把已经处理过的key给删除，防止下次循环出现问题。
iterator.remove();
log.debug("循环的key： {}",key);
// 5.区分事件类型
if(key.isAcceptable()){// 如果是accept
ServerSocketChannel channel = (ServerSocketChannel) key.channel();
SocketChannel sc = channel.accept();// 如果建立了连接，就会被标记位处理过了，他的事件会被移除，但是key不会从selectedKey中移除。
sc.configureBlocking(false);// 设置非阻塞
SelectionKey scKey = sc.register(selector, 0, null);// 给客户端连接channel注册读事件
scKey.interestOps(SelectionKey.OP_READ);
log.debug("{}",sc);
log.debug("scKey {}",scKey);
}else if (key.isReadable()){
SocketChannel channel = (SocketChannel) key.channel();
ByteBuffer buffer = ByteBuffer.allocate(16);
channel.read(buffer);
buffer.flip();// 切换读
debugRead(buffer);
}
}
}
}
}
红色为所有注册到seleector的集合，绿色为发生事件时候把红色中的给复制过来，来发生事件，发生完了后会被标记，但不会删除，需要手动删除。
处理客户端的异常情况，和防止服务端终止
package com.dapeng.netty.c4;
import lombok.extern.slf4j.Slf4j;
import java.io.IOException;
import java.net.InetSocketAddress;
import java.nio.ByteBuffer;
import java.nio.channels.*;
import java.util.Iterator;
import static com.dapeng.netty.utils.ByteBufferUtil.debugRead;
@Slf4j
public class ServerSelector {
public static void main(String[] args) throws IOException {
// 1.创建selector，管理多个channel
Selector selector = Selector.open(); // 通过静态方法调用
ServerSocketChannel ssc = ServerSocketChannel.open();// 创建serversocketchannel
ssc.configureBlocking(false);// 设置非阻塞
// 2.建立selector和selector的联系(注册),0代表不监听任何事件
SelectionKey sscKey = ssc.register(selector, 0, null);
// key只关注监听事件
sscKey.interestOps(SelectionKey.OP_ACCEPT);
log.debug("register key {}",sscKey);
ssc.bind(new InetSocketAddress(8080));
while (true) {
// 3.selector方法，没有事件发生的时候，阻塞，有事件的时候线程才会恢复。
// 如果select事件没有被处理，例如channel.accept()；会直接把未处理事件加入到selectorKeys的set集合里面，继续循环。直到处理
selector.select();//阻塞
// 4.处理事件
Iterator<SelectionKey> iterator = selector.selectedKeys().iterator();
if (iterator.hasNext()) {
SelectionKey key = iterator.next();
// 处理key时候，要从selectedkeys里面把已经处理过的key给删除，防止下次循环出现问题。
iterator.remove();
log.debug("循环的key： {}",key);
// 5.区分事件类型
if(key.isAcceptable()){// 如果是accept
ServerSocketChannel channel = (ServerSocketChannel) key.channel();
SocketChannel sc = channel.accept();// 如果建立了连接，就会被标记位处理过了，他的事件会被移除，但是key不会从selectedKey中移除。
sc.configureBlocking(false);// 设置非阻塞
SelectionKey scKey = sc.register(selector, 0, null);// 给客户端连接channel注册读事件
scKey.interestOps(SelectionKey.OP_READ);
log.debug("{}",sc);
log.debug("scKey {}",scKey);
}else if (key.isReadable()){
try {
SocketChannel channel = (SocketChannel) key.channel();
ByteBuffer buffer = ByteBuffer.allocate(16);
int read = channel.read(buffer);// 如果正常断开，read方法返回-1；
if (read == -1){
key.cancel();
}else {
buffer.flip();// 切换读
debugRead(buffer);
}
} catch (IOException e) {
e.printStackTrace();
// 因为客户端断开了，因此需要将key从selector 的key的集合中真正删除掉key
key.cancel();
}
}
}
}
}
}
-
- 处理消息的边界问题（汉字占win10占3个字节，导致出现乱码）
-
5.1按照约定的规则来，但有可能会造成空间的浪费：（比如约定好都是1024个字节，然后第一个客户端发了满满的，第二个客户端只发了100个字节，后面的空白会造成带宽的浪费）
-
5.2 第二种方式，通过\n来分割，但有问题的是如果传输的内容长度，大于临时的ByteBuffer，会造成读不全，然后得考虑扩容问题。还有就是效率低下。
-
5.3 （Http 1.0 是TLV格式的，Http 2.0 是LTV格式，L：length，T：type，V：value）第三种方式，比较常用的，先读取一个整形，发现需要4byte，然后分配一个4字节长度的ByteBuffer，然后在读一个整形，发现需要8byte长度的ByteBuffer，然后按需分配就行。
-
5.4 使用附属品（attachment）来解决各个channel之间ByteBuffer的隔离。并使用扩容处理消息边界问题。
通过在处理读取事件的时候，注册selecetor上面的同时添加一个attachment，把ByteBuffer注册到上面，然后再到后面处理数据的时候，使用自己的split(Bytebuffer buffer )方法来打印处理，后面做判断buffer的position和limit相等的时候说明buffer满了，然后进行扩容，拷贝到新ByteBuffer上面。
- 处理消息的边界问题（汉字占win10占3个字节，导致出现乱码）
package com.dapeng.netty.c4;
import lombok.extern.slf4j.Slf4j;
import java.io.IOException;
import java.net.InetSocketAddress;
import java.nio.ByteBuffer;
import java.nio.channels.*;
import java.util.Iterator;
import static com.dapeng.netty.utils.ByteBufferUtil.debugAll;
import static com.dapeng.netty.utils.ByteBufferUtil.debugRead;
@Slf4j
public class ServerSelector {
private static void split(ByteBuffer source){
// 先切换到读模式
source.flip();
// 对数据进行处理
for (int i = 0; i < source.limit(); i++) {
// 找到一条完整的信息
if (source.get(i) == '\n') { // 字符'\n'
// 如果遇到了\n，则说明是一条完整的信息
// 重新新建一个ByteBuffer来接收
int length = i + 1 - source.position();
ByteBuffer target = ByteBuffer.allocate(length);
// 从source读，像target写
for (int j = 0; j < length; j++) {
target.put(source.get());
}
debugAll(target);
}
}
// 处理完以后切换到写模式，如果压缩后的position和limit相等，compact方法会将未读的数据往前压缩，然后令positoin等于未读的位置，比如16字节position = 16 ，limit = 16.
source.compact();
}
public static void main(String[] args) throws IOException {
// 1.创建selector，管理多个channel
Selector selector = Selector.open(); // 通过静态方法调用
ServerSocketChannel ssc = ServerSocketChannel.open();// 创建serversocketchannel
ssc.configureBlocking(false);// 设置非阻塞
// 2.建立selector和selector的联系(注册),0代表不监听任何事件
SelectionKey sscKey = ssc.register(selector, 0, null);
// key只关注监听事件
sscKey.interestOps(SelectionKey.OP_ACCEPT);
log.debug("register key {}",sscKey);
ssc.bind(new InetSocketAddress(8080));
while (true) {
// 3.selector方法，没有事件发生的时候，阻塞，有事件的时候线程才会恢复。
// 如果select事件没有被处理，例如channel.accept()；会直接把未处理事件加入到selectorKeys的set集合里面，继续循环。直到处理
selector.select();//阻塞
// 4.处理事件
Iterator<SelectionKey> iterator = selector.selectedKeys().iterator();
if (iterator.hasNext()) {
SelectionKey key = iterator.next();
// 处理key时候，要从selectedkeys里面把已经处理过的key给删除，防止下次循环出现问题。
iterator.remove();
log.debug("循环的key： {}",key);
// 5.区分事件类型
if(key.isAcceptable()){// 如果是accept
ServerSocketChannel channel = (ServerSocketChannel) key.channel();
SocketChannel sc = channel.accept();// 如果建立了连接，就会被标记位处理过了，他的事件会被移除，但是key不会从selectedKey中移除。
sc.configureBlocking(false);// 设置非阻塞
// 第三个参数为附件，添加attachment
ByteBuffer buffer = ByteBuffer.allocate(16);
SelectionKey scKey = sc.register(selector, 0, buffer);// 给客户端连接channel注册读事件，
// SelectionKey scKey = sc.register(selector, 0, null);// 给客户端连接channel注册读事件，
scKey.interestOps(SelectionKey.OP_READ);
log.debug("{}",sc);
log.debug("scKey {}",scKey);
}else if (key.isReadable()){
try {
SocketChannel channel = (SocketChannel) key.channel();
// ByteBuffer buffer = ByteBuffer.allocate(16);
// 1.把从注册到selector上面的channelkey的附件拿到
ByteBuffer buffer = (ByteBuffer) key.attachment();
int read = channel.read(buffer);// 如果正常断开，read方法返回-1；
if (read == -1){
key.cancel();
}else {
// buffer.flip();// 切换读
// debugRead(buffer);
split(buffer);
// 如果buffer的position和limit相等，则说明buffer满了，需要处理扩容
if (buffer.position() == buffer.limit()){
ByteBuffer newBuffer = ByteBuffer.allocate(buffer.capacity() * 2);
buffer.flip();// 切换读
newBuffer.put(buffer);
key.attach(newBuffer);// 把新的buffer作为附件传送上去。
}
}
} catch (IOException e) {
e.printStackTrace();
// 因为客户端断开了，因此需要将key从selector 的key的集合中真正删除掉key
key.cancel();
}
}
}
}
}
}
-
- ByteBuffer大小分配
- ByteBuffer大小分配
- 7.处理可写事件。
服务端
package com.dapeng.netty.c4;
import java.io.IOException;
import java.net.InetSocketAddress;
import java.nio.ByteBuffer;
import java.nio.channels.*;
import java.nio.charset.Charset;
import java.util.Iterator;
public class WriteServer {
public static void main(String[] args) throws IOException {
// 1.建立服务器channel
ServerSocketChannel ssc = ServerSocketChannel.open();
// 2. 设置非阻塞
ssc.configureBlocking(false);
// 3. 建立selector选择器，防止服务器循环轮询
Selector selector = Selector.open();
ssc.register(selector, SelectionKey.OP_ACCEPT); // 4.设置监听连接
// 5.绑定端口
ssc.bind(new InetSocketAddress(8080));
while (true) {
// 6.通过selector阻塞
selector.select();
// 7.迭代所有的selectorkeys
Iterator<SelectionKey> iterator = selector.selectedKeys().iterator();
while (iterator.hasNext()) {
SelectionKey key = iterator.next();
// 8.去除掉已经标记完的selectedKey
iterator.remove();
// 9.如果为可连接的key
if (key.isAcceptable()) {
SocketChannel sc = ssc.accept();
sc.configureBlocking(false);// 连接的客户端让其读写为非阻塞
SelectionKey sckey = sc.register(selector, 0, null);// 把客户端绑定到selector
sckey.interestOps(SelectionKey.OP_READ);// 绑定可读事件
// 向客户端发送大量数据
StringBuilder builder = new StringBuilder();
for (int i = 0; i < 30000000; i++) {
builder.append("a");
}
ByteBuffer buffer = Charset.defaultCharset().encode(builder.toString());
int write = sc.write(buffer);// 返回值代表写入的字节数
System.out.println(write);
if (buffer.hasRemaining()) {
//关注可读可写事件。
sckey.interestOps(sckey.interestOps() + SelectionKey.OP_WRITE);
// sckey.interestOps(sckey.interestOps() | SelectionKey.OP_WRITE;
// 采用附件方式，把未写完的buffer放到附件中
sckey.attach(buffer);
}
} else if (key.isWritable()) {
ByteBuffer buffer = (ByteBuffer) key.attachment();
SocketChannel channel = (SocketChannel) key.channel();
// 继续向客户端发送消息
int write = channel.write(buffer);
System.out.println(write);
// 清理操作，buffer写完后再去把buffer清理掉，防止占内存。还有所关注的可写事件
if ( !buffer.hasRemaining()){
key.attach(null);// 清除buffer
key.interestOps(key.interestOps() - SelectionKey.OP_WRITE);// 不需要关注可写事件
}
}
}
}
}
}
客户端
package com.dapeng.netty.c4;
import java.io.IOException;
import java.net.InetSocketAddress;
import java.nio.ByteBuffer;
import java.nio.channels.SocketChannel;
public class WriteClient {
public static void main(String[] args) throws IOException {
SocketChannel sc = SocketChannel.open();
sc.connect(new InetSocketAddress(8080));
int count = 0;
// 接收数据
while (true){
ByteBuffer buffer = ByteBuffer.allocate(1024 * 1024);
int read = sc.read(buffer);// 返回读取的字节数
count += read;
System.out.println(count);
buffer.clear();// 清除buffer，下次继续写入buffer。/切换到写。
}
}
}
网络编程小结：
非阻塞 VS 阻塞
1.1 阻塞
1.2 非阻塞
1.3 多路复用
selector何时不阻塞
监听channel事件
更进一步优化
利用多线程优化（现在都是多核cpu，设计时要充分考虑别让cpu的力量白白浪费）
前面的代码只有一个Selector选择器，没有充分利用到多核cpu，该如何改进呢？
分两组选择器
- 单线程配一个选择器，专门处理accept事件
- 创建cpu核心数的线程，每个线程配一个选择器，轮流处理read事件
关于多线程，一个boss里面开启worker线程案例：（需要注意顺序问题，这里使用队列 ConcurrentLinkedDeque ，通过主动唤醒selector来解决顺序问题。）
package com.dapeng.netty.c5;
import lombok.extern.slf4j.Slf4j;
import java.io.IOException;
import java.net.InetSocketAddress;
import java.nio.ByteBuffer;
import java.nio.channels.*;
import java.util.Iterator;
import java.util.concurrent.ConcurrentLinkedDeque;
import static com.dapeng.netty.utils.ByteBufferUtil.debugAll;
@Slf4j
public class ChannelThreadDemo {
public static void main(String[] args) throws IOException {
Thread.currentThread().setName("boss");
ServerSocketChannel ssc = ServerSocketChannel.open();
ssc.configureBlocking(false);//设置非阻塞模式
Selector boss = Selector.open(); // 打开选择器
SelectionKey bosskey = ssc.register(boss, 0, null);
bosskey.interestOps(SelectionKey.OP_ACCEPT);// 监听链接
ssc.bind(new InetSocketAddress(8080)); // 绑定端口号
// 1. 创建固定数量的worker 并初始化
Worker worker = new Worker("worker-0");
while (true) {
boss.select();// 阻塞监听等待连接
Iterator<SelectionKey> iter = boss.selectedKeys().iterator();
while (iter.hasNext()) {
SelectionKey key = iter.next();
iter.remove();
if (key.isAcceptable()) {
SocketChannel sc = ssc.accept();
sc.configureBlocking(false); // 设置读非阻塞。
log.debug("连接成功。。。。{}", sc.getRemoteAddress());
// 2.在这里sc需要注册到workorder的selector上面。
log.debug("关联workorder之前");
worker.register(sc); //boss调用，初始化selecotr，启动workorder-0；
log.debug("关联workorder之后");
}
}
}
}
static class Worker implements Runnable {
private Thread thread; // 线程
private Selector selector; // worker选择器
private String name; // 线程名
private volatile boolean start = false; //还未初始化 volatile保证可见性（A改了B可看见）
private ConcurrentLinkedQueue<Runnable> queue = new ConcurrentLinkedQueue<>();
public Worker(String name) {
this.name = name;
}
// 初始化线程和selector
public void register(SocketChannel sc) throws IOException {
if (!start) {
selector = Selector.open();// 开启Selector选择器
thread = new Thread(this, name);
thread.start();
start = true;
}
// 向队列添加了任务，但这个任务并没有立即执行。
queue.add(() -> {
try {
sc.register(selector, SelectionKey.OP_READ, null);
} catch (ClosedChannelException e) {
e.printStackTrace();
}
});
// 唤醒selector
selector.wakeup();
// 还有一种方法：去掉上面的队列，
// selector.wakeup();
// sc.register(selector, SelectionKey.OP_READ, null);
}
@Override
public void run() {
while (true) {
try {
selector.select();// 监听连接 先等待连接建立SocketChannel
Runnable task = queue.poll();
if (task != null) {
task.run(); //sc.register(selector, SelectionKey.OP_READ, null);
}
Iterator<SelectionKey> iterator = selector.selectedKeys().iterator();
while (iterator.hasNext()) {
SelectionKey key = iterator.next();
iterator.remove();
if (key.isReadable()) {
ByteBuffer buffer = ByteBuffer.allocate(16);
SocketChannel channel = (SocketChannel) key.channel();
log.debug("读取数据：。。。{}", channel.getRemoteAddress());
channel.read(buffer);
buffer.flip();
debugAll(buffer);
}
}
} catch (IOException e) {
e.printStackTrace();
}
}
}
}
}
多线程优化，多个worker
有个重要的点：使用Round Robin轮询调度（RR），使用计数器和worker取余，来进行轮询。（–当前小节-- 为修改的部分）
package com.dapeng.netty.c5;
import lombok.extern.slf4j.Slf4j;
import java.io.IOException;
import java.net.InetSocketAddress;
import java.nio.ByteBuffer;
import java.nio.channels.*;
import java.util.Iterator;
import java.util.concurrent.ConcurrentLinkedQueue;
import java.util.concurrent.atomic.AtomicInteger;
import static com.dapeng.netty.utils.ByteBufferUtil.debugAll;
@Slf4j
public class ChannelThreadDemo {
public static void main(String[] args) throws IOException {
Thread.currentThread().setName("boss");
ServerSocketChannel ssc = ServerSocketChannel.open();
ssc.configureBlocking(false);//设置非阻塞模式
Selector boss = Selector.open(); // 打开选择器
SelectionKey bosskey = ssc.register(boss, 0, null);
bosskey.interestOps(SelectionKey.OP_ACCEPT);// 监听链接
ssc.bind(new InetSocketAddress(8080)); // 绑定端口号
// 1. 创建固定数量的worker 并初始化
// Worker worker = new Worker("worker-0");
// --当前小节-- 1.2多线程的优化，创建多个线程,创建多个固定数量的worker，并初始化
Worker[] workers = new Worker[Runtime.getRuntime().availableProcessors()];// 获取该硬件的核心数量，部署在docker里面会有问题，会直接读取物理硬件，并不会读取docker容器里面的核心数，jdk10版本才会修复此内容。
for (int i = 0; i < workers.length; i++) {
// 给每个线程起名字
workers[i] = new Worker("worker-" + i);
}
// 定义一个计数器
AtomicInteger index = new AtomicInteger();
while (true) {
boss.select();// 阻塞监听等待连接
Iterator<SelectionKey> iter = boss.selectedKeys().iterator();
while (iter.hasNext()) {
SelectionKey key = iter.next();
iter.remove();
if (key.isAcceptable()) {
SocketChannel sc = ssc.accept();
sc.configureBlocking(false); // 设置读非阻塞。
log.debug("连接成功。。。。{}", sc.getRemoteAddress());
// 2.在这里sc需要注册到workorder的selector上面。
log.debug("关联workorder之前");
// worker.register(sc); //boss调用，初始化selecotr，启动workorder-0；
// --当前小节-- 优化,使用轮询算法，round robin
workers[index.incrementAndGet() % workers.length].register(sc); //boss调用，初始化selecotr，启动workorder-0；
log.debug("关联workorder之后");
}
}
}
}
static class Worker implements Runnable {
private Thread thread; // 线程
private Selector selector; // worker选择器
private String name; // 线程名
private volatile boolean start = false; //还未初始化
private ConcurrentLinkedQueue<Runnable> queue = new ConcurrentLinkedQueue<>();
public Worker(String name) {
this.name = name;
}
// 初始化线程和selector
public void register(SocketChannel sc) throws IOException {
if (!start) {
selector = Selector.open();// 开启Selector选择器
thread = new Thread(this, name);
thread.start();
start = true;
}
// 向队列添加了任务，但这个任务并没有立即执行。
// 队列
queue.add(() -> {
try {
sc.register(selector, SelectionKey.OP_READ, null);
} catch (ClosedChannelException e) {
e.printStackTrace();
}
});
// 唤醒selector
selector.wakeup();
// 还有一种方法：去掉上面的队列，
// selector.wakeup();
// sc.register(selector, SelectionKey.OP_READ, null);
}
@Override
public void run() {
while (true) {
try {
selector.select();// 监听连接 先等待连接建立SocketChannel
Runnable task = queue.poll();
if (task != null) {
task.run(); //sc.register(selector, SelectionKey.OP_READ, null);
}
Iterator<SelectionKey> iterator = selector.selectedKeys().iterator();
while (iterator.hasNext()) {
SelectionKey key = iterator.next();
iterator.remove();
if (key.isReadable()) {
ByteBuffer buffer = ByteBuffer.allocate(16);
SocketChannel channel = (SocketChannel) key.channel();
log.debug("读取数据：。。。{}", channel.getRemoteAddress());
channel.read(buffer);
buffer.flip();
debugAll(buffer);
}
}
} catch (IOException e) {
e.printStackTrace();
}
}
}
}
}
5. NIO vs BIO
5.1 stream VS channel
5.2 IO模型(下面的几个IO概念可能不清晰，有个大概，如果要详细看可以百度认真了解IO概念)
（同步阻塞，同步非阻塞，多路复用，异步阻塞（不存在），异步非阻塞）
1.阻塞IO，如下图：【同步阻塞】
阻塞IO会在用户调用read后，会等待内核将数据等待接手后，并复制一份数据再返回到用户。
2.非阻塞IO，如下图：【同步非阻塞】
之前的案例：while（true）的时候不停地循环读取，当从网络上得到数据后，需要阻塞住，等待数据复制完成才会继续非阻塞，并返回用户数据。（每次系统频繁调用read都会影响系统性能）
3.多路复用，如下图：【【同步】多路复用】
使用Selector的select无参的方法来阻塞住，等待有事件发生，一旦有事件发生，内核这边会告诉用户有可读事件了，用户会调用read方法去内核读取数据，内核需要阻塞住等待数据的复制完成，再将结果返回给用户。
【阻塞IO下】:
【多路复用下】
多路复用可以在selector下用一个循环把所有事件给处理掉。
同步和异步
- 同步：线程自己去获得结果（单个线程）。【严格按照顺序执行】
- 异步：线程自己不会获得结果，而是由其他线程送结果（至少两个线程）。【遇到需要等待结果的，不用等待直接执行后面程序。】
4.异步，异步情况下，不存在异步阻塞，所以只有【异步非阻塞】
用户有两个线程，线程1负责调用read方法，并立即返回结果。线程2用于等待内核数据的复制完成后，将数据作为参数，调用线程1 的回调方法，返回给用户。这样两个线程就可以实现数据的传输。
5.3零拷贝（针对java里面不用拷贝）
传统IO问题
传统的IO将一个文件通过socket写出
内部的工作流程：
NIO优化
（HeapByteBuffer使用Java内存，跟系统隔离的，使用DirectByteBuffer是跟操作系统共用的，减少了一次拷贝）
进一步优化
FileChannel有transferTo方法和transferFrom。
SocketChannel没有
Java里面的transerto 和 transeferfrom对应linux里面的sendFile方法。
进一步优化（linux2.4）
5.4 AIO
Netty在5.0版本发现异步IO性能没优势，而且提高了复杂度，所以废弃了5.0版本，最高是4.几的版本。
文件AIO
package com.dapeng.netty.AIO;
import lombok.extern.slf4j.Slf4j;
import java.io.IOException;
import java.nio.ByteBuffer;
import java.nio.channels.AsynchronousFileChannel;
import java.nio.channels.CompletionHandler;
import java.nio.file.Paths;
import java.nio.file.StandardOpenOption;
import static com.dapeng.netty.utils.ByteBufferUtil.debugAll;
@Slf4j
public class AioFileChannel {
// 多线程才能异步
public static void main(String[] args) throws IOException {
try (AsynchronousFileChannel channel = AsynchronousFileChannel.open(Paths.get("data.txt"), StandardOpenOption.READ)) {
// 参数1. ByteBuffer（读取数据到该buffer）
// 参数2. 读取的起始位置
// 参数3. 附件
// 参数4. 回调对象 CompletionHandler
ByteBuffer buffer = ByteBuffer.allocate(16);
log.debug("读取之前----read begin...");
channel.read(buffer, 0, buffer, new CompletionHandler<Integer, ByteBuffer>() {
@Override // 读取成功时候调用
public void completed(Integer result, ByteBuffer attachment) {
log.debug("读取成功---- read completed...{}",result);
attachment.flip();// 切换读模式
debugAll(attachment);
}
@Override // 读取失败
public void failed(Throwable exc, ByteBuffer attachment) {
log.error("读取失败");
exc.printStackTrace();
}
});
log.debug("读取完成----read finished...");
} catch (IOException e) {
e.printStackTrace();
}//.twr用来写try catch
System.in.read(); // 控制台输入，用来阻塞主线程，来等待子守护线程的读取结果，如果不阻塞，会导致回调对象新开的线程也会结束
}
}
网络AIO
暂时没有相关资料，可以百度搜搜