关于BIO(即Java传统的网络编程,参考)

NIO网络编程

1.逐步演变

1.单线程阻塞模式

创建服务器

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;/**  NIO开发客户端* */
public class NIO_Blocking_Server {public static void main(String[] args) throws Exception {//得到一个bufferByteBuffer buffer = ByteBuffer.allocate(30);//NIO的API创建服务器ServerSocketChannel socketChannel = ServerSocketChannel.open();//绑定端口socketChannel.bind(new InetSocketAddress("localhost", 9091));//创建一个集合,将获取到的连接放到集合中List<SocketChannel> channels = new ArrayList<>();while (true) {//获取连接 默认是阻塞方法,获取不到连接一直阻塞。SocketChannel channel = socketChannel.accept();channels.add(channel);for (SocketChannel chann : channels) {//read()默认是阻塞 没有数据读那么会一直阻塞。int read = chann.read(buffer);buffer.flip();//...处理数据 这里直接打印出来for (int i = 0; i < buffer.limit(); i++) {System.out.print((char) buffer.get() + " ");}buffer.clear();}}}
}

客户端

import java.io.IOException;
import java.net.InetSocketAddress;
import java.nio.channels.SocketChannel;
import java.nio.charset.Charset;public class NIO_Client {public static void main(String[] args) throws IOException {//NIO API 创建一个客户端channelSocketChannel clientChannel = SocketChannel.open();//连接服务器clientChannel.connect(new InetSocketAddress("localhost", 9091));//写数据clientChannel.write(Charset.defaultCharset().encode("sdsdsd"));}
}

概述

  • 默认NIO实现的网络编程还是一个阻塞式的,即accept()获取客户端连接以及进行read()读取数据都是阻塞式的。
  • 这种情况跟Java传统的BIO没有什么区别,阻塞式的缺点channel之间会进行影响,比如说某一个客户端进行了连接,但是一直没有写数据,服务器调用channel.read()时那么就会一直阻塞,那么其他的客户端都连接不到服务器 (因为是单线程),并且可能其他的连接channel有数据了,因为这个没有数据的连接的阻塞而一直处理不了

基于这些缺点,最大的缺点就是单线程阻塞下各个channel会互相影响,所以多线程的阻塞模式以及非阻塞模式就衍生出来

2.多线程阻塞模式

  • 多线程阻塞式的核心就是,在获取到了客户端的连接channel后,开启一个新的线程去处理这个channel的读写请求,然后服务器就能继续处理连接,代码参考BIO多线程模式
  • 这种模式的好处就是连接之间不会相互的影响,服务器只要获取了一个连接就开启线程去处理读写,上一个连接的读写不会影响其他的连接的建立。
  • 弊端就是这种模式非常的消耗资源,每来一个连接就开一个线程,那么在大量的连接下服务器会崩掉。(可以使用线程池优化)

多线程下的阻塞模式可以解决一些问题,但是非常消耗资源,这在高并发系统中是几乎不可能使用的一种模式,所以必须弃用多线程的设计,要解决各个channel之间会进行影响除了从多线程角度解决问题,还可以从阻塞这个角度解决问题。单线程下我们可不可以将accept()和read()这两个阻塞方法变为非阻塞呢,那么各个连接之间就不会相互影响了 那么NIO的非阻塞API来了~

3.非阻塞NIO模式

服务器

/**  NIO开发客户端* */
public class NIO_NoBlocking_Server {public static void main(String[] args) throws Exception {//申请一个bufferByteBuffer buffer = ByteBuffer.allocate(30);//NIO的API创建服务器ServerSocketChannel socketChannel = ServerSocketChannel.open();/** 设置服务器连接为非阻塞* 即accept()方法为非阻塞 获取不到连接直接返回null。* */socketChannel.configureBlocking(false);//绑定端口socketChannel.bind(new InetSocketAddress("localhost", 9090));List<SocketChannel> channels = new ArrayList<>();while (true) {//上面设置为false后,变为非阻塞,获取不到连接直接返回null。SocketChannel channel = socketChannel.accept();/** 设置连接读为非阻塞* 即read()方法为非阻塞* */if (channel != null) {channel.configureBlocking(false);channels.add(channel);}for (SocketChannel chann : channels) {/** 经过设置为false之后,read()变为非阻塞,没有数据返回0* */int len = chann.read(buffer);//len == 0说明此客户端还未发送来数据if(len == 0) continue;//。。。有数据,处理数据buffer.flip();//....处理数据buffer.clear();}}}
}

总结

可以看到,使用NIO提供的非阻塞API只需要修改两个配置即可

// 1.将获取客户端连接的方法变为非阻塞,即获取不到直接返回NULL,不阻塞。
socketChannel.configureBlocking(false);// 2.将从连接channel中读取数据变为非阻塞,没有数据时直接返回0,不阻塞。
channel.configureBlocking(false);

NIO非阻塞的模式的工作模式

缺点

  • 在这种工作模式下,我们的线程没有阻塞,循环一直在工作,那么就会一直占用CPU(即使没有连接channel,也会一直循环),并且每次我们都需要遍历整个channel集合,调用其read()方法判断channel是否有数据就绪,我们知道read()方法底层是一次系统调用(OS的知识),会涉及到用户态以及核心态的转换,是非常消耗资源的。总结起来就是:没有连接到来时会循环,没有数据可读时也会循环
  • 所以我们希望的是 ,当某个或者某些channel有数据就绪或者需要向channel写数据时(即有事件发生时)告诉我们,我们才处理,而不是直接遍历所有的channel去挨个判断。这就是IO多路复用的思想

Java的开发者们当然想到了这一点,在NIO中为我们提供了一个Selector,下面详细讲解Selector的使用

2.Selector使用详解

首先看一下NIO中定义的四种事件类型

四种事件类型

  • accept (服务器端) ServerSocketChannel有连接请求时触发
  • connect (客户端) 连接建立后触发
  • read (服务器端) 建立连接后SocketChannel可读事件
  • write (服务器端) 建立连接后SocketChannel可写事件
//如果一个channel需要绑定多个事件,直接将多个相加即可(或者'|'起来 '|' => 不进位加法)public static final int OP_READ = 1 << 0;public static final int OP_WRITE = 1 << 2;public static final int OP_CONNECT = 1 << 3;public static final int OP_ACCEPT = 1 << 4;

2.1Selector简单介绍

#mermaid-svg-eDJnmhyhtNSoEj8A {font-family:"trebuchet ms",verdana,arial,sans-serif;font-size:16px;fill:#333;}#mermaid-svg-eDJnmhyhtNSoEj8A .error-icon{fill:#552222;}#mermaid-svg-eDJnmhyhtNSoEj8A .error-text{fill:#552222;stroke:#552222;}#mermaid-svg-eDJnmhyhtNSoEj8A .edge-thickness-normal{stroke-width:2px;}#mermaid-svg-eDJnmhyhtNSoEj8A .edge-thickness-thick{stroke-width:3.5px;}#mermaid-svg-eDJnmhyhtNSoEj8A .edge-pattern-solid{stroke-dasharray:0;}#mermaid-svg-eDJnmhyhtNSoEj8A .edge-pattern-dashed{stroke-dasharray:3;}#mermaid-svg-eDJnmhyhtNSoEj8A .edge-pattern-dotted{stroke-dasharray:2;}#mermaid-svg-eDJnmhyhtNSoEj8A .marker{fill:#333333;stroke:#333333;}#mermaid-svg-eDJnmhyhtNSoEj8A .marker.cross{stroke:#333333;}#mermaid-svg-eDJnmhyhtNSoEj8A svg{font-family:"trebuchet ms",verdana,arial,sans-serif;font-size:16px;}#mermaid-svg-eDJnmhyhtNSoEj8A .label{font-family:"trebuchet ms",verdana,arial,sans-serif;color:#333;}#mermaid-svg-eDJnmhyhtNSoEj8A .cluster-label text{fill:#333;}#mermaid-svg-eDJnmhyhtNSoEj8A .cluster-label span{color:#333;}#mermaid-svg-eDJnmhyhtNSoEj8A .label text,#mermaid-svg-eDJnmhyhtNSoEj8A span{fill:#333;color:#333;}#mermaid-svg-eDJnmhyhtNSoEj8A .node rect,#mermaid-svg-eDJnmhyhtNSoEj8A .node circle,#mermaid-svg-eDJnmhyhtNSoEj8A .node ellipse,#mermaid-svg-eDJnmhyhtNSoEj8A .node polygon,#mermaid-svg-eDJnmhyhtNSoEj8A .node path{fill:#ECECFF;stroke:#9370DB;stroke-width:1px;}#mermaid-svg-eDJnmhyhtNSoEj8A .node .label{text-align:center;}#mermaid-svg-eDJnmhyhtNSoEj8A .node.clickable{cursor:pointer;}#mermaid-svg-eDJnmhyhtNSoEj8A .arrowheadPath{fill:#333333;}#mermaid-svg-eDJnmhyhtNSoEj8A .edgePath .path{stroke:#333333;stroke-width:2.0px;}#mermaid-svg-eDJnmhyhtNSoEj8A .flowchart-link{stroke:#333333;fill:none;}#mermaid-svg-eDJnmhyhtNSoEj8A .edgeLabel{background-color:#e8e8e8;text-align:center;}#mermaid-svg-eDJnmhyhtNSoEj8A .edgeLabel rect{opacity:0.5;background-color:#e8e8e8;fill:#e8e8e8;}#mermaid-svg-eDJnmhyhtNSoEj8A .cluster rect{fill:#ffffde;stroke:#aaaa33;stroke-width:1px;}#mermaid-svg-eDJnmhyhtNSoEj8A .cluster text{fill:#333;}#mermaid-svg-eDJnmhyhtNSoEj8A .cluster span{color:#333;}#mermaid-svg-eDJnmhyhtNSoEj8A div.mermaidTooltip{position:absolute;text-align:center;max-width:200px;padding:2px;font-family:"trebuchet ms",verdana,arial,sans-serif;font-size:12px;background:hsl(80, 100%, 96.2745098039%);border:1px solid #aaaa33;border-radius:2px;pointer-events:none;z-index:100;}#mermaid-svg-eDJnmhyhtNSoEj8A :root{--mermaid-font-family:"trebuchet ms",verdana,arial,sans-serif;}

selector 版
selector
thread
channel
channel
channel

好处

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

2.1.1创建

Selector selector = Selector.open();

2.1.2绑定 Channel 事件

也称之为注册事件,绑定的事件 selector 才会关心

channel.configureBlocking(false);
SelectionKey key = channel.register(selector, 绑定事件);
  • channel 必须工作在非阻塞模式
  • FileChannel 没有非阻塞模式,因此不能配合 selector 一起使用
  • 绑定的事件类型可以有
    • connect - 客户端连接成功时触发
    • accept - 服务器端成功接受连接时触发
    • read - 数据可读入时触发,有因为接收能力弱,数据暂不能读入的情况
    • write - 数据可写出时触发,有因为发送能力弱,数据暂不能写出的情况

2.1.3监听 Channel 事件

可以通过下面三种方法来监听是否有事件发生,方法的返回值代表有多少 channel 发生了事件

方法1,阻塞直到绑定事件发生

int count = selector.select();

方法2,阻塞直到绑定事件发生,或是超时(时间单位为 ms)

int count = selector.select(long timeout);

方法3,不会阻塞,也就是不管有没有事件,立刻返回,自己根据返回值检查是否有事件

int count = selector.selectNow();

2.1.4select 何时不阻塞

  • 事件发生时

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

2.2处理accept()/read()

注意点:

  • 不管是什么类型的Channel,想要注册到Selector上,必须设置Channel为非阻塞模式

  • 遍历事件发生的key的集合时(注意一定要使用迭代器遍历,为了删除key)注意:

    1. 在默认的水平触发模式下,所有的事件必须进行处理或者cancel(),不处理的话下次进行select()时,Selctor还认为Channel上发生了事件的就绪,那就不会阻塞,导致一直循环下去。

    2. 使用迭代器遍历得到key后,必须删除key,否则下次会空指针异常。

    看一下Slector的原理然后解释:

    解释第一点:事件处理的标志就是在左边的集合中(keys集合)每一个Key发生事件时有一个标志位,处理了那么标志位就会清除,cancel()是直接将Key删除,即直接将Channel从keys集合中移除掉。如果没有处理或者cancel(),Selector发现未处理,则继续让你处理,不会阻塞,导致一直循环。

    解释第二点,每次遍历事件就绪集合时,必须手动调用迭代器的remove()将拿到的key删除,否则下次遍历右边就绪集合(selectedKeys集合)时,可能这个key对应的channel没有发生事件,那么还处理时就可能出现异常。

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

  • 在处理读事件时,必须要使用try-catch处理,并且判断channel.read()的返回值情况进行处理

    因为客户端在正常或者非正常断开时,会触发一次 read() 事件,如果不处理可能会导致服务器直接挂或者导致select()方法不阻塞的严重问题

    详细处理见代码

public class NIO_Selector {/** 注意:实际操作中需要将所有的代码使用try-catch-finally处理* 这里只是为了讲解使用,只讲核心的代码使用tcf处理。*/public static void main(String[] args) throws IOException {/** 创建一个Selector,可以管理多个不同种类的channel,* 使用时直接将channel注册到Selector上即可* */Selector selector = Selector.open();//创建一个ServerServerSocketChannel serverSocketChannel = ServerSocketChannel.open();serverSocketChannel.bind(new InetSocketAddress("localhost", 9090));//注意:要注册到Selector那么channel必须设置为非阻塞serverSocketChannel.configureBlocking(false);/** 注册,将此负责获取连接的Channel注册到Selector上,* 设置Selector关注此Channel上accept()事件即连接事件的发生,返回一个关联的key* 四种事件 OP_ACCEPT / OP_CONNECT / OP_READ / OP_WRITE*/SelectionKey serverSocketKey = serverSocketChannel.register(selector, SelectionKey.OP_ACCEPT);while (true) {/** 有事件就绪时执行,没有事件就绪时阻塞* 返回值是就绪的Channel的个数*/int n = selector.select();//返回一个Set集合,包含了所有有事件发生的channel的key。Set<SelectionKey> keys = selector.selectedKeys();Iterator<SelectionKey> iterator = keys.iterator();//迭代器遍历可以在遍历的同时进行删除while (iterator.hasNext()) {SelectionKey key = iterator.next();//删除keyiterator.remove();/** 注意:这里是根据注册时设置的事件(OP_XX)来判断的。* */if (key.isAcceptable()) {/** 获取key关联到selector上的channel,说明此serverChannel有连接到来* 因此能走到这个判断里,那么当前key一定是ServerSocketChannel,* 所以调用key.channel()时,直接强转为ServerSocketChannel即可。(下面同理)]*/ServerSocketChannel serverChannel = (ServerSocketChannel) key.channel();//获取socketChannel连接SocketChannel socketChannel = serverChannel.accept();//要注册到Selector那么channel必须设置为非阻塞socketChannel.configureBlocking(false);//将socketChannel连接也注册到Selector上,关注的是读Read事件socketChannel.register(selector, SelectionKey.OP_READ);} else if (key.isReadable()) { //处理读事件               try {SocketChannel channel = (SocketChannel) key.channel();ByteBuffer buffer = ByteBuffer.allocate(12);/* 客户端正常断开返回-1,此时需要取消Selector监视的Channel* 正常读取返回读到的字节数 ,并进行数据处理    * 异常断开走下面 的catch块*/int read = channel.read(buffer);//正常断开if(read == -1) {channel.close();key.cancel();}else{ //正常处理数据的流程// ...}} catch (Exception e) {/** 客户端异常断开,服务器会抛出异常,* 必须处理,即将此channel从keys集合中删除      */key.cancel(); channel.close();}}}}}
}

2.2处理消息的边界

核心:为每个Channel设置一个专有的附件,在注册时传入。

  • 一种思路是固定消息长度,数据包大小一样,服务器按预定长度读取,缺点是浪费带宽
  • 另一种思路是按分隔符拆分,缺点是效率低
  • TLV 格式,即 Type 类型、Length 长度、Value 数据,类型和长度已知的情况下,就可以方便获取消息大小,分配合适的 buffer,缺点是 buffer 需要提前分配,如果内容过大,则影响 server 吞吐量
    • Http 1.1 是 TLV 格式
    • Http 2.0 是 LTV 格式
#mermaid-svg-IJcmF1Taa4Z0yC0e {font-family:"trebuchet ms",verdana,arial,sans-serif;font-size:16px;fill:#333;}#mermaid-svg-IJcmF1Taa4Z0yC0e .error-icon{fill:#552222;}#mermaid-svg-IJcmF1Taa4Z0yC0e .error-text{fill:#552222;stroke:#552222;}#mermaid-svg-IJcmF1Taa4Z0yC0e .edge-thickness-normal{stroke-width:2px;}#mermaid-svg-IJcmF1Taa4Z0yC0e .edge-thickness-thick{stroke-width:3.5px;}#mermaid-svg-IJcmF1Taa4Z0yC0e .edge-pattern-solid{stroke-dasharray:0;}#mermaid-svg-IJcmF1Taa4Z0yC0e .edge-pattern-dashed{stroke-dasharray:3;}#mermaid-svg-IJcmF1Taa4Z0yC0e .edge-pattern-dotted{stroke-dasharray:2;}#mermaid-svg-IJcmF1Taa4Z0yC0e .marker{fill:#333333;stroke:#333333;}#mermaid-svg-IJcmF1Taa4Z0yC0e .marker.cross{stroke:#333333;}#mermaid-svg-IJcmF1Taa4Z0yC0e svg{font-family:"trebuchet ms",verdana,arial,sans-serif;font-size:16px;}#mermaid-svg-IJcmF1Taa4Z0yC0e .actor{stroke:hsl(259.6261682243, 59.7765363128%, 87.9019607843%);fill:#ECECFF;}#mermaid-svg-IJcmF1Taa4Z0yC0e text.actor>tspan{fill:black;stroke:none;}#mermaid-svg-IJcmF1Taa4Z0yC0e .actor-line{stroke:grey;}#mermaid-svg-IJcmF1Taa4Z0yC0e .messageLine0{stroke-width:1.5;stroke-dasharray:none;stroke:#333;}#mermaid-svg-IJcmF1Taa4Z0yC0e .messageLine1{stroke-width:1.5;stroke-dasharray:2,2;stroke:#333;}#mermaid-svg-IJcmF1Taa4Z0yC0e #arrowhead path{fill:#333;stroke:#333;}#mermaid-svg-IJcmF1Taa4Z0yC0e .sequenceNumber{fill:white;}#mermaid-svg-IJcmF1Taa4Z0yC0e #sequencenumber{fill:#333;}#mermaid-svg-IJcmF1Taa4Z0yC0e #crosshead path{fill:#333;stroke:#333;}#mermaid-svg-IJcmF1Taa4Z0yC0e .messageText{fill:#333;stroke:#333;}#mermaid-svg-IJcmF1Taa4Z0yC0e .labelBox{stroke:hsl(259.6261682243, 59.7765363128%, 87.9019607843%);fill:#ECECFF;}#mermaid-svg-IJcmF1Taa4Z0yC0e .labelText,#mermaid-svg-IJcmF1Taa4Z0yC0e .labelText>tspan{fill:black;stroke:none;}#mermaid-svg-IJcmF1Taa4Z0yC0e .loopText,#mermaid-svg-IJcmF1Taa4Z0yC0e .loopText>tspan{fill:black;stroke:none;}#mermaid-svg-IJcmF1Taa4Z0yC0e .loopLine{stroke-width:2px;stroke-dasharray:2,2;stroke:hsl(259.6261682243, 59.7765363128%, 87.9019607843%);fill:hsl(259.6261682243, 59.7765363128%, 87.9019607843%);}#mermaid-svg-IJcmF1Taa4Z0yC0e .note{stroke:#aaaa33;fill:#fff5ad;}#mermaid-svg-IJcmF1Taa4Z0yC0e .noteText,#mermaid-svg-IJcmF1Taa4Z0yC0e .noteText>tspan{fill:black;stroke:none;}#mermaid-svg-IJcmF1Taa4Z0yC0e .activation0{fill:#f4f4f4;stroke:#666;}#mermaid-svg-IJcmF1Taa4Z0yC0e .activation1{fill:#f4f4f4;stroke:#666;}#mermaid-svg-IJcmF1Taa4Z0yC0e .activation2{fill:#f4f4f4;stroke:#666;}#mermaid-svg-IJcmF1Taa4Z0yC0e .actorPopupMenu{position:absolute;}#mermaid-svg-IJcmF1Taa4Z0yC0e .actorPopupMenuPanel{position:absolute;fill:#ECECFF;box-shadow:0px 8px 16px 0px rgba(0,0,0,0.2);filter:drop-shadow(3px 5px 2px rgb(0 0 0 / 0.4));}#mermaid-svg-IJcmF1Taa4Z0yC0e .actor-man line{stroke:hsl(259.6261682243, 59.7765363128%, 87.9019607843%);fill:#ECECFF;}#mermaid-svg-IJcmF1Taa4Z0yC0e .actor-man circle,#mermaid-svg-IJcmF1Taa4Z0yC0e line{stroke:hsl(259.6261682243, 59.7765363128%, 87.9019607843%);fill:#ECECFF;stroke-width:2px;}#mermaid-svg-IJcmF1Taa4Z0yC0e :root{--mermaid-font-family:"trebuchet ms",verdana,arial,sans-serif;}客户端1服务器ByteBuffer1ByteBuffer2发送 01234567890abcdef3333\r第一次 read 存入 01234567890abcdef扩容拷贝 01234567890abcdef第二次 read 存入 3333\r01234567890abcdef3333\r客户端1服务器ByteBuffer1ByteBuffer2

服务器端

private static void split(ByteBuffer source) {source.flip();for (int i = 0; i < source.limit(); i++) {// 找到一条完整消息if (source.get(i) == '\n') {int length = i + 1 - source.position();// 把这条完整消息存入新的 ByteBufferByteBuffer target = ByteBuffer.allocate(length);// 从 source 读,向 target 写for (int j = 0; j < length; j++) {target.put(source.get());}debugAll(target);}}source.compact(); // 0123456789abcdef  position 16 limit 16
}public static void main(String[] args) throws IOException {Selector selector = Selector.open();ServerSocketChannel ssc = ServerSocketChannel.open();ssc.configureBlocking(false);// 2. 建立 selector 和 channel 的联系(注册)// SelectionKey 就是将来事件发生后,通过它可以知道事件和哪个channel的事件SelectionKey sscKey = ssc.register(selector, SelectionKey.OP_ACCEPT, null);// key 只关注 accept 事件sscKey.interestOps(SelectionKey.OP_ACCEPT);ssc.bind(new InetSocketAddress(8080));while (true) {selector.select();Iterator<SelectionKey> iter = selector.selectedKeys().iterator(); // accept, readwhile (iter.hasNext()) {SelectionKey key = iter.next();// 处理key 时,要从 selectedKeys 集合中删除,否则下次处理就会有问题iter.remove();// 5. 区分事件类型if (key.isAcceptable()) { // 如果是 acceptServerSocketChannel channel = (ServerSocketChannel) key.channel();SocketChannel sc = channel.accept();sc.configureBlocking(false);ByteBuffer buffer = ByteBuffer.allocate(16); // attachment// 将一个 byteBuffer 作为附件关联到 selectionKey 上sc.register(selector, SelecteKey.OP_READ, buffer);} else if (key.isReadable()) { // 如果是 readtry {SocketChannel channel = (SocketChannel) key.channel();// 获取 selectionKey 上关联的附件ByteBuffer buffer = (ByteBuffer) key.attachment();// 如果是正常断开,read 的方法的返回值是 -1int read = channel.read(buffer); if(read == -1) {key.cancel();} else {//调用split()进行处理数据split(buffer);// 需要扩容if (buffer.position() == buffer.limit()) {ByteBuffer newBuffer = ByteBuffer.allocate(buffer.capacity() * 2);buffer.flip();//将原buffer的内容写入新的buffernewBuffer.put(buffer); //将扩容后的BUffer设置为新的附件key.attach(newBuffer);}}} catch (IOException e) {e.printStackTrace();key.cancel();  // 因为客户端断开了,因此需要将 key 取消(从 selector 的 keys 集合中真正删除 key)}}}}
}

ByteBuffer 大小分配

  • 每个 channel 都需要记录可能被切分的消息,因为 ByteBuffer不能被多个 channel 共同使用,因此需要为每个 channel 维护一个独立的 ByteBuffer(附件)
  • ByteBuffer 不能太大,比如一个 ByteBuffer 1Mb 的话,要支持百万连接就要 1Tb 内存,因此需要设计大小可变的 ByteBuffer
    • 一种思路是首先分配一个较小的 buffer,例如 4k,如果发现数据不够,再分配 8k 的 buffer,将 4k buffer 内容拷贝至 8k buffer,优点是消息连续容易处理,缺点是数据拷贝耗费性能,参考实现
    • 另一种思路是用多个数组组成 buffer,一个数组不够,把多出来的内容写入新的数组,与前面的区别是消息存储不连续解析复杂,优点是避免了拷贝引起的性能损耗

2.3处理write()事件

一次无法写完例子

  • 非阻塞模式下,无法保证把 buffer 中所有数据都写入 channel,因此需要追踪 write 方法的返回值(代表实际写入字节数)
  • 用 selector 监听所有 channel 的可写事件,每个 channel 都需要一个 key 来跟踪 buffer,但这样又会导致占用内存过多,就有两阶段策略
    • 当消息处理器第一次写入消息时,才将 channel 注册到 selector 上
    • selector 检查 channel 上的可写事件,如果所有的数据写完了,就取消 channel 的注册
    • 如果不取消,会每次可写均会触发 write 事件
public class WriteServer {public static void main(String[] args) throws IOException {ServerSocketChannel ssc = ServerSocketChannel.open();ssc.configureBlocking(false);ssc.bind(new InetSocketAddress(8080));Selector selector = Selector.open();ssc.register(selector, SelectionKey.OP_ACCEPT);while(true) {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, SelectionKey.OP_READ);// 1. 向客户端发送内容StringBuilder sb = new StringBuilder();for (int i = 0; i < 3000000; i++) {sb.append("a");}ByteBuffer buffer = Charset.defaultCharset().encode(sb.toString());int write = sc.write(buffer);// 3. write 表示实际写了多少字节System.out.println("实际写入字节:" + write);// 4. 如果有剩余未读字节,才需要关注写事件if (buffer.hasRemaining()) {// read 1  write 4// 在原有关注事件的基础上,多关注 写事件sckey.interestOps(sckey.interestOps() + SelectionKey.OP_WRITE);// 把 buffer 作为附件加入 sckeysckey.attach(buffer);}} else if (key.isWritable()) {ByteBuffer buffer = (ByteBuffer) key.attachment();SocketChannel sc = (SocketChannel) key.channel();int write = sc.write(buffer);System.out.println("实际写入字节:" + write);if (!buffer.hasRemaining()) { // 写完了key.interestOps(key.interestOps() - SelectionKey.OP_WRITE);key.attach(null);}}}}}
}

客户端

public class WriteClient {public static void main(String[] args) throws IOException {Selector selector = Selector.open();SocketChannel sc = SocketChannel.open();sc.configureBlocking(false);sc.register(selector, SelectionKey.OP_CONNECT | SelectionKey.OP_READ);sc.connect(new InetSocketAddress("localhost", 8080));int count = 0;while (true) {selector.select();Iterator<SelectionKey> iter = selector.selectedKeys().iterator();while (iter.hasNext()) {SelectionKey key = iter.next();iter.remove();if (key.isConnectable()) {System.out.println(sc.finishConnect());} else if (key.isReadable()) {ByteBuffer buffer = ByteBuffer.allocate(1024 * 1024);count += sc.read(buffer);buffer.clear();System.out.println(count);}}}}
}

write 为何要取消

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

2.4多线程优化

码云地址

现在都是多核 cpu,设计时要充分考虑别让 cpu 的力量被白白浪费

前面的代码只有一个选择器,没有充分利用多核 cpu,如何改进呢?

分两组选择器

  • 单线程配一个选择器,专门处理 accept 事件
  • 创建 cpu 核心数的线程,每个线程配一个选择器,轮流处理 read 事件

总流程:

主线程只监听accept,有连接到来,获取SocketChannel,将其注册到专门的read()线程中(配置一个Selector),以后这个Channel的读事件就有read()线程专门监听处理。(可以配置多个read()线程,轮流注册并处理)

2.5总结

阻塞

  • 阻塞模式下,相关方法都会导致线程暂停

    • ServerSocketChannel.accept 会在没有连接建立时让线程暂停
    • SocketChannel.read 会在没有数据可读时让线程暂停
    • 阻塞的表现其实就是线程暂停了,暂停期间不会占用 cpu,但线程相当于闲置
  • 单线程下,阻塞方法之间相互影响,几乎不能正常工作,需要多线程支持

  • 但多线程下,有新的问题,体现在以下方面

    • 32 位 jvm 一个线程 320k,64 位 jvm 一个线程 1024k,如果连接数过多,必然导致 OOM,并且线程太多,反而会因为频繁上下文切换导致性能降低
    • 可以采用线程池技术来减少线程数和线程上下文切换,但治标不治本,如果有很多连接建立,但长时间 inactive,会阻塞线程池中所有线程,因此不适合长连接,只适合短连接

非阻塞

  • 非阻塞模式下,相关方法都会不会让线程暂停

    • 在 ServerSocketChannel.accept 在没有连接建立时,会返回 null,继续运行
    • SocketChannel.read 在没有数据可读时,会返回 0,但线程不必阻塞,可以去执行其它 SocketChannel 的 read 或是去执行 ServerSocketChannel.accept
    • 写数据时,线程只是等待数据写入 Channel 即可,无需等 Channel 通过网络把数据发送出去
  • 但非阻塞模式下,即使没有连接建立,和可读数据,线程仍然在不断运行,白白浪费了 cpu
  • 数据复制过程中,线程实际还是阻塞的(AIO 改进的地方)

多路复用

单线程可以配合 Selector 完成对多个 Channel 可读写事件的监控,这称之为多路复用

  • 多路复用仅针对网络 IO、普通文件 IO 没法利用多路复用(因为文件IO没有非阻塞API)
  • 如果不用 Selector 的非阻塞模式,线程大部分时间都在做无用功,而 Selector 能够保证
    • 有可连接事件时才去连接
    • 有可读事件才去读取
    • 有可写事件才去写入
      • 限于网络传输能力,Channel 未必时时可写,一旦 Channel 可写,会触发 Selector 的可写事件

好处

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

3.NIO vs BIO

3.1 stream vs channel

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

3.2 IO 模型

同步阻塞、同步非阻塞、同步多路复用、异步阻塞(没有此情况)、异步非阻塞

注意,同步还是非同步最大的区别在于

  • 同步:线程自己去获取结果(一个线程)
  • 异步:线程自己不去获取结果,而是由其它线程送结果(至少两个线程)

当调用一次 channel.read 或 stream.read 后,会切换至操作系统内核态来完成真正数据读取,而读取又分为两个阶段,分别为:

  • 等待数据阶段(等待网卡中的数据到达内核缓冲区)

  • 复制数据阶段(将内核缓冲区的数据拷贝到用户缓冲区)

  • 阻塞 IO

    用户线程等待数据需要阻塞,一直阻塞到有数据到来

  • 非阻塞 IO

    用户在等待数据时不需要阻塞,没有数据可读直接返回,然后一直循环,等到数据到来时进行读取。

  • 多路复用

    单线程可以配合 Selector完成对多个连接的事件的监控

信号驱动

  • 异步IO

    用户线程(例如下面的thread1)只是调用一下非阻塞的read()告诉操作系统我要读数据了,不管有没有数据都直接返回,然后操作系统负责另开线程(thread2)将数据最终拷贝到用户的缓冲区。

  • 阻塞 IO vs 多路复用

Netty从使用到源码3_NIO网络编程详解相关推荐

  1. Xposed源码剖析——app_process作用详解

    Xposed源码剖析--app_process作用详解 首先吐槽一下CSDN的改版吧,发表这篇文章之前其实我已经将此篇文章写过了两三次了.就是发表不成功.而且CSDN将我的文章草稿也一带>删除掉 ...

  2. mysql data文件夹恢复_【专注】Zabbix源码安装教程—步骤详解(2)安装并配置mysql...

    四.安装并配置mysql(1) 解压mysql-5.7.26.tar.gz与boost_1_59_0.tar.gz #tar -xvf mysql-5.7.26.tar.gz #tar -xvf bo ...

  3. php+mysql案例含源码_【专注】Zabbix源码安装教程—步骤详解(1)安装前准备

    一.实验环境准备 Rhel 7.6 x86_64(server) 192.168.163.72 Rhel 6.5 x86_64(agent) 192.168.163.61 均已配置操作安装光盘为YUM ...

  4. React 源码系列 | React Context 详解

    目前来看 Context 是一个非常强大但是很多时候不会直接使用的 api.大多数项目不会直接使用 createContext 然后向下面传递数据,而是采用第三方库(react-redux). 想想项 ...

  5. python编程入门与案例详解-Pythony运维入门之Socket网络编程详解

    Socket是什么? Socket 是电脑网络中进程间数据流的端点Socket 是操作系统的通信机制应用程序通过Socket进行网络数据的传输 首先,简单了解一下TCP通信过程: TCP三次握手(面试 ...

  6. Linux网络编程---详解TCP

    Linux网络编程---详解TCP的三次握手和四次挥手_shanghx_123的博客-CSDN博客_tcp的协议数据单元被称为 TCP协议详解(TCP报文.三次握手.四次挥手.TIME_WAIT状态. ...

  7. dockerfile源码安装mysql_docker容器详解五: dockerfile实现tomcat环境以及源码安装mysql...

    tomcat 上一节讲到了dockerfile的基础,这一次咱们来作一个小的练习 首先要了解tomcat安装的整个过程 首先搭建 jdk环境: 下载jdk包,解压以后添加环境变量 而后搭建tomcat ...

  8. android 网络编程实现,Android开发使用HttpURLConnection进行网络编程详解【附源码下载】...

    本文实例讲述了Android开发使用HttpURLConnection进行网络编程.分享给大家供大家参考,具体如下: --HttpURLConnection URLConnection已经可以非常方便 ...

  9. tcpip网络编程 尹圣雨源码_网络编程——学习笔记

    一.计算机网络 1.简介 计算机网络,是把分布在不同地理区域的计算机与专门的外部设备用通信线路互连成一个规模大.功能强的网络系统,从而使众多计算机可以方便的相互传递信息,共享硬件.软件.数据等资源.计 ...

最新文章

  1. S4 HANA物料凭证表MATDOC
  2. 每瓶啤酒2元,2个空酒瓶或4个瓶盖可换1瓶啤酒。10元最多可喝多少瓶啤酒?
  3. C# json解析字符串总是多出双引号_在JavaScript应用中将CSV转换为JSON
  4. 算法 msrcr_一种快速简便优秀的全局曲线调整与局部信息想结合的非线性彩色增强算法(多图深度分析和探索)...
  5. apache2.2配置
  6. BugkuCTF-Misc:Linux
  7. java重命名sheet失败_java jxl excel 数据导出 重新命名无效的工作表名称 | 学步园...
  8. C#调用Server_SQL
  9. 一个收入很低的人,能装大款到什么地步?
  10. 如果不使用 SQL Mail,如何在 SQL Server 中发送电子邮件
  11. xkcd目录_12条展示AI真相的XKCD片段
  12. Pyqt Model/view框架 3.可编辑项
  13. endnote初始化数据库支持_考前复习必备MySQL数据库(关系型数据库管理系统)
  14. Origin2021安装过程中的系列问题
  15. 学习电商的第一个业务,发票管理,发票作废(发票作废为重)
  16. 机器人C++库(6)Robotics Library 例程三之正逆运动学计算
  17. 人生如常,常亦无常,亦无进,茫然终身,谓常
  18. 使用fsck修复文件系统错误
  19. 佛山科目二仙塘考场(B场)-考试要点
  20. sm羞耻任务_羞耻驱动的发展

热门文章

  1. 单片机8位数码管时钟c语言汇编,基于8051单片机和DS12C8878位数码管显示的时钟.docx...
  2. NGB基本知识(一)
  3. 传感器技术实验及实训QY-812G
  4. C++隐藏任务栏和桌面
  5. Matlab GUI编程技巧(七):matlablistbox操作-列表框(ListBox)和uilistbox常用操作
  6. 一张图学会Python
  7. Presto RBO之谓词下推
  8. 计算机常用电脑 英语 缩写学习
  9. JavaScript的3种书写格式详细介绍
  10. 毕设模板-目录的更新与格式设置