Netty历险记
Netty的底层依然是依赖于JDK的NIO . 开发NIO服务端的代码如下所示
import java.io.IOException;import java.net.InetSocketAddress;import java.nio.ByteBuffer;import java.nio.channels.*;import java.util.Iterator;import java.util.Set;public class Server {// 缓冲区的大小private static final int BUFFER_SIZE = 1024;// 缓冲区private static ByteBuffer buffer = ByteBuffer.allocate(BUFFER_SIZE);// 选择器private static Selector selector = null;public static void main(String[] args) {ServerSocketChannel serverSocketChannel;try {selector = Selector.open();serverSocketChannel = ServerSocketChannel.open();serverSocketChannel.socket().bind(new InetSocketAddress(8080), 64);serverSocketChannel.configureBlocking(false);serverSocketChannel.register(selector, SelectionKey.OP_ACCEPT);for (;;) {int readyChannels = selector.select();if (readyChannels == 0) {continue;}Set<SelectionKey> selectedKeys = selector.selectedKeys();Iterator iterator = selectedKeys.iterator();while (iterator.hasNext()) {SelectionKey key = (SelectionKey) iterator.next();handleKey(key);iterator.remove();}}} catch (Exception ignored) {}}// 处理SelectionKeyprivate static void handleKey(SelectionKey key) throws IOException {// 是否有连接进来if (key.isAcceptable()) {ServerSocketChannel server = (ServerSocketChannel) key.channel();SocketChannel socketChannel = server.accept();// SocketChannel通道的可读事件注册到Selector中registerChannel(selector, socketChannel, SelectionKey.OP_READ);// 连接成功 向Client打个招呼if (socketChannel.isConnected()) {buffer.clear();buffer.put("I am Server...".getBytes());buffer.flip();socketChannel.write(buffer);}}// 通道的可读事件就绪if (key.isReadable()) {SocketChannel socketChannel = (SocketChannel) key.channel();buffer.clear(); // 清空缓冲区// 读取数据int len = 0;while ((len = socketChannel.read(buffer)) > 0) {buffer.flip();while (buffer.hasRemaining()) {System.out.println("Server读取的数据:" + new String(buffer.array(), 0, len));}}if (len < 0) {// 非法的SelectionKey 关闭ChannelsocketChannel.close();}// SocketChannel通道的可写事件注册到Selector中registerChannel(selector, socketChannel, SelectionKey.OP_WRITE);}// 通道的可写事件就绪if (key.isWritable()) {SocketChannel socketChannel = (SocketChannel) key.channel();buffer.clear(); // 清空缓冲区// 准备发送的数据String message_from_server = "Hello,Client... " + socketChannel.getLocalAddress();buffer.put(message_from_server.getBytes());buffer.flip();socketChannel.write(buffer);System.out.println("Server发送的数据:" + message_from_server);// SocketChannel通道的可写事件注册到Selector中registerChannel(selector, socketChannel, SelectionKey.OP_READ);}}private static void registerChannel(Selector selector, SelectableChannel channel, int ops) throws IOException {if (channel == null) {return;}channel.configureBlocking(false);channel.register(selector, ops);}}
本篇文章就来讲解下selector.select()的功能 .
个人认为, 好多功能都是按照三部曲来实现的
1.生产一个冰箱
2.把大象装进冰箱
3.把大象从冰箱取出来
1.生产一个冰箱
在调用Selector.open()的时候, 底层会创建各种属性和数据结构,用于存储相关信息 .
// 源码位置 java.nio.channels.Selector#openpublic static Selector open() throws IOException {return SelectorProvider.provider().openSelector();}
在Windows平台返回 WindowsSelectorImpl 对象 , 在Linux平台返回 EPollSelectorImpl 对象 . 这里以EPollSelectorImpl分析 .
在Windows平台 , 通过跟踪open()源码的方式, 看不到sun.nio.ch.EPollSelectorImpl 这个类, 可以在Linux平台 或者 直接下载JDK源码 或者互联网搜索EPollSelectorImpl都可以看到这个类 .
EPollSelectorImpl 继承 SelectorImpl , 在 EPollSelectorImpl 内部有个EPollArrayWrapper类 , EPollArrayWrapper内部就是关于epoll相关的操作 .
IO多路复用的实现方式有 select, poll, epoll .
epoll 主要涉及三个方法: epoll_create, epoll_ctl, epoll_wait
个人认为, 要想学好Java, 依然要对C语言, 包括一些系统调用了解或熟悉.
在实例化EPollSelectorImpl的时候, 创建了
Set<SelectionKey> selectedKeys ,
Map<Integer,SelectionKeyImpl> fdToKey,
byte[] eventsLow 或 Map<Integer,Byte> eventsHigh 等重要属性 .
我们不必在意这些属性'散落'在哪些类里, 我们更关注的是, 实例化EPollSelectorImpl的时候 会 创建一些集合等属性对象, 用于存储数据. 这就是在生产一个冰箱, 为后面存储数据使用.
而且还会创建一个堆外内存的pollArray对象, 这个对象用于接收内核返回的可读写的文件描述符. 因为在进行调用epoll_wait的时候, 需要给内核传递一个对象, 内核会将已经准备就绪的文件描述符填充到这个对象.
通过man epoll_wait查看
第二个参数 struct epoll_event *events 是一个传出参数, 而pollArray对象就会传到这个参数上 .
所有与内核交互的对象, 必须是堆外内存的对象 .
2.把大象装进冰箱
在我们自己的代码中 会通过
serverSocketChannel.register(selector, SelectionKey.OP_ACCEPT) 注册感兴趣的事件 .
文章后面会有一个函数之间调用的总图
// 源码位置 EPollSelectorImpl#implRegisterprotected void implRegister(SelectionKeyImpl ski) {if (closed)throw new ClosedSelectorException();SelChImpl ch = ski.channel;int fd = Integer.valueOf(ch.getFDVal());// <fd, SelectionKeyImpl>关系fdToKey.put(fd, ski);pollWrapper.add(fd);keys.add(ski);}
会将<fd, SelectionKeyImpl>的对应关系存储到 Map<Integer,SelectionKeyImpl> fdToKey 集合中, 假如我们有个6号文件描述符 fd=6, 把它存储到 fdToKey中, 即<6, SelectionKeyImpl>这样的关系 . 当6号文件描述符有数据进来的时候, 调用epoll_wait的时候, 内核就会把6号文件描述符返回给用户态, 我们再根据<6, SelectionKeyImpl>这个关系,就能找到这个SelectionKeyImpl 了 .
在调用register方法的时候, 不仅会存储<fd, SelectionKeyImpl>的对应关系, 还会将所有的fd存储到 int[] updateDescriptors 中, 也会将 <fd, events>的关系存储到byte[] eventsLow 或 Map<Integer,Byte> eventsHigh中.
一句话, 在上面我们已经生产了一个冰箱, 在这里, 我们把数据(也就是大象)放进这个冰箱里面.
3.把大象从冰箱取出来
selector.select()
关键代码最终会调用到EPollArrayWrapper 这个类里的方法.
会将之前上一步的文件描述符和对应的事件, 通过epoll_ctl系统调用, 放到epoll的红黑树上.
最终会调用到epoll_wait系统调用函数, 如果有文件描述符就绪, 就将对应的文件描述符放到堆外内存的pollArray对象上.
用户态拿到pollArray对象之后, 通过遍历, 根据fd从fdToKey中将SelectionKeyImpl 放到 Set<SelectionKey> selectedKeys集合中, 用户态在调用selector.selectedKeys()的时候, 就会将selectedKeys集合返回 . 这样我们的业务代码就拿到了 selectedKeys集合, 进行后续操作处理.
关于函数之间的调用如下图, 具体也可以查看
https://www.yuque.com/infuq/default/wy8fap#cGAYr
