NIO基础 - 网络编程

4. 网络编程

4.1 阻塞模式

单线程模式下，阻塞方法之间存在互相影响

ServerSocketChannel.accept() 方法会在没有连接建立时阻塞
SocketChannel.read() 在没有可读数据时阻塞

Sever

package com.sw.netty._04.block;

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 utils.ByteBufferUtil.debugAll;

@Slf4j
public class Server {
    public static void main(String[] args) throws IOException {
        ByteBuffer bf = ByteBuffer.allocate(16);
        ServerSocketChannel ssc = ServerSocketChannel.open();
        ssc.bind(new InetSocketAddress(8088));
        List<SocketChannel> channelList = new ArrayList<>();
        while (true) {
            log.info("connecting...");
            SocketChannel channel = ssc.accept();
            log.info("connected - [{}]", channel);
            channelList.add(channel);
            for (SocketChannel sc : channelList) {
                log.info("before read - [{}]", channel);
                sc.read(bf);
                bf.flip();
                debugAll(bf);
                bf.clear();
                log.info("after read - [{}]", channel);
            }
        }
    }
}

Client

package com.sw.netty._04.block;

import lombok.extern.slf4j.Slf4j;

import java.io.IOException;
import java.net.InetSocketAddress;
import java.nio.channels.SocketChannel;

@Slf4j
public class Client {
    public static void main(String[] args) throws IOException {
        SocketChannel sc = SocketChannel.open();
        sc.connect(new InetSocketAddress("localhost", 8088));
        log.info("waiting...");
    }
}

4.2 非阻塞模式

非阻塞模式下，相关方法的线程不会阻塞

ServerSocketChannel.accept() 方法会在没有连接建立时返回 null，继续运行
SocketChannel.read() 在没有可读数据时返回 0
写数据时，数据写入 Channel 后，线程即可继续运行，无需等待 Channel 通过网络把数据发送出去或发送完

但非阻塞模式下，即使没有新的连接建立、可读数据，线程仍在运行；且数据复的制过程线程是阻塞的

Sever

package com.sw.netty._04.nonBlock;

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 utils.ByteBufferUtil.debugAll;

@Slf4j
public class Server {
    public static void main(String[] args) throws IOException {
        ByteBuffer bf = ByteBuffer.allocate(16);
        ServerSocketChannel ssc = ServerSocketChannel.open();
        // 切换为非阻塞模式
        ssc.configureBlocking(false);
        ssc.bind(new InetSocketAddress(8088));
        List<SocketChannel> channelList = new ArrayList<>();
        while (true) {
            SocketChannel channel = ssc.accept();
            if (channel != null) {
                log.info("connected - [{}]", channel);
                // 切换为非阻塞模式
                channel.configureBlocking(false);
                channelList.add(channel);
            }
            for (SocketChannel sc : channelList) {
                if (sc.read(bf) > 0) {
                    bf.flip();
                    debugAll(bf);
                    bf.clear();
                    log.info("after read - [{}]", channel);
                }
            }
        }
    }
}

Client

package com.sw.netty._04.nonBlock;

import lombok.extern.slf4j.Slf4j;

import java.io.IOException;
import java.net.InetSocketAddress;
import java.nio.channels.SocketChannel;

@Slf4j
public class Client {
    public static void main(String[] args) throws IOException {
        SocketChannel sc = SocketChannel.open();
        sc.connect(new InetSocketAddress("localhost", 8088));
        log.info("waiting...");
    }
}

4.3 多路复用

概念：单线程配合 Selector 完成对多个 Channel 可读、可写事件的监听

仅针对网络 IO、文件 IO 操作无法使用多路复用
Selector 的作用：
- 有可连接事件时建立连接
- 有可读事件时执行读取操作
- 有可写事件时执行写入操作

注：Channel 不一定时时可写，当 Channel 可写时，则会触发 Selector 的可写事件

4.4 Selector

4.4Selector.png

与 Selector 协作的线程可以监听多个 Channel，当事件发生时才去处理对应的事件，此时的线程可被充分利用，同时也节约的线程的数量，减少了线程间的上下文切换

（1）创建

Selector selector = Selector.open();

（2）绑定 Channel 事件（注册）

Selector selector = Selector.open();
ServerSocketChannel ssc = ServerSocketChannel.open();
// 切换为非阻塞模式
ssc.configureBlocking(false);

// 注册 Channel
SelectionKey sscKey = ssc.register(selector, 0, null);

注：

Channel 必须以非阻塞模式运行
绑定的事件类型如下：
- accept：有连接请求时触发
- connection：（客户端）连接建立后触发
- read：读事件
- write：写事件

（3）监听 Channel 事件

// 方式一：阻塞直到绑定事件发生
int count = selector.select();

// 方式二：阻塞到超时时间（ms）或绑定事件发生
int count = selector.select(long timeout);

// 方式三：selector 立即返回，后续流程根据返回值检查是否有事件发生
int count = selector.selectNow();

（4）selector 何时不阻塞

发生对应事件时：
- accept 事件 - 客户端发起连接请求
- read 事件 - 客户端发送数据、正常/异常关闭（当客户端发送的数据过大，服务端无法一次处理完时，会触发多次读取事件）
- write 事件 - channel 当前状态可写出数据
调用 selector.wakeup()
调用 selector.close()
selector 所在的线程中断

4.5 处理 accept 事件

package com.sw.netty._04.selector;

import lombok.extern.slf4j.Slf4j;

import java.io.IOException;
import java.net.InetSocketAddress;
import java.nio.ByteBuffer;
import java.nio.channels.*;
import java.util.ArrayList;
import java.util.Iterator;
import java.util.List;

import static utils.ByteBufferUtil.debugAll;

@Slf4j
public class Server {
    public static void main(String[] args) throws IOException {
        // 1. Selector
        Selector selector = Selector.open();

        ByteBuffer bf = ByteBuffer.allocate(16);
        ServerSocketChannel ssc = ServerSocketChannel.open();
        // 切换为非阻塞模式
        ssc.configureBlocking(false);

        // 2. 注册 Channel
        /**
         * 通过 SelectionKey，可以知道发生的事件和发生事件的 Channel
         * 事件类型：
         * accept：有连接请求时触发
         * connection：（客户端）连接建立后触发
         * read：读事件
         * write：写事件
         */
        SelectionKey sscKey = ssc.register(selector, 0, null);
        // sscKey 只关注 accept 事件
        sscKey.interestOps(SelectionKey.OP_ACCEPT);
        log.info("register channel key-[{}]", sscKey);

        ssc.bind(new InetSocketAddress(8088));
        while (true) {
            // 3. select 方法，没有事件发生或事件取消时，线程阻塞，否则反之
            selector.select();

            // 4. 处理事件
            Iterator<SelectionKey> iterator = selector.selectedKeys().iterator();
            while (iterator.hasNext()) {
                SelectionKey key = iterator.next();
                // 处理完事件后，需要显式删除对应的key（事件处理完了，但 Selector 不会删除对应的key）
                iterator.remove();

                // 5. 区分事件类型
                if (key.isAcceptable()) {
                    log.info("Deal Accept Event");
                    ServerSocketChannel channel = (ServerSocketChannel) key.channel();
                    SocketChannel sc = channel.accept();
                    sc.configureBlocking(false);
                    SelectionKey scKey = sc.register(selector, 0, null);
                    scKey.interestOps(SelectionKey.OP_READ);
                }
            }
        }
    }
}

注：当事件发生后，要么处理，要么取消，如果什么都不做，下次该事件还会触发

4.6 处理 read 事件

（1）演示 Demo

package com.sw.netty._04.selector;

import lombok.extern.slf4j.Slf4j;

import java.io.IOException;
import java.net.InetSocketAddress;
import java.nio.ByteBuffer;
import java.nio.channels.*;
import java.util.ArrayList;
import java.util.Iterator;
import java.util.List;

import static utils.ByteBufferUtil.debugAll;

@Slf4j
public class Server {
    public static void main(String[] args) throws IOException {
        // 1. Selector
        Selector selector = Selector.open();

        ByteBuffer bf = ByteBuffer.allocate(16);
        ServerSocketChannel ssc = ServerSocketChannel.open();
        // 切换为非阻塞模式
        ssc.configureBlocking(false);

        // 2. 注册 Channel
        /**
         * 通过 SelectionKey，可以知道发生的事件和发生事件的 Channel
         * 事件类型：
         * accept：有连接请求时触发
         * connection：（客户端）连接建立后触发
         * read：读事件
         * write：写事件
         */
        SelectionKey sscKey = ssc.register(selector, 0, null);
        // sscKey 只关注 accept 事件
        sscKey.interestOps(SelectionKey.OP_ACCEPT);
        log.info("register channel key-[{}]", sscKey);

        ssc.bind(new InetSocketAddress(8088));
        while (true) {
            // 3. select 方法，没有事件发生或事件取消时，线程阻塞，否则反之
            selector.select();

            // 4. 处理事件
            Iterator<SelectionKey> iterator = selector.selectedKeys().iterator();
            while (iterator.hasNext()) {
                SelectionKey key = iterator.next();
                // 处理完事件后，需要显式删除对应的key（事件处理完了，但 Selector 不会删除对应的key）
                iterator.remove();

                // 5. 区分事件类型
                if (key.isAcceptable()) {
                    log.info("Deal Accept Event");
                    ServerSocketChannel channel = (ServerSocketChannel) key.channel();
                    SocketChannel sc = channel.accept();
                    sc.configureBlocking(false);
                    SelectionKey scKey = sc.register(selector, 0, null);
                    scKey.interestOps(SelectionKey.OP_READ);
                }

                if (key.isReadable()) {
                    try {
                        log.info("Deal Read Event");
                        SocketChannel channel = (SocketChannel) key.channel();
                        ByteBuffer readBf = ByteBuffer.allocate(16);
                        if (-1 != channel.read(readBf)) {
                            readBf.flip();
                            debugAll(readBf);
                        } else {
                            // 处理客户端读事件结束，正常断开
                            log.info("Client - [{}] close", key);
                            key.cancel();
                        }
                    } catch (IOException e) {
                        // 对发生异常的事件取消注册（从 Selector key 集合中移除）
                        // 如：客户端关闭主动断开连接
                        key.cancel();
                        e.printStackTrace();
                    }
                }
            }
        }
    }
}

（2）iterator.remove() 解释

当 selector 事件发生后，会将对应的 key 存入 selectedKeys 集合，但在事件处理完后并不会删除对应的 key，需要显式删除处理，如：第一次触发 accept 事件，处理完后，下一次循环再进来，key.isAcceptable() 判断为真，进入对应的判断后 channel.accept() 值为空（此时并不是 accept 事件），触发空指针异常

（3）cancel 的作用

取消注册在 selector 上的 channel，并从 SelectionKeys 集合中删除对应的事件 key

4.7 处理消息边界

固定消息长度（数据包大小一样），但是在一定程度上会浪费带宽
按分隔符拆分，效率低
TLV 格式，即：Type 类型、Length 长度、Value 数据，在消息类型和长度已知的情况下，就可以方便的获取消息的大小，以此分配 buffer；需要提前分配 buffer，如果内容过大，会对 server 的吞吐量造成影响
- HTTP 1.0 TLV 格式
- HTTP 2.0 LTV 格式

（1）此处以按分隔符拆分为例：

Server

package com.sw.netty._04.msgBoundary;

import lombok.extern.slf4j.Slf4j;

import java.io.IOException;
import java.net.InetSocketAddress;
import java.nio.ByteBuffer;
import java.nio.channels.SelectionKey;
import java.nio.channels.Selector;
import java.nio.channels.ServerSocketChannel;
import java.nio.channels.SocketChannel;
import java.util.Iterator;

import static utils.ByteBufferUtil.debugAll;

@Slf4j
public class Server {
    public static void main(String[] args) throws IOException {
        // 1. Selector
        Selector selector = Selector.open();

        ByteBuffer bf = ByteBuffer.allocate(16);
        ServerSocketChannel ssc = ServerSocketChannel.open();
        // 切换为非阻塞模式
        ssc.configureBlocking(false);

        // 2. 注册 Channel
        /**
         * 通过 SelectionKey，可以知道发生的事件和发生事件的 Channel
         * 事件类型：
         * accept：有连接请求时触发
         * connection：（客户端）连接建立后触发
         * read：读事件
         * write：写事件
         */
        SelectionKey sscKey = ssc.register(selector, 0, null);
        // sscKey 只关注 accept 事件
        sscKey.interestOps(SelectionKey.OP_ACCEPT);
        log.info("register channel key-[{}]", sscKey);

        ssc.bind(new InetSocketAddress(8088));
        while (true) {
            // 3. select 方法，没有事件发生或事件取消时，线程阻塞，否则反之
            selector.select();

            // 4. 处理事件
            Iterator<SelectionKey> iterator = selector.selectedKeys().iterator();
            while (iterator.hasNext()) {
                SelectionKey key = iterator.next();
                // 处理完事件后，需要显式删除对应的key（事件处理完了，但 Selector 不会删除对应的key）
                iterator.remove();

                // 5. 区分事件类型
                if (key.isAcceptable()) {
                    log.info("Deal Accept Event");
                    ServerSocketChannel channel = (ServerSocketChannel) key.channel();
                    SocketChannel sc = channel.accept();
                    sc.configureBlocking(false);

                    ByteBuffer readBf = ByteBuffer.allocate(16);
                    // 将 readBf 作为附件关联到事件 key 上
                    // 将 readBf 的生命周期提升到与 SelectionKey 平级
                    SelectionKey scKey = sc.register(selector, 0, readBf);
                    scKey.interestOps(SelectionKey.OP_READ);
                }

                if (key.isReadable()) {
                    try {
                        log.info("Deal Read Event");
                        SocketChannel channel = (SocketChannel) key.channel();
                        ByteBuffer readBf = (ByteBuffer) key.attachment();
                        if (-1 != channel.read(readBf)) {
                            split(readBf);

                            // 当传输的内容超过 readbf 初始容量时，需进行扩容
                            if (readBf.position() == readBf.limit()) {
                                ByteBuffer newReadBf = ByteBuffer.allocate(readBf.capacity() * 2);
                                readBf.flip();
                                newReadBf.put(readBf);
                                key.attach(newReadBf);
                            }
                        } else {
                            // 处理客户端读事件结束，正常断开
                            log.info("Client - [{}] close", key);
                            key.cancel();
                        }
                    } catch (IOException e) {
                        // 对发生异常的事件取消注册（从 Selector key 集合中移除）
                        // 如：客户端关闭主动断开连接
                        key.cancel();
                        e.printStackTrace();
                    }
                }
            }
        }
    }

    private static void split(ByteBuffer source) {
        source.flip();
        for (int i = 0; i < source.limit(); i++) {
            if ('\n' == source.get(i)) {
                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);
            }
        }

        // 此处不使用 clear，需使用 compact 将剩余未读的部分向前移动
        source.compact();
    }
}

Client

package com.sw.netty._04.msgBoundary;

import lombok.extern.slf4j.Slf4j;

import java.io.IOException;
import java.net.InetSocketAddress;
import java.nio.channels.SocketChannel;
import java.nio.charset.Charset;

@Slf4j
public class Client {
    public static void main(String[] args) throws IOException {
        SocketChannel sc = SocketChannel.open();
        sc.connect(new InetSocketAddress("localhost", 8088));
        sc.write(Charset.defaultCharset().encode("012\n0123456789abcdefyao\n"));
        sc.write(Charset.defaultCharset().encode("0123456789abcdefsunxiaochuan\nyaoshuige\n"));
        sc.close();
    }
}

4.8 ByteBuffer 大小分配

每个 Channel 都需要记录可能被切分的消息，因为 ByteBuffer 不能被多个 Channel 共享，需要独立维护
ByteBuffer 不能太大，需要可变：
- 思路一：先分配一个小的 buffer，不够的时候再进行扩容，优点是消息连续存储，缺点是数据拷贝存在一定的性能损耗
- 思路二：用数组维护 buffer，可以避免数据拷贝产生的性能损耗，但消息存储不连续

4.9 处理 write 事件

此处以写入内容过多问题（一次性发送数据）为例：

Server

package com.sw.netty._04.writeableEvents;

import java.io.IOException;
import java.net.InetSocketAddress;
import java.nio.ByteBuffer;
import java.nio.channels.SelectionKey;
import java.nio.channels.Selector;
import java.nio.channels.ServerSocketChannel;
import java.nio.channels.SocketChannel;
import java.nio.charset.Charset;
import java.util.Iterator;
import java.util.UUID;

public class Server {
    public static void main(String[] args) throws IOException {
        ServerSocketChannel ssc = ServerSocketChannel.open();
        ssc.configureBlocking(false);

        Selector selector = Selector.open();
        ssc.register(selector, SelectionKey.OP_ACCEPT);

        ssc.bind(new InetSocketAddress(8088));
        while (true) {
            selector.select();
            Iterator<SelectionKey> iterator = selector.selectedKeys().iterator();
            while (iterator.hasNext()) {
                SelectionKey key = iterator.next();
                iterator.remove();

                if (key.isAcceptable()) {
                    SocketChannel sc = ssc.accept();
                    sc.configureBlocking(false);

                    // 1. 向客户端发送大量数据
                    StringBuilder sb = new StringBuilder();
                    for (int i = 0; i < 999999; i++) {
                        sb.append(UUID.randomUUID().toString().replace("-", ""));
                    }

                    ByteBuffer bf = Charset.defaultCharset().encode(sb.toString());
                    while (bf.hasRemaining()) {
                        int write = sc.write(bf);
                        System.out.println("发送：" + write + " 字节数据");
                    }
                }
            }
        }
    }
}

Client

package com.sw.netty._04.writeableEvents;

import java.io.IOException;
import java.net.InetSocketAddress;
import java.nio.ByteBuffer;
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", 8088));

        int count = 0;
        while (true) {
            ByteBuffer bf = ByteBuffer.allocate(1024 * 1024);
            count += sc.read(bf);
            System.out.println("接收到：" + count + " 字节数据");
        }
    }
}

改进优化：

Server

package com.sw.netty._04.writeableEvents;

import java.io.IOException;
import java.net.InetSocketAddress;
import java.nio.ByteBuffer;
import java.nio.channels.SelectionKey;
import java.nio.channels.Selector;
import java.nio.channels.ServerSocketChannel;
import java.nio.channels.SocketChannel;
import java.nio.charset.Charset;
import java.util.Iterator;
import java.util.UUID;

public class Server {
    public static void main(String[] args) throws IOException {
        ServerSocketChannel ssc = ServerSocketChannel.open();
        ssc.configureBlocking(false);

        Selector selector = Selector.open();
        ssc.register(selector, SelectionKey.OP_ACCEPT);

        ssc.bind(new InetSocketAddress(8088));
        while (true) {
            selector.select();
            Iterator<SelectionKey> iterator = selector.selectedKeys().iterator();
            while (iterator.hasNext()) {
                SelectionKey key = iterator.next();
                iterator.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 < 999999; i++) {
                        sb.append(UUID.randomUUID().toString().replace("-", ""));
                    }

                    ByteBuffer bf = Charset.defaultCharset().encode(sb.toString());
                    int write = sc.write(bf);
                    System.out.println("OP_READ - 发送：" + write + " 字节数据");

                    // 2. 判断是否有剩余内容
                    if (bf.hasRemaining()) {
                        // 3. 关注可写事件（在原关注事件的基础上，需额外关注写事件）
                        scKey.interestOps(scKey.interestOps() + SelectionKey.OP_WRITE);
                        // scKey.interestOps(scKey.interestOps() | SelectionKey.OP_WRITE);
                        // 4. 将未写完的数据挂载到 scKey 上
                        scKey.attach(bf);
                    }
                }

                if (key.isWritable()) {
                    ByteBuffer bf = (ByteBuffer) key.attachment();
                    SocketChannel sc = (SocketChannel) key.channel();
                    int write = sc.write(bf);
                    System.out.println("OP_WRITE - 发送：" + write + " 字节数据");

                    // 5. 关闭资源
                    if (!bf.hasRemaining()) {
                        // 清除挂载的 buffer
                        key.attach(null);

                        // 清除关注的事件
                        key.interestOps(key.interestOps() - SelectionKey.OP_WRITE);
                    }
                }
            }
        }
    }
}

注：当 channel 发送数据，且 socket 缓冲区可写时，对应的事件会频繁发生，故需要在 socket 缓冲区写不下时再关注写事件，写完之后需要取消关注

Client

package com.sw.netty._04.writeableEvents;

import lombok.extern.slf4j.Slf4j;

import java.io.IOException;
import java.net.InetSocketAddress;
import java.nio.ByteBuffer;
import java.nio.channels.SelectionKey;
import java.nio.channels.Selector;
import java.nio.channels.SocketChannel;
import java.util.Iterator;

@Slf4j
public class Client {
    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", 8088));

        int count = 0;
        while (true) {
            selector.select();
            Iterator<SelectionKey> iterator = selector.selectedKeys().iterator();
            while (iterator.hasNext()) {
                SelectionKey key = iterator.next();
                iterator.remove();

                if (key.isConnectable()) {
                    log.info("key [{}] connected", key);
                    sc.finishConnect();
                }

                if (key.isReadable()) {
                    ByteBuffer bf = ByteBuffer.allocate(1024 * 1024);
                    count += sc.read(bf);
                    bf.clear();
                    System.out.println("接收到：" + count + " 字节数据");
                }
            }
        }
    }
}

NIO基础 - 网络编程

4. 网络编程

4.1 阻塞模式

4.2 非阻塞模式

4.3 多路复用

4.4 Selector

4.5 处理 accept 事件

4.6 处理 read 事件

4.7 处理消息边界

4.8 ByteBuffer 大小分配

4.9 处理 write 事件

Netty - 协议设计与解析

Netty - 粘包、半包

Netty - ByteBuf

Netty - Channel & Pipeline

评论 (0)