• Netty-NIO

    文章目录

    一、NIO-Selector

    1.处理accept

    //1.创建selector,管理多个channel
Selector selector = Selector.open();
ByteBuffer buffer = ByteBuffer.allocate(16);
ServerSocketChannel ssc = ServerSocketChannel.open();
ssc.configureBlocking(false);
//2.建立selector和channel的联系（注册）
//SelectionKey就是将来事件发生后，通过它可以知道事件和哪个channel的事件
//四个事件：
//accept 会在有连接请求时触发
//connect 是客户端，连接建立后触发
//read 可读事件
//write 可写事件
SelectionKey sscKey = ssc.register(selector, 0, null);
sscKey.interestOps(SelectionKey.OP_ACCEPT);
ssc.bind(new InetSocketAddress(8080));
while(true){
	//3.select方法，没有事件发生，线程阻塞，有事件，线程才会恢复运行
	selector.select();
	//4.处理事件，selectedKeys内部包含了所有发生的事件
	Iterator<SelectionKey> iter = selector.selectedKeys.iterator();
	while(iter.next()){
		SelectionKey key = iter.next();
		ServerSocketChannel channel = (ServerSocketChannel)key.channel();
		SocketChannel sc = channel.accept();
	}
}

    • 1
    • 2
    • 3
    • 4
    • 5
    • 6
    • 7
    • 8
    • 9
    • 10
    • 11
    • 12
    • 13
    • 14
    • 15
    • 16
    • 17
    • 18
    • 19
    • 20
    • 21
    • 22
    • 23
    • 24
    • 25
    • 26

    2.cancel

    //1.创建selector,管理多个channel
Selector selector = Selector.open();
ByteBuffer buffer = ByteBuffer.allocate(16);
ServerSocketChannel ssc = ServerSocketChannel.open();
ssc.configureBlocking(false);
//2.建立selector和channel的联系（注册）
SelectionKey sscKey = ssc.register(selector, 0, null);
sscKey.interestOps(SelectionKey.OP_ACCEPT);
ssc.bind(new InetSocketAddress(8080));
while(true){
	//3.select方法，没有事件发生，线程阻塞，有事件，线程才会恢复运行
	//select在事件未处理时，它不会阻塞，事件发生后要么处理，要么取消，不能置之不理
	selector.select();
	//4.处理事件，selectedKeys内部包含了所有发生的事件
	Iterator<SelectionKey> iter = selector.selectedKeys.iterator();
	while(iter.next()){
		SelectionKey key = iter.next();
		key.cancel();
	}
}

    • 1
    • 2
    • 3
    • 4
    • 5
    • 6
    • 7
    • 8
    • 9
    • 10
    • 11
    • 12
    • 13
    • 14
    • 15
    • 16
    • 17
    • 18
    • 19
    • 20

    3.处理read

    用完key必须要remove

    //1.创建selector,管理多个channel
Selector selector = Selector.open();
ByteBuffer buffer = ByteBuffer.allocate(16);
ServerSocketChannel ssc = ServerSocketChannel.open();
ssc.configureBlocking(false);
//2.建立selector和channel的联系（注册）
SelectionKey sscKey = ssc.register(selector, 0, null);
sscKey.interestOps(SelectionKey.OP_ACCEPT);
ssc.bind(new InetSocketAddress(8080));
while(true){
	//3.select方法，没有事件发生，线程阻塞，有事件，线程才会恢复运行
	selector.select();
	//4.处理事件，selectedKeys内部包含了所有发生的事件
	//selector会在发生事件后，向集合中加入key,但不会删除
	Iterator<SelectionKey> iter = selector.selectedKeys.iterator();
	while(iter.next()){
		SelectionKey key = iter.next();
		//处理key时，要从selectedKeys集合中删除，否则下次处理就会有问题
		iter.remove();
		//5.区分事件类型
		if(key.isAcceptable()){ //如果是accept
			ServerSocketChannel channel = (ServerSocketChannel)key.channel();
			SocketChannel sc = channel.accept();
			sc.configureBlocking(false);
			SelectionKey sckey = sc.register(selector, 0, null);
			scKey.interestOps(SelectionKey.OP_READ);
		}elseif(key.isReadable()){
			//拿到触发事件的channel
			ServerSocketChannel channel = (ServerSocketChannel)key.channel();
			ByteBuffer buffer = ByteBuffer.allocate(16);
			channel.read(buffer);
			buffer.flip();
			debugRead(buffer);
		}
	}
}

    • 1
    • 2
    • 3
    • 4
    • 5
    • 6
    • 7
    • 8
    • 9
    • 10
    • 11
    • 12
    • 13
    • 14
    • 15
    • 16
    • 17
    • 18
    • 19
    • 20
    • 21
    • 22
    • 23
    • 24
    • 25
    • 26
    • 27
    • 28
    • 29
    • 30
    • 31
    • 32
    • 33
    • 34
    • 35
    • 36

    4.处理客户端断开

    //1.创建selector,管理多个channel
Selector selector = Selector.open(); 
ByteBuffer buffer = ByteBuffer.allocate(16); 
ServerSocketChannel ssc = ServerSocketChannel.open(); 
ssc.configureBlocking(false);
//2.建立selector和channel的联系（注册）
SelectionKey sscKey = ssc.register(selector, 0, null); sscKey.interestOps(SelectionKey.OP_ACCEPT); 
ssc.bind(new InetSocketAddress(8080)); 
while(true){ 
	//3.select方法，没有事件发生，线程阻塞，有事件，线程才会恢复运行 
	selector.select(); 
	//4.处理事件，selectedKeys内部包含了所有发生的事件 
	//selector会在发生事件后，向集合中加入key,但不会删除 
	Iterator<SelectionKey> iter = selector.selectedKeys.iterator(); 
	while(iter.next()){ 		
		SelectionKey key = iter.next(); 
		//处理key时，要从selectedKeys集合中删除，否则下次处理就会有问题 
		iter.remove(); 
		//5.区分事件类型 
		if(key.isAcceptable()){ 
			//如果是accept 
			ServerSocketChannel channel = (ServerSocketChannel)key.channel(); 
			SocketChannel sc = channel.accept();
		 	sc.configureBlocking(false); 
		 	SelectionKey sckey = sc.register(selector, 0, null); 
		 	scKey.interestOps(SelectionKey.OP_READ); 
		 }elseif(key.isReadable()){ 
		 	try{ 
			 	//拿到触发事件的channel 
			 	ServerSocketChannel channel = (ServerSocketChannel)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 的keys集合中真正删除key) 
		 	 	key.cancel();
			}
		}
	}
}

    • 1
    • 2
    • 3
    • 4
    • 5
    • 6
    • 7
    • 8
    • 9
    • 10
    • 11
    • 12
    • 13
    • 14
    • 15
    • 16
    • 17
    • 18
    • 19
    • 20
    • 21
    • 22
    • 23
    • 24
    • 25
    • 26
    • 27
    • 28
    • 29
    • 30
    • 31
    • 32
    • 33
    • 34
    • 35
    • 36
    • 37
    • 38
    • 39
    • 40
    • 41
    • 42
    • 43
    • 44
    • 45
    • 46

    5. 处理消息的边界

    1. 固定消息长度，数据包大小一样，服务器按预定长度读取，缺点是浪费带宽
    2. 按分隔符拆分，缺点是效率低
    3. TLV格式，Type类型,Length长度，Value数据，可以方便获取消息大小，分配合适的buffer，缺点是buffer需要提前分配，如果内容过大，影响server吞吐量
      • Http1.1是TLV格式
      • Http2.0是LTV格式
    private static void split(ByteBuffer source){
	source.flip();
	for(int i = 0; i < source.limit(); i++){
		//找到一条完整消息
		if(source.get(i) == '\n'){
			int length = i + 1 -source.position();
			//把这条完整消息存入新的ByteBuffer
			ByteBuffer target = ByteBuffer.allocate(length);
			//从source读，向target写
			for(int j = 0; j < length; j++){
				target.put(source.get());
			}
			debugAll(target);
		}
	}
	source.compact();
}

public static void main(){
	//1.创建selector,管理多个channel
	Selector selector = Selector.open(); 
	ByteBuffer buffer = ByteBuffer.allocate(16); 
	ServerSocketChannel ssc = ServerSocketChannel.open(); 
	ssc.configureBlocking(false);
	//2.建立selector和channel的联系（注册）
	SelectionKey sscKey = ssc.register(selector, 0, null); 		
	sscKey.interestOps(SelectionKey.OP_ACCEPT); 
	ssc.bind(new InetSocketAddress(8080)); 
	while(true){ 
		//3.select方法，没有事件发生，线程阻塞，有事件，线程才会恢复运行 
		selector.select(); 
		//4.处理事件，selectedKeys内部包含了所有发生的事件 
		//selector会在发生事件后，向集合中加入key,但不会删除 
		Iterator<SelectionKey> iter = selector.selectedKeys.iterator(); 
		while(iter.next()){ 		
			SelectionKey key = iter.next(); 
			//处理key时，要从selectedKeys集合中删除，否则下次处理就会有问题 
			iter.remove(); 
			//5.区分事件类型 
			if(key.isAcceptable()){ 
				//如果是accept 
				ServerSocketChannel channel = (ServerSocketChannel)key.channel(); 
				SocketChannel sc = channel.accept();
			 	sc.configureBlocking(false); 
			 	ByteBuffer buffer = ByteBuffer.allocate(16); //attachment附件
			 	//将一个byteBuffer作为附件关联到selectionKey上
			 	SelectionKey sckey = sc.register(selector, 0, buffer); 
			 	scKey.interestOps(SelectionKey.OP_READ); 
			 }elseif(key.isReadable()){ 
			 	try{ 
				 	//拿到触发事件的channel 
				 	ServerSocketChannel channel = (ServerSocketChannel)key.channel(); 
				 	//获取selectionKey上关联的附件
				 	ByteBuffer buffer = (ByteBuffer)key.attatchment();
				 	int read = channel.read(buffer);//如果是正常断开，read的方法的返回值是-1 
				 	if（read == -1){ 
				 		key.cancel(); 
				 	}else{ 
				 		split(buffer);
				 		if(buffer.position() == buffer.limit()){
				 			//扩容
				 			ByteBuffer newBuffer = ByteBuffer.allocate(buffer.capacity()*2);
				 			buffer.flip();
				 			newBuffer.put(buffer);//复制
				 			key.attach(newbuffer);//替换掉key上原有的buffer
				 		}
			 		} 
			 	}catch(IOException e){ 
			 		e.printStackTrace();
			 	 	//因为客户端断开了，因此需要将key取消（从selector 的keys集合中真正删除key) 
			 	 	key.cancel();
				}
			}
		}
	}
}

    • 1
    • 2
    • 3
    • 4
    • 5
    • 6
    • 7
    • 8
    • 9
    • 10
    • 11
    • 12
    • 13
    • 14
    • 15
    • 16
    • 17
    • 18
    • 19
    • 20
    • 21
    • 22
    • 23
    • 24
    • 25
    • 26
    • 27
    • 28
    • 29
    • 30
    • 31
    • 32
    • 33
    • 34
    • 35
    • 36
    • 37
    • 38
    • 39
    • 40
    • 41
    • 42
    • 43
    • 44
    • 45
    • 46
    • 47
    • 48
    • 49
    • 50
    • 51
    • 52
    • 53
    • 54
    • 55
    • 56
    • 57
    • 58
    • 59
    • 60
    • 61
    • 62
    • 63
    • 64
    • 65
    • 66
    • 67
    • 68
    • 69
    • 70
    • 71
    • 72
    • 73
    • 74
    • 75
    • 76

    6. 写入内容过多的问题

    //服务器
public static void main(){
	ServerSocketChannel ssc = ServerSocketChannrl.open();
	ssc.configureBlocking(false);
	Selector selector = Selector.open();
	ssc.register(selector, SelectionKey.OP_ACCEPT);
	ssc.bind(new InetSocketAddress(8080));
	while(trye){
		selector.select();
		Iterator<SelectionKey> iter = selector.selectedKeys.iterator();
		while(iter.hasNext()){
			SelectionKey key = iter.next();
			iter.remove();
			if(key.isAcceptable()){
				SocketChannel sc = ssc.accept();
				sc.configureBlocking(false);
				//1.向客户端发送大量数据
				StringBuilder sb = new StringBuilder();
				for(int i = 0; i < 3000000; i++){
					sb.append("a");
				}
				BytrBuffer buffer = Charset.defaultCharset().encode(sb.toString());
				//不符合非阻塞模式
				while(buffer.hasRemaining()){
					//2.返回值代表实际写入的字节数
					//不能一次性写完
					//write == 0 缓冲区满，写不了
					int write = sc.write(buffer);
					System.out.println(write):
				}
			}
		}
	}
}

//客户端
public static void main(){
	SocketChannel sc = SocketChannel.open();
	sc.connect(new InetSocketAddress("localhost",8080));
	//3.接收数据
	int count = 0;
	while(true){
		ByteBuffer buffer = ByteBuffer.allocate(1024*1024);
		count += sc.read(buffer);
		System.out.println(count);
		buffer.clear();
	}
}

    • 1
    • 2
    • 3
    • 4
    • 5
    • 6
    • 7
    • 8
    • 9
    • 10
    • 11
    • 12
    • 13
    • 14
    • 15
    • 16
    • 17
    • 18
    • 19
    • 20
    • 21
    • 22
    • 23
    • 24
    • 25
    • 26
    • 27
    • 28
    • 29
    • 30
    • 31
    • 32
    • 33
    • 34
    • 35
    • 36
    • 37
    • 38
    • 39
    • 40
    • 41
    • 42
    • 43
    • 44
    • 45
    • 46
    • 47
    • 48

    7. 处理可写事件

    //服务器
public static void main(){
	ServerSocketChannel ssc = ServerSocketChannrl.open();
	ssc.configureBlocking(false);
	Selector selector = Selector.open();
	ssc.register(selector, SelectionKey.OP_ACCEPT);
	ssc.bind(new InetSocketAddress(8080));
	while(trye){
		selector.select();
		Iterator<SelectionKey> iter = selector.selectedKeys.iterator();
		while(iter.hasNext()){
			SelectionKey key = iter.next();
			iter.remove();
			if(key.isAcceptable()){
				SocketChannel sc = ssc.accept();
				sc.configureBlocking(false);
				SelectionKey sckey = sc.register(selector, 0, null);
				sckey.interestOps(SelectionKey.OP_READ);
				//1.向客户端发送大量数据
				StringBuilder sb = new StringBuilder();
				for(int i = 0; i < 3000000; i++){
					sb.append("a");
				}
				BytrBuffer buffer = Charset.defaultCharset().encode(sb.toString());
				//2.返回值代表实际写入的字节数
				//不能一次性写完
				//先写一次
				int write = sc.write(buffer);
				System.out.println(write):
				//3.判断是否有剩余内容
				while(buffer.hasRemaining()){
					//4.关注可写事件
					sckey.interestOps(sckey.interestOps() + SelectionKey.OP_WRITE);
					//sckey.interestOps(sckey.interestOps() | SelectionKey.OP_WRITE);
					//5.把未写完的数据挂到sckey上
					sckey.attach(buffer);
				}
			}elseif(key.isWritable())[
				ByteBuffer buffer = (ByteBuffer) key.attachment();
				SocketChannel sc = (SocketChannel)key.channel();
				int write = sc.write(buffer);
				System.out.println(write):
				//6.清理操作，内存释放
				if（!buffer.haeRemaining()){
					key.attach(null);//需要清除buffer
					key.interestOps(key.interestOps() - SelectionKey.OP_WRITE);//不需关注可写事件
				}
			}
		}
	}
}

    • 1
    • 2
    • 3
    • 4
    • 5
    • 6
    • 7
    • 8
    • 9
    • 10
    • 11
    • 12
    • 13
    • 14
    • 15
    • 16
    • 17
    • 18
    • 19
    • 20
    • 21
    • 22
    • 23
    • 24
    • 25
    • 26
    • 27
    • 28
    • 29
    • 30
    • 31
    • 32
    • 33
    • 34
    • 35
    • 36
    • 37
    • 38
    • 39
    • 40
    • 41
    • 42
    • 43
    • 44
    • 45
    • 46
    • 47
    • 48
    • 49
    • 50
    • 51

    二、多线程优化

    前面的代码只有一个选择器，没有充分利用多核cpu，如何改进呢？
    分两组选择器：（boss建立连接，worker负责数据读写）

    • 单线程配一个选择器，专门处理accept事件
    • 创建cpu核心数的线程，每个线程配一个选择器，轮流处理read事件
    public static void main(){
	Thread.currentThrea().setName("boss");
	ServerSocketChannel ssc = ServerSocketChannel.open();
	ssc.configuraBlocking(flase);
	Selector boss = Selector.open();
	SelectionKey bosskey = ssc.register(boss, 0, null);
	bosskey.interestOps(SelectionKey.OP_ACCEPT):
	ssc.bind(new InetSocketAddress(8080));
	//1.创建固定数量的worker并初始化
	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){
		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):
				//2.关联selector
				//轮询
				workers[index.getAndIncrement() % workers.length}.register(sc);
			}
		}
	}
}

static class Worker implements Runnable{
	private Thread thread;
	private Selector worker;
	private String name;
	private volatile boolean star = false;//还未初始化
	private ConcurrentLinkedQueue<Runnable> queue = new ConcurrentLinkedQueue<>():

	public Worker(String name){
		this.name = name;
	}

	//初始化线程和selector
	public void register(SocketChannel sc){
		if（!start){
			selector = Selector.open();
			thread = new Thread(this, name);
			thread.start():
			start = true;
		}
		//向队列添加任务，但这个任务并没有被boss立刻执行
		queue.add()->{
			try{
				sc.register(worker.selector, SelectionKey.OP_READ, null);
			}catch(ClosedChannelException e) {
				e.printStackTrace()
			}
		}
		//唤醒run（）中的select方法
		selector.wakeup();
		//也可以用以下方式，先wakeup，后select阻塞时也能被唤醒
		/* selector.wakeup();
		sc.register(worker.selector, SelectionKey.OP_READ, null);*/
	}

	@Override
	public void run(){
		while(true){
			try{
				worker.select();
				Runnable task = queue.poll();
				if(task != null){
					task.run();
				}
				Iterator<SelectionKey> iter = worker.selectedKeys().iterator();
				while(iter.hasNext()){
					SlectionKey key = iter.next();
					iter.remove();
					if(key.isReadable()){
						ByteBuffer buffer = ByteBuffer.allocate(16);
						SocketChannel channel = (SocketChannel)key.channel();
						channel.read(buffer);
						buffer.flip();
						debugAll(buffer);
					}
				}
			}catch(IOException e){
				e.printStackTrace();
			}
		}
	}
}

    • 1
    • 2
    • 3
    • 4
    • 5
    • 6
    • 7
    • 8
    • 9
    • 10
    • 11
    • 12
    • 13
    • 14
    • 15
    • 16
    • 17
    • 18
    • 19
    • 20
    • 21
    • 22
    • 23
    • 24
    • 25
    • 26
    • 27
    • 28
    • 29
    • 30
    • 31
    • 32
    • 33
    • 34
    • 35
    • 36
    • 37
    • 38
    • 39
    • 40
    • 41
    • 42
    • 43
    • 44
    • 45
    • 46
    • 47
    • 48
    • 49
    • 50
    • 51
    • 52
    • 53
    • 54
    • 55
    • 56
    • 57
    • 58
    • 59
    • 60
    • 61
    • 62
    • 63
    • 64
    • 65
    • 66
    • 67
    • 68
    • 69
    • 70
    • 71
    • 72
    • 73
    • 74
    • 75
    • 76
    • 77
    • 78
    • 79
    • 80
    • 81
    • 82
    • 83
    • 84
    • 85
    • 86
    • 87
    • 88
    • 89
    • 90
    • 91
    • 92
    • 93

    三、NIO概念剖析

    1. stream 和 channel

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

    2. IO模型

    2.1 阻塞IO

    在这里插入图片描述
    用户线程被阻塞（同步）

    2.2 非阻塞IO

    在这里插入图片描述
    read是中运行，无数据立刻返回，有数据复制完返回
    等待数据非阻塞，复制数据阻塞（同步）
    缺点：多测内核切换

    2.3多路复用

    在这里插入图片描述
    select等待数据阻塞，read复制数据阻塞（同步）
    阻塞IO
    多路复用
    一次性处理多个channel上的事件

    2.4 同步异步

    同步：线程自己去获取结果（一个线程）
    异步：一个线程发送，一个线程送结果（两个线程）
    异步非阻塞
    read非阻塞

    异步阻塞不存在

    3. 零拷贝

    传统io将一个文件通过socket写出，内部工作流程：
    在这里插入图片描述
    用户和内核态的切换发生了3次，这个操作比较重量级
    数据拷贝了4次

    3.1 NIO优化

    通过DirectByteBuffer
    ByteBuffer.allocate(10) ,返回HeapByteBuffer,使用Java内存
    ByteBuffer.allocateDirect(10),返回DirectByteBuffer,使用操作系统内存
    在这里插入图片描述
    java可以使用DirectByteBuffer将堆内存映射到jvm内存中来直接访问使用
    减少了一次数据拷贝，用户态与内核态的切换次数没有减少

    3.2 sendFile优化

    Linux2.1后提供sendFile方法，Java中对应着两个channel调用transferTo/transferFrom方法拷贝数据
    在这里插入图片描述
    只发生了一次用户态与内核态的切换
    数据拷贝了3次

    3.3 进一步优化

    在这里插入图片描述
    一次切换，2次拷贝

    零拷贝，并不是真正无拷贝，而是在不会拷贝重复数据到jvm内存中，零拷贝的优点有：

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

    4. AIO(异步IO）

    netty不支持异步IO

    public static void main(){
	try(AsynchronousFileChannel channel = AsynchronousFileChannel.open(Paths.get("data.txt"), StandardOpenOption.READ)){
		//参数1 ByteBuffer
		//读取的起始位置
		//附件
		//回调对象 CompletionHandler
		ByteBuffer buffer =ByteBuffer.allocate(16);
		channel.read(buffer, 0, buffer, new CompletionHandler<Integer, ByteBuffer>(){
			@Override//read成功
			public void completed(Integer result, ByteBuffer attachment){
				attachment.flip();
				debugAll(attachment);
			}
			@Override// read失败
			public void failed(Throwable exc, ByteBuffer attachment){
				exc.printStachTrace();
			}
		}):
	}catch(IOException e){
		e.printStackTrace();
	}
	//主线程结束，守护线程结束
	//接收控制台的输入，控制台不输入，就停在这儿
	System.in.read();
}

    • 1
    • 2
    • 3
    • 4
    • 5
    • 6
    • 7
    • 8
    • 9
    • 10
    • 11
    • 12
    • 13
    • 14
    • 15
    • 16
    • 17
    • 18
    • 19
    • 20
    • 21
    • 22
    • 23
    • 24
    • 25
  • 相关阅读:
    数字孪生：实现保险行业数字化转型
    elf文件与链接脚本
    ＜ Linux ＞ 进度条小程序 + git三板斧
    前端测试&OA工程师区别
    【python中级】func_timeout程序超时处理
    WHERE‘S MY PAYROLL‽‽‽
    [nlp] 索引：正向索引&倒排索引
    数据结构-复杂度（一）
    Jetson JetPack-5.1.2-L4T-R35.4.1 修复libvargus内存损坏问题
    Day16--购物车页面-演示效果并创建编译模式
  • 原文地址：https://blog.csdn.net/qq_44473695/article/details/132523288