海口网站建设咨询,怎么增加网站首页权重,广州市医院网站建设哪家好,推广网站怎么建设2019独角兽企业重金招聘Python工程师标准 前言 Java NIO 由以下几个核心部分组成#xff1a; 1、Buffer 2、Channel 3、Selector Buffer和Channel在深入浅出NIO之Channel、Buffer一文中已经介绍过#xff0c;本文主要讲解NIO的Selector实现原理。 之前进行sock… 2019独角兽企业重金招聘Python工程师标准 前言 Java NIO 由以下几个核心部分组成 1、Buffer 2、Channel 3、Selector Buffer和Channel在深入浅出NIO之Channel、Buffer一文中已经介绍过本文主要讲解NIO的Selector实现原理。 之前进行socket编程时accept方法会一直阻塞直到有客户端请求的到来并返回socket进行相应的处理。整个过程是流水线的处理完一个请求才能去获取并处理后面的请求当然也可以把获取socket和处理socket的过程分开一个线程负责accept一个线程池负责处理请求。 但NIO提供了更好的解决方案采用选择器Selector返回已经准备好的socket并按顺序处理基于通道Channel和缓冲区Buffer来进行数据的传输。 Selector 这里出来一个新概念selector具体是一个什么样的东西 想想一个场景在一个养鸡场有这么一个人每天的工作就是不停检查几个特殊的鸡笼如果有鸡进来有鸡出去有鸡生蛋有鸡生病等等就把相应的情况记录下来如果鸡场的负责人想知道情况只需要询问那个人即可。 在这里这个人就相当Selector每个鸡笼相当于一个SocketChannel每个线程通过一个Selector可以管理多个SocketChannel。 为了实现Selector管理多个SocketChannel必须将具体的SocketChannel对象注册到Selector并声明需要监听的事件这样Selector才知道需要记录什么数据一共有4种事件 1、connect客户端连接服务端事件对应值为SelectionKey.OP_CONNECT(8) 2、accept服务端接收客户端连接事件对应值为SelectionKey.OP_ACCEPT(16) 3、read读事件对应值为SelectionKey.OP_READ(1) 4、write写事件对应值为SelectionKey.OP_WRITE(4) 这个很好理解每次请求到达服务器都是从connect开始connect成功后服务端开始准备accept准备就绪开始读数据并处理最后写回数据返回。 所以当SocketChannel有对应的事件发生时Selector都可以观察到并进行相应的处理。 服务端代码 为了更好的理解先看一段服务端的示例代码 ServerSocketChannel serverChannel ServerSocketChannel.open();
serverChannel.configureBlocking(false);
serverChannel.socket().bind(new InetSocketAddress(port));
Selector selector Selector.open();
serverChannel.register(selector, SelectionKey.OP_ACCEPT);
while(true){int n selector.select();if (n 0) continue;Iterator ite this.selector.selectedKeys().iterator();while(ite.hasNext()){SelectionKey key (SelectionKey)ite.next();if (key.isAcceptable()){SocketChannel clntChan ((ServerSocketChannel) key.channel()).accept();clntChan.configureBlocking(false);//将选择器注册到连接到的客户端信道//并指定该信道key值的属性为OP_READ//同时为该信道指定关联的附件clntChan.register(key.selector(), SelectionKey.OP_READ, ByteBuffer.allocate(bufSize));}if (key.isReadable()){handleRead(key);}if (key.isWritable() key.isValid()){handleWrite(key);}if (key.isConnectable()){System.out.println(isConnectable true);}ite.remove();}
}服务端操作过程 1、创建ServerSocketChannel实例并绑定指定端口 2、创建Selector实例 3、将serverSocketChannel注册到selector并指定事件OP_ACCEPT最底层的socket通过channel和selector建立关联 4、如果没有准备好的socketselect方法会被阻塞一段时间并返回0 5、如果底层有socket已经准备好selector的select方法会返回socket的个数而且selectedKeys方法会返回socket对应的事件connect、accept、read or write 6、根据事件类型进行不同的处理逻辑 在步骤3中selector只注册了serverSocketChannel的OP_ACCEPT事件 1、如果有客户端A连接服务执行select方法时可以通过serverSocketChannel获取客户端A的socketChannel并在selector上注册socketChannel的OP_READ事件。 2、如果客户端A发送数据会触发read事件这样下次轮询调用select方法时就能通过socketChannel读取数据同时在selector上注册该socketChannel的OP_WRITE事件实现服务器往客户端写数据。 Selector实现原理 SocketChannel、ServerSocketChannel和Selector的实例初始化都通过SelectorProvider类实现其中Selector是整个NIO Socket的核心实现。 public static SelectorProvider provider() {synchronized (lock) {if (provider ! null)return provider;return AccessController.doPrivileged(new PrivilegedActionSelectorProvider() {public SelectorProvider run() {if (loadProviderFromProperty())return provider;if (loadProviderAsService())return provider;provider sun.nio.ch.DefaultSelectorProvider.create();return provider;}});}
}SelectorProvider在windows和linux下有不同的实现provider方法会返回对应的实现。 这里不禁要问Selector是如何做到同时管理多个socket 下面我们看看Selector的具体实现Selector初始化时会实例化PollWrapper、SelectionKeyImpl数组和Pipe。 WindowsSelectorImpl(SelectorProvider sp) throws IOException {super(sp);pollWrapper new PollArrayWrapper(INIT_CAP);wakeupPipe Pipe.open();wakeupSourceFd ((SelChImpl)wakeupPipe.source()).getFDVal();// Disable the Nagle algorithm so that the wakeup is more immediateSinkChannelImpl sink (SinkChannelImpl)wakeupPipe.sink();(sink.sc).socket().setTcpNoDelay(true);wakeupSinkFd ((SelChImpl)sink).getFDVal();pollWrapper.addWakeupSocket(wakeupSourceFd, 0);
}pollWrapper用Unsafe类申请一块物理内存pollfd存放socket句柄fdVal和events其中pollfd共8位0-3位保存socket句柄4-7位保存events。 pollWrapper提供了fdVal和event数据的相应操作如添加操作通过Unsafe的putInt和putShort实现。 void putDescriptor(int i, int fd) {pollArray.putInt(SIZE_POLLFD * i FD_OFFSET, fd);
}
void putEventOps(int i, int event) {pollArray.putShort(SIZE_POLLFD * i EVENT_OFFSET, (short)event);
}先看看serverChannel.register(selector, SelectionKey.OP_ACCEPT)是如何实现的 public final SelectionKey register(Selector sel, int ops, Object att)throws ClosedChannelException {synchronized (regLock) {SelectionKey k findKey(sel);if (k ! null) {k.interestOps(ops);k.attach(att);}if (k null) {// New registrationsynchronized (keyLock) {if (!isOpen())throw new ClosedChannelException();k ((AbstractSelector)sel).register(this, ops, att);addKey(k);}}return k;}
}如果该channel和selector已经注册过则直接添加事件和附件。否则通过selector实现注册过程。protected final SelectionKey register(AbstractSelectableChannel ch,int ops, Object attachment) {if (!(ch instanceof SelChImpl))throw new IllegalSelectorException();SelectionKeyImpl k new SelectionKeyImpl((SelChImpl)ch, this);k.attach(attachment);synchronized (publicKeys) {implRegister(k);}k.interestOps(ops);return k;
}protected void implRegister(SelectionKeyImpl ski) {synchronized (closeLock) {if (pollWrapper null)throw new ClosedSelectorException();growIfNeeded();channelArray[totalChannels] ski;ski.setIndex(totalChannels);fdMap.put(ski);keys.add(ski);pollWrapper.addEntry(totalChannels, ski);totalChannels;}
}1、以当前channel和selector为参数初始化SelectionKeyImpl 对象selectionKeyImpl 并添加附件attachment。 2、如果当前channel的数量totalChannels等于SelectionKeyImpl数组大小对SelectionKeyImpl数组和pollWrapper进行扩容操作。 3、如果totalChannels % MAX_SELECTABLE_FDS 0则多开一个线程处理selector。 4、pollWrapper.addEntry将把selectionKeyImpl中的socket句柄添加到对应的pollfd。 5、k.interestOps(ops)方法最终也会把event添加到对应的pollfd。 所以不管serverSocketChannel还是socketChannel在selector注册的事件最终都保存在pollArray中。 接着再来看看selector中的select是如何实现一次获取多个有事件发生的channel的底层由selector实现类的doSelect方法实现如下 protected int doSelect(long timeout) throws IOException {if (channelArray null)throw new ClosedSelectorException();this.timeout timeout; // set selector timeoutprocessDeregisterQueue();if (interruptTriggered) {resetWakeupSocket();return 0;}// Calculate number of helper threads needed for poll. If necessary// threads are created here and start waiting on startLockadjustThreadsCount();finishLock.reset(); // reset finishLock// Wakeup helper threads, waiting on startLock, so they start polling.// Redundant threads will exit here after wakeup.startLock.startThreads();// do polling in the main thread. Main thread is responsible for// first MAX_SELECTABLE_FDS entries in pollArray.try {begin();try {subSelector.poll();} catch (IOException e) {finishLock.setException(e); // Save this exception}// Main thread is out of poll(). Wakeup others and wait for themif (threads.size() 0)finishLock.waitForHelperThreads();} finally {end();}// Done with poll(). Set wakeupSocket to nonsignaled for the next run.finishLock.checkForException();processDeregisterQueue();int updated updateSelectedKeys();// Done with poll(). Set wakeupSocket to nonsignaled for the next run.resetWakeupSocket();return updated;}其中 subSelector.poll() 是select的核心由native函数poll0实现readFds、writeFds 和exceptFds数组用来保存底层select的结果数组的第一个位置都是存放发生事件的socket的总数其余位置存放发生事件的socket句柄fd。 private final int[] readFds new int [MAX_SELECTABLE_FDS 1];
private final int[] writeFds new int [MAX_SELECTABLE_FDS 1];
private final int[] exceptFds new int [MAX_SELECTABLE_FDS 1];
private int poll() throws IOException{ // poll for the main threadreturn poll0(pollWrapper.pollArrayAddress,Math.min(totalChannels, MAX_SELECTABLE_FDS),readFds, writeFds, exceptFds, timeout);
}执行 selector.select() poll0函数把指向socket句柄和事件的内存地址传给底层函数。 1、如果之前没有发生事件程序就阻塞在select处当然不会一直阻塞因为epoll在timeout时间内如果没有事件也会返回 2、一旦有对应的事件发生poll0方法就会返回 3、processDeregisterQueue方法会清理那些已经cancelled的SelectionKey 4、updateSelectedKeys方法统计有事件发生的SelectionKey数量并把符合条件发生事件的SelectionKey添加到selectedKeys哈希表中提供给后续使用。 在早期的JDK1.4和1.5 update10版本之前Selector基于select/poll模型实现是基于IO复用技术的非阻塞IO不是异步IO。在JDK1.5 update10和linux core2.6以上版本sun优化了Selctor的实现底层使用epoll替换了select/poll。 read实现 通过遍历selector中的SelectionKeyImpl数组获取发生事件的socketChannel对象其中保存了对应的socket实现如下 public int read(ByteBuffer buf) throws IOException {if (buf null)throw new NullPointerException();synchronized (readLock) {if (!ensureReadOpen())return -1;int n 0;try {begin();synchronized (stateLock) {if (!isOpen()) { return 0;}readerThread NativeThread.current();}for (;;) {n IOUtil.read(fd, buf, -1, nd);if ((n IOStatus.INTERRUPTED) isOpen()) {// The system call was interrupted but the channel// is still open, so retrycontinue;}return IOStatus.normalize(n);}} finally {readerCleanup(); // Clear reader thread// The end method, which end(n 0 || (n IOStatus.UNAVAILABLE));// Extra case for socket channels: Asynchronous shutdown//synchronized (stateLock) {if ((n 0) (!isInputOpen))return IOStatus.EOF;}assert IOStatus.check(n);}}
}最终通过Buffer的方式读取socket的数据。 wakeup实现 public Selector wakeup() {synchronized (interruptLock) {if (!interruptTriggered) {setWakeupSocket();interruptTriggered true;}}return this;
}// Sets Windows wakeup socket to a signaled state.
private void setWakeupSocket() {setWakeupSocket0(wakeupSinkFd);
}
private native void setWakeupSocket0(int wakeupSinkFd);看来wakeupSinkFd这个变量是为wakeup方法使用的。 其中interruptTriggered为中断已触发标志当pollWrapper.interrupt()之后该标志即为true了因为这个标志连续两次wakeup只会有一次效果。 epoll原理 epoll是Linux下的一种IO多路复用技术可以非常高效的处理数以百万计的socket句柄。 三个epoll相关的系统调用 int epoll_create(int size) epoll_create建立一个epoll对象。参数size是内核保证能够正确处理的最大句柄数多于这个最大数时内核可不保证效果。int epoll_ctl(int epfd, int op, int fd, struct epoll_event event) epoll_ctl可以操作epoll_create创建的epoll如将socket句柄加入到epoll中让其监控或把epoll正在监控的某个socket句柄移出epoll。int epoll_wait(int epfd, struct epoll_event events,int maxevents, int timeout) epoll_wait在调用时在给定的timeout时间内所监控的句柄中有事件发生时就返回用户态的进程。epoll内部实现大概如下 epoll初始化时会向内核注册一个文件系统用于存储被监控的句柄文件调用epoll_create时会在这个文件系统中创建一个file节点。同时epoll会开辟自己的内核高速缓存区以红黑树的结构保存句柄以支持快速的查找、插入、删除。还会再建立一个list链表用于存储准备就绪的事件。当执行epoll_ctl时除了把socket句柄放到epoll文件系统里file对象对应的红黑树上之外还会给内核中断处理程序注册一个回调函数告诉内核如果这个句柄的中断到了就把它放到准备就绪list链表里。所以当一个socket上有数据到了内核在把网卡上的数据copy到内核中后就把socket插入到就绪链表里。当epoll_wait调用时仅仅观察就绪链表里有没有数据如果有数据就返回否则就sleep超时时立刻返回。觉得不错请点赞支持欢迎留言或进我的个人群855801563领取【架构资料专题目合集90期】、【BATJTMD大厂JAVA面试真题1000】本群专用于学习交流技术、分享面试机会拒绝广告我也会在群内不定期答题、探讨。 转载于:https://my.oschina.net/u/3959491/blog/3032417
