文章目录一. Selector.open();二. ServerSocketChannel.open()三. serverChannel.register()四. selector.select()先回顾一下创建server与selector的过程*相当于开启一个server端并绑定为一个端口用于客户端向此发起请求*其中这个server端是非阻塞的selector帮助server端监听事件并当事件到来时处理事件 publicNIOServerDemo(intport){try{this.portport;selectorSelector.open();//创建 可选择通道并配置为非阻塞式//通过open方法创建 服务器套接字 通道ServerSocketChannel serverServerSocketChannel.open();server.configureBlocking(false);server.register(selector,SelectionKey.OP_ACCEPT);//注册感兴趣的事件server.bind(newInetSocketAddress(this.port));//绑定指定端口}catch(Exception e){e.printStackTrace();}}先从Selector的open方法开始看我们看java.nio.channels.Selector类的源码。看看open做了哪些事情一. Selector.open();open来创建SelectorselectorSelector.open();publicstaticSelectoropen()throws IOException{returnSelectorProvider.provider().openSelector();}SelectorProvider.provider() 提供默认的selectorpublicstaticSelectorProviderprovider(){synchronized(lock){//保证了整个Server程序中只有一个KQueueSelectorProvider对象if(provider!null)returnprovider;returnAccessController.doPrivileged(new PrivilegedActionSelectorProvider(){public SelectorProviderrun(){if(loadProviderFromProperty())returnprovider;if(loadProviderAsService())returnprovider;//会根据操作系统来返回不同的实现类mac平台返回KQueueSelectorProviderprovidersun.nio.ch.DefaultSelectorProvider.create();returnprovider;}});}}DefaultSelectorProvider.create默认会创建KQueueSelectorProvidermac环境下publicstaticSelectorProvidercreate(){returnnewKQueueSelectorProvider();}看看 new KQueueSelectorProvider(); 做了哪些事情KQueueSelectorImpl(SelectorProvider var1){super(var1);//关键方法longvar2IOUtil.makePipe(false);//移位处理。this.fd0(int)(var232);this.fd1(int)var2;try{this.kqueueWrappernewKQueueArrayWrapper();this.kqueueWrapper.initInterrupt(this.fd0,this.fd1);this.fdMapnewHashMap();this.totalChannels1;}catch(Throwable var8){try{FileDispatcherImpl.closeIntFD(this.fd0);}catch(IOException var7){var8.addSuppressed(var7);}try{FileDispatcherImpl.closeIntFD(this.fd1);}catch(IOException var6){var8.addSuppressed(var6);}throw var8;}}IOUtil.makePipe(true)是一个本地方法。它返回两个文件描述符高位存放的是通道read端的文件描述符低32位存放的是write端的文件描述符。longvar2IOUtil.makePipe(false);this.fd0(int)(var232);this.fd1(int)var2;把返回的pipe的write端的FD、read端的FD放在this.kqueueWrapper中后面会发现这么做是为了实现Selector的wakeup()。this.kqueueWrappernewKQueueArrayWrapper();this.kqueueWrapper.initInterrupt(this.fd0,this.fd1);二. ServerSocketChannel.open()ServerSocketChannel的实现publicstaticServerSocketChannelopen()throws IOException{returnSelectorProvider.provider().openServerSocketChannel();}public ServerSocketChannelopenServerSocketChannel()throws IOException{returnnewServerSocketChannelImpl(this);}ServerSocketChannelImpl(SelectorProvider var1)throws IOException{super(var1);this.fdNet.serverSocket(true);this.fdValIOUtil.fdVal(this.fd);this.state0;}super(var1); 中var1代表SelectorProvider而默认的SelectorProvider是KQueueSelectorProvider。 ing三. serverChannel.register()接着通过serverChannel.register(selectorSelectionKey.OP_ACCEPT)把Selector和Channel绑定在一起也就是把新建ServerSocketChannel时创建的FD与Selector绑定在一起。到此Server端已启动完成主要创建了以下对象。KQueueSelector为单例对象实际上是调用操作系统的API。KQueueSelectorImpl中包含如下内容。a. IOUtil.makePipe其实就是创建两个FD一个是read端的一个是write端的。b. initInterrupt(this.fd0, this.fd1)保存Selector上注册的FD包括pipe的write端FD和ServerSocketChannel所用的FD。四. selector.select()selector.select()主要调用了KQueueSelectorImpl的doSelectprotectedintdoSelect(longvar1)throws IOException{boolean var3false;if(this.closed){throw newClosedSelectorException();}else{this.processDeregisterQueue();intvar7;try{this.begin();var7this.kqueueWrapper.poll(var1);}finally{this.end();}this.processDeregisterQueue();returnthis.updateSelectedKeys(var7);}}this.kqueueWrapper.poll()是核心也就是轮询kqueueWrapper中保存的FD具体实现是调用native方法kevent0()。var7this.kqueueWrapper.poll(var1);intpoll(longvar1){this.updateRegistrations();intvar3this.kevent0(this.kq,this.keventArrayAddress,128,var1);returnvar3;}private nativeintkevent0(intvar1,longvar2,intvar4,longvar5);顺便看看findNative的逻辑// Invoked in the VM class linking code.staticlongfindNative(ClassLoader loader,String name){VectorNativeLibrarylibsloader!null?loader.nativeLibraries:systemNativeLibraries;synchronized(libs){intsizelibs.size();for(inti0;isize;i){NativeLibrary liblibs.elementAt(i);longentrylib.find(name);if(entry!0)returnentry;}}return0;}kevent0.()会监听kqueueWrapper中的FD有没有数据进出这会造成I/O阻塞直到有数据读写事件发生。比如由于kqueueWrapper中保存的也有ServerSocketChannel的FD所以只要ClientSocket发一份数据到ServerSocket那么kevent0()就会返回又由于kqueueWrapper中保存的也有pipe的write端的FD所以只要pipe的write端向FD发一份数据也会造成poll0()返回如果这两种情况都没有发生那么kevent0()就会一直阻塞也就是selector.select()会一直阻塞如果有任何一种情况发生那么selector.select()就会返回所以在OperationServer的run()里要用while(true)这样可以保证在Selector接收数据并处理完后继续监听poll()。KQueueSelectorImpl.wakeup()public Selectorwakeup(){synchronized(this.interruptLock){if(!this.interruptTriggered){this.kqueueWrapper.interrupt();this.interruptTriggeredtrue;}returnthis;}}wakeup()通过pipe的write端send(scoutFdbyte10)发送一个字节1来唤醒poll()的所以在需要的时候(when ing )就可以调用selector.wakeup()来唤醒Selector。参考《Netty4核心原理与手写RPC框架》
文章目录一. Selector.open();二. ServerSocketChannel.open()三. serverChannel.register()四. selector.select()先回顾一下创建server与selector的过程*相当于开启一个server端并绑定为一个端口用于客户端向此发起请求*其中这个server端是非阻塞的selector帮助server端监听事件并当事件到来时处理事件 publicNIOServerDemo(intport){try{this.portport;selectorSelector.open();//创建 可选择通道并配置为非阻塞式//通过open方法创建 服务器套接字 通道ServerSocketChannel serverServerSocketChannel.open();server.configureBlocking(false);server.register(selector,SelectionKey.OP_ACCEPT);//注册感兴趣的事件server.bind(newInetSocketAddress(this.port));//绑定指定端口}catch(Exception e){e.printStackTrace();}}先从Selector的open方法开始看我们看java.nio.channels.Selector类的源码。看看open做了哪些事情一. Selector.open();open来创建SelectorselectorSelector.open();publicstaticSelectoropen()throws IOException{returnSelectorProvider.provider().openSelector();}SelectorProvider.provider() 提供默认的selectorpublicstaticSelectorProviderprovider(){synchronized(lock){//保证了整个Server程序中只有一个KQueueSelectorProvider对象if(provider!null)returnprovider;returnAccessController.doPrivileged(new PrivilegedActionSelectorProvider(){public SelectorProviderrun(){if(loadProviderFromProperty())returnprovider;if(loadProviderAsService())returnprovider;//会根据操作系统来返回不同的实现类mac平台返回KQueueSelectorProviderprovidersun.nio.ch.DefaultSelectorProvider.create();returnprovider;}});}}DefaultSelectorProvider.create默认会创建KQueueSelectorProvidermac环境下publicstaticSelectorProvidercreate(){returnnewKQueueSelectorProvider();}看看 new KQueueSelectorProvider(); 做了哪些事情KQueueSelectorImpl(SelectorProvider var1){super(var1);//关键方法longvar2IOUtil.makePipe(false);//移位处理。this.fd0(int)(var232);this.fd1(int)var2;try{this.kqueueWrappernewKQueueArrayWrapper();this.kqueueWrapper.initInterrupt(this.fd0,this.fd1);this.fdMapnewHashMap();this.totalChannels1;}catch(Throwable var8){try{FileDispatcherImpl.closeIntFD(this.fd0);}catch(IOException var7){var8.addSuppressed(var7);}try{FileDispatcherImpl.closeIntFD(this.fd1);}catch(IOException var6){var8.addSuppressed(var6);}throw var8;}}IOUtil.makePipe(true)是一个本地方法。它返回两个文件描述符高位存放的是通道read端的文件描述符低32位存放的是write端的文件描述符。longvar2IOUtil.makePipe(false);this.fd0(int)(var232);this.fd1(int)var2;把返回的pipe的write端的FD、read端的FD放在this.kqueueWrapper中后面会发现这么做是为了实现Selector的wakeup()。this.kqueueWrappernewKQueueArrayWrapper();this.kqueueWrapper.initInterrupt(this.fd0,this.fd1);二. ServerSocketChannel.open()ServerSocketChannel的实现publicstaticServerSocketChannelopen()throws IOException{returnSelectorProvider.provider().openServerSocketChannel();}public ServerSocketChannelopenServerSocketChannel()throws IOException{returnnewServerSocketChannelImpl(this);}ServerSocketChannelImpl(SelectorProvider var1)throws IOException{super(var1);this.fdNet.serverSocket(true);this.fdValIOUtil.fdVal(this.fd);this.state0;}super(var1); 中var1代表SelectorProvider而默认的SelectorProvider是KQueueSelectorProvider。 ing三. serverChannel.register()接着通过serverChannel.register(selectorSelectionKey.OP_ACCEPT)把Selector和Channel绑定在一起也就是把新建ServerSocketChannel时创建的FD与Selector绑定在一起。到此Server端已启动完成主要创建了以下对象。KQueueSelector为单例对象实际上是调用操作系统的API。KQueueSelectorImpl中包含如下内容。a. IOUtil.makePipe其实就是创建两个FD一个是read端的一个是write端的。b. initInterrupt(this.fd0, this.fd1)保存Selector上注册的FD包括pipe的write端FD和ServerSocketChannel所用的FD。四. selector.select()selector.select()主要调用了KQueueSelectorImpl的doSelectprotectedintdoSelect(longvar1)throws IOException{boolean var3false;if(this.closed){throw newClosedSelectorException();}else{this.processDeregisterQueue();intvar7;try{this.begin();var7this.kqueueWrapper.poll(var1);}finally{this.end();}this.processDeregisterQueue();returnthis.updateSelectedKeys(var7);}}this.kqueueWrapper.poll()是核心也就是轮询kqueueWrapper中保存的FD具体实现是调用native方法kevent0()。var7this.kqueueWrapper.poll(var1);intpoll(longvar1){this.updateRegistrations();intvar3this.kevent0(this.kq,this.keventArrayAddress,128,var1);returnvar3;}private nativeintkevent0(intvar1,longvar2,intvar4,longvar5);顺便看看findNative的逻辑// Invoked in the VM class linking code.staticlongfindNative(ClassLoader loader,String name){VectorNativeLibrarylibsloader!null?loader.nativeLibraries:systemNativeLibraries;synchronized(libs){intsizelibs.size();for(inti0;isize;i){NativeLibrary liblibs.elementAt(i);longentrylib.find(name);if(entry!0)returnentry;}}return0;}kevent0.()会监听kqueueWrapper中的FD有没有数据进出这会造成I/O阻塞直到有数据读写事件发生。比如由于kqueueWrapper中保存的也有ServerSocketChannel的FD所以只要ClientSocket发一份数据到ServerSocket那么kevent0()就会返回又由于kqueueWrapper中保存的也有pipe的write端的FD所以只要pipe的write端向FD发一份数据也会造成poll0()返回如果这两种情况都没有发生那么kevent0()就会一直阻塞也就是selector.select()会一直阻塞如果有任何一种情况发生那么selector.select()就会返回所以在OperationServer的run()里要用while(true)这样可以保证在Selector接收数据并处理完后继续监听poll()。KQueueSelectorImpl.wakeup()public Selectorwakeup(){synchronized(this.interruptLock){if(!this.interruptTriggered){this.kqueueWrapper.interrupt();this.interruptTriggeredtrue;}returnthis;}}wakeup()通过pipe的write端send(scoutFdbyte10)发送一个字节1来唤醒poll()的所以在需要的时候(when ing )就可以调用selector.wakeup()来唤醒Selector。参考《Netty4核心原理与手写RPC框架》
