NIO-网络编程-阿里云开发者社区

网络编程

阻塞

阻塞模式下，相关方法都会导致线程暂停
- ServerSocketChannel.accept 会在没有连接建立时让线程暂停
- SocketChannel.read 会在通道中没有数据可读时让线程暂停
- 阻塞的表现其实就是线程暂停了，暂停期间不会占用 cpu，但线程相当于闲置
单线程下，阻塞方法之间相互影响，几乎不能正常工作，需要多线程支持
但多线程下，有新的问题，体现在以下方面
- 32 位 jvm 一个线程 320k，64 位 jvm 一个线程 1024k，如果连接数过多，必然导致 OOM，并且线程太多，反而会因为频繁上下文切换导致性能降低
- 可以采用线程池技术来减少线程数和线程上下文切换，但治标不治本，如果有很多连接建立，但长时间 inactive，会阻塞线程池中所有线程，因此不适合长连接，只适合短连接

多线程版服务器端代码

@Slf4j
public class Server {
    public static void main(String[] args) throws IOException {
        ServerSocketChannel channel=ServerSocketChannel.open();
        channel.bind(new InetSocketAddress(8888));
        ByteBuffer buffer = ByteBuffer.allocate(20);
        while (true){
            log.info("connecting ...");
            SocketChannel newSocketChannel = channel.accept();//建立与客户端的连接
            log.info("connected... {}",newSocketChannel);
            new Thread(()->{
                try {
                    log.info("启动会话线程:{}",Thread.currentThread().getName());
                    readContext(newSocketChannel,buffer);
                } catch (IOException e) {
                    e.printStackTrace();
                }
            }).start();
        }
    }

    public static void readContext(SocketChannel channel,ByteBuffer buffer) throws IOException {
        log.debug("before read...");
        channel.read(buffer);
        buffer.flip();//切换为读模式
        debugAll(buffer);
        buffer.clear();//切换为写模式
    }
}

客户端

@Slf4j
public class Client {
    public static void main(String[] args) throws IOException {
        SocketChannel sc = SocketChannel.open();
        sc.connect(new InetSocketAddress("localhost",8888));
        ByteBuffer buffer = StandardCharsets.UTF_8.encode("hello world");
        log.info("waiting...");
        sc.write(buffer);
    }
}

非阻塞

可以通过ServerSocketChannel的configureBlocking(false)方法将获得连接设置为非阻塞的。此时若没有连接，accept会返回null
可以通过SocketChannel的configureBlocking(false)方法将从通道中读取数据设置为非阻塞的。若此时通道中没有数据可读，read会返回-1

服务端代码

@Slf4j
public class NIOServer {
    public static void main(String[] args) throws IOException {
        ServerSocketChannel serverSocketChannel = ServerSocketChannel.open();
        serverSocketChannel.bind(new InetSocketAddress(8888));
        serverSocketChannel.configureBlocking(false);//设置为非阻塞模式
        List<SocketChannel> channels = new ArrayList<>();
        ByteBuffer buffer = ByteBuffer.allocate(24);
        log.info("listen ...");
        while (true) {
            SocketChannel socketChannel = serverSocketChannel.accept();//以非阻塞模式方式运行，如果没有连接则返回null
            if (socketChannel != null) {
                log.info("accept client connect:{}", socketChannel);
                socketChannel.configureBlocking(false);//设置为非阻塞模式
                channels.add(socketChannel);
            }
            for (SocketChannel channel : channels) {
                int readSize = channel.read(buffer);//read方法处于非阻塞模式,未读取到数据时返回0
                if (readSize > 0) {
                    buffer.flip();//转为读模式
                    debugAll(buffer);
                    buffer.clear();//转为写模式
                }
            }
        }
    }
}

这样写存在一个问题，因为设置为了非阻塞，会一直执行while(true)中的代码，CPU一直处于忙碌状态，会使得性能变低，所以实际情况中不使用这种方法处理请求

Selector

多路复用

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

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

步骤解析

获得选择器Selector
```
Selector selector = Selector.open();
```
将通道设置为非阻塞模式，并注册到选择器中，并设置感兴趣的事件
- channel 必须工作在非阻塞模式
- FileChannel 没有非阻塞模式，因此不能配合 selector 一起使用
- 绑定的事件类型可以有
  - connect - 客户端连接成功时触发
  - accept - 服务器端成功接受连接时触发
  - read - 数据可读入时触发，有因为接收能力弱，数据暂不能读入的情况
  - write - 数据可写出时触发，有因为发送能力弱，数据暂不能写出的情况
```
// 通道必须设置为非阻塞模式
server.configureBlocking(false);
// 将通道注册到选择器中，并设置关注的事件
server.register(selector, SelectionKey.OP_ACCEPT);
```
通过Selector监听事件，并获得就绪的通道个数，若没有通道就绪，线程会被阻塞
- 阻塞直到绑定事件发生
```
int count = selector.select();
```
- 阻塞直到绑定事件发生，或是超时（时间单位为 ms）
```
int count = selector.select(long timeout);
```
- 不会阻塞，也就是不管有没有事件，立刻返回，自己根据返回值检查是否有事件
```
int count = selector.selectNow();
```

获取就绪事件并得到对应的通道，然后进行处理

// 获取所有事件
Set<SelectionKey> selectionKeys = selector.selectedKeys();                
// 使用迭代器遍历事件
Iterator<SelectionKey> iterator = selectionKeys.iterator();
while (iterator.hasNext()) {
    SelectionKey key = iterator.next();          
    // 判断key的类型，此处为Accept类型
    if(key.isAcceptable()) {
        // 获得key对应的channel
        ServerSocketChannel channel = (ServerSocketChannel) key.channel();
        // 获取连接并处理，而且是必须处理，否则需要取消
        SocketChannel socketChannel = channel.accept();
        // 处理完毕后移除
        iterator.remove();
    }
}

事件发生后能否不处理

事件发生后，要么处理，要么取消（cancel），不能什么都不做，否则下次该事件仍会触发，这是因为 nio 底层使用的是水平触发

Read事件

在Accept事件中，若有客户端与服务器端建立了连接，需要将其对应的SocketChannel设置为非阻塞，并注册到选择其中

添加Read事件，触发后进行读取操作

@Slf4j
public class NIOServer2 {
    public static void main(String[] args) throws IOException {
        //创建selector
        Selector selector = Selector.open();
        ServerSocketChannel serverSocketChannel = ServerSocketChannel.open();
        serverSocketChannel.configureBlocking(false);
        //注册socket到selecor
        SelectionKey ssckey = serverSocketChannel.register(selector, 0, null);//设置事件类型为空（0）
        ssckey.interestOps(SelectionKey.OP_ACCEPT);//关注accept事件
        serverSocketChannel.bind(new InetSocketAddress(8888));
        log.info("server started ...");
        while (true){
            selector.select();//select方法，当有事件时线程恢复运行，没有事件时线程阻塞
            //获取事件集合
            Iterator<SelectionKey> keys = selector.selectedKeys().iterator();
            while (keys.hasNext()){
                SelectionKey key = keys.next();
                //获取accept事件
                if (key.isAcceptable()){
                    ServerSocketChannel channel =(ServerSocketChannel) key.channel();
                    SocketChannel clientSocket = channel.accept();//获取连接并处理，而且是必须处理，否则需要取消
                    log.info("create connect:{}",clientSocket);
                    clientSocket.configureBlocking(false);
                    //注册会话channel
                    SelectionKey sk = clientSocket.register(selector, 0, null);
                    sk.interestOps(SelectionKey.OP_READ);//监听读事件
                 //处理读事件
                }else if (key.isReadable()){
                    SocketChannel channel = (SocketChannel) key.channel();
                    ByteBuffer buffer = ByteBuffer.allocate(32);
                    while (true){
                        int length = channel.read(buffer);
                        buffer.flip();//切换为读模式
                        debugAll(buffer);
                        if (length<buffer.capacity()){
                            break;
                        }
                        buffer.clear();//清空buffer内容并改为写模式
                    }
                    buffer.flip();//切换到读模式
                }
                //移除事件，事件不会主动从Set中移除，会造成死循环
                keys.remove();
            }
        }
    }
}

删除事件

当处理完一个事件后，一定要调用迭代器的remove方法移除对应事件，否则会出现错误

当调用了 server.register(selector, SelectionKey.OP_ACCEPT)后，Selector中维护了一个集合，用于存放SelectionKey以及其对应的通道

final class WindowsSelectorImpl extends SelectorImpl {
    private SelectionKeyImpl[] channelArray = new SelectionKeyImpl[8];
}

public class SelectionKeyImpl extends AbstractSelectionKey {
    // Key对应的通道
    final SelChImpl channel;
    ...
}

当选择器中的通道对应的事件发生后，selecionKey会被放到另一个集合中，但是selecionKey不会自动移除，所以需要我们在处理完一个事件后，通过迭代器手动移除其中的selecionKey。否则会导致已被处理过的事件再次被处理，就会引发错误

断开处理

当客户端与服务器之间的连接断开时，会给服务器端发送一个读事件，对异常断开和正常断开需要加以不同的方式进行处理

正常断开

正常断开时，服务器端的channel.read(buffer)方法的返回值为-1，所以当结束到返回值为-1时，需要调用key的cancel方法取消此事件，并在取消后移除该事件

int read = channel.read(buffer);
// 断开连接时，客户端会向服务器发送一个写事件，此时read的返回值为-1
if(read == -1) {
    // 取消该事件的处理,取消注册
    key.cancel();
    channel.close();
} else {
    ...
}
// 取消或者处理，都需要移除key
iterator.remove();

异常断开
- 异常断开时，会抛出IOException异常，在try-catch的catch块中捕获异常并调用key的cancel方法即可

消息边界

不处理消息边界存在的问题：将缓冲区的大小设置为4个字节，发送2个汉字（你好），通过decode解码并打印时，会出现乱码

ByteBuffer buffer = ByteBuffer.allocate(4);
// 解码并打印
System.out.println(StandardCharsets.UTF_8.decode(buffer));

你�
��

这是因为UTF-8字符集下，1个汉字占用3个字节，此时缓冲区大小为4个字节，一次读时间无法处理完通道中的所有数据，所以一共会触发两次读事件。这就导致 你好 的 好 字被拆分为了前半部分和后半部分发送，解码时就会出现问题

处理消息边界

传输的文本可能有以下三种情况

文本大于缓冲区大小
- 此时需要将缓冲区进行扩容
发生半包现象
发生粘包现象

解决思路大致有以下三种

固定消息长度，数据包大小一样，服务器按预定长度读取，当发送的数据较少时，需要将数据进行填充，直到长度与消息规定长度一致。缺点是浪费带宽
另一种思路是按分隔符拆分，缺点是效率低，需要一个一个字符地去匹配分隔符
TLV 格式，即 Type 类型、Length 长度、Value 数据（也就是在消息开头用一些空间存放后面数据的长度），如HTTP请求头中的Content-Type与Content-Length。类型和长度已知的情况下，就可以方便获取消息大小，分配合适的 buffer，缺点是 buffer 需要提前分配，如果内容过大，则影响 server 吞吐量
- Http 1.1 是 TLV 格式
- Http 2.0 是 LTV 格式

下文的消息边界处理方式为第二种：按分隔符拆分

附件与扩容

Channel的register方法还有第三个参数：附件，可以向其中放入一个Object类型的对象，该对象会与登记的Channel以及其对应的SelectionKey绑定，可以从SelectionKey获取到对应通道的附件

public final SelectionKey register(Selector sel, int ops, Object att)

可通过SelectionKey的attachment()方法获得附件

ByteBuffer buffer = (ByteBuffer) key.attachment();

我们需要在Accept事件发生后，将通道注册到Selector中时，对每个通道添加一个ByteBuffer附件，让每个通道发生读事件时都使用自己的通道，避免与其他通道发生冲突而导致问题

// 设置为非阻塞模式，同时将连接的通道也注册到选择其中，同时设置附件
socketChannel.configureBlocking(false);
ByteBuffer buffer = ByteBuffer.allocate(16);
// 添加通道对应的Buffer附件
socketChannel.register(selector, SelectionKey.OP_READ, buffer);

当Channel中的数据大于缓冲区时，需要对缓冲区进行扩容操作。此代码中的扩容的判定方法：Channel调用compact方法后，的position与limit相等，说明缓冲区中的数据并未被读取（容量太小），此时创建新的缓冲区，其大小扩大为两倍。同时还要将旧缓冲区中的数据拷贝到新的缓冲区中，同时调用SelectionKey的attach方法将新的缓冲区作为新的附件放入SelectionKey中

// 如果缓冲区太小，就进行扩容
if (buffer.position() == buffer.limit()) {
    ByteBuffer newBuffer = ByteBuffer.allocate(buffer.capacity()*2);
    // 将旧buffer中的内容放入新的buffer中
    newBuffer.put(buffer);
    // 将新buffer作为附件放到key中
    key.attach(newBuffer);
}

ByteBuffer的大小分配

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

Write事件

服务器通过Buffer向通道中写入数据时，可能因为通道容量小于Buffer中的数据大小，导致无法一次性将Buffer中的数据全部写入到Channel中，这时便需要分多次写入，具体步骤如下

执行一次写操作，将buffer中的内容写入到SocketChannel中，然后判断Buffer中是否还有数据
若Buffer中还有数据，则需要将SockerChannel注册到Seletor中，并关注写事件，同时将未写完的Buffer作为附件一起放入到SelectionKey中

int write = socket.write(buffer);
// 通道中可能无法放入缓冲区中的所有数据
if (buffer.hasRemaining()) {
    // 注册到Selector中，关注可写事件，并将buffer添加到key的附件中
    socket.configureBlocking(false);
    socket.register(selector, SelectionKey.OP_WRITE, buffer);
}

添加写事件的相关操作key.isWritable()，对Buffer再次进行写操作
- 每次写后需要判断Buffer中是否还有数据（是否写完）。若写完，需要移除SelecionKey中的Buffer附件，避免其占用过多内存，同时还需移除对写事件的关注

SocketChannel socket = (SocketChannel) key.channel();
// 获得buffer
ByteBuffer buffer = (ByteBuffer) key.attachment();
// 执行写操作
int write = socket.write(buffer);
System.out.println(write);
// 如果已经完成了写操作，需要移除key中的附件，同时不再对写事件感兴趣
if (!buffer.hasRemaining()) {
    key.attach(null);
    key.interestOps(0);
}

NIO-网络编程

网络编程

阻塞

多线程版服务器端代码

客户端

非阻塞

服务端代码

Selector

多路复用

步骤解析

事件发生后能否不处理

Read事件

删除事件

断开处理

消息边界

附件与扩容

ByteBuffer的大小分配

Write事件

热门文章

最新文章

相关课程

相关电子书

相关实验场景

热门

活动广场

任务中心

开发者评测

高校计划

乘风者计划

训练营

阿里云MVP

话题

直播

下载

镜像站

技术资料

插件

NIO-网络编程

网络编程

阻塞

多线程版服务器端代码

客户端

非阻塞

服务端代码

Selector

多路复用

步骤解析

事件发生后能否不处理

Read事件

删除事件

断开处理

消息边界

附件与扩容

ByteBuffer的大小分配

Write事件

热门文章

最新文章

相关课程

相关电子书

相关实验场景