前文传送门： Netty客户端接入流程NioSocketChannel创建
我们回到最初的NioMessageUnsafe的read()方法:
|
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 |
public void read() { //必须是NioEventLoop方法调用的, 不能通过外部线程调用 assert eventLoop().inEventLoop(); //服务端channel的config final ChannelConfig config = config(); //服务端channel的pipeline final ChannelPipeline pipeline = pipeline(); //处理服务端接入的速率 final RecvByteBufAllocator.Handle allocHandle = unsafe().recvBufAllocHandle(); //设置配置 allocHandle.reset(config); boolean closed = false ; Throwable exception = null ; try { try { do { //创建jdk底层的channel //readBuf用于临时承载读到链接 int localRead = doReadMessages(readBuf); if (localRead == 0 ) { break ; } if (localRead < 0 ) { closed = true ; break ; } //分配器将读到的链接进行计数 allocHandle.incMessagesRead(localRead); //连接数是否超过最大值 } while (allocHandle.continueReading()); } catch (Throwable t) { exception = t; } int size = readBuf.size(); //遍历每一条客户端连接 for ( int i = 0 ; i < size; i ++) { readPending = false ; //传递事件, 将创建NioSokectChannel进行传递 //最终会调用ServerBootstrap的内部类ServerBootstrapAcceptor的channelRead()方法 pipeline.fireChannelRead(readBuf.get(i)); } readBuf.clear(); allocHandle.readComplete(); pipeline.fireChannelReadComplete(); //代码省略 } finally { //代码省略 } } |
在while循环结束之后, 将会通过一个for循环遍历readBuf集合, 并将创建的NioSocketChannel传入fireChannelRead()中, 传播channel的读取事件
有关pipeline的知识, 我们下一章会详细剖析, 并会根据剖析后的内容回顾之前的有关pipeline的操作, 这里我们只需知道, 通过fireChannelRead()我们最终调用了ServerBootstrap的内部类ServerBootstrapAcceptor 中的channelRead()方法
跟到channelRead()方法中:
|
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 |
public void channelRead(ChannelHandlerContext ctx, Object msg) { final Channel child = (Channel) msg; //代码省略 try { //work线程注册channel childGroup.register(child).addListener( new ChannelFutureListener() { @Override public void operationComplete(ChannelFuture future) throws Exception { if (!future.isSuccess()) { forceClose(child, future.cause()); } } }); } catch (Throwable t) { forceClose(child, t); } } |
其中参数的msg就是最初传入fireChannelRead()方法的NioSocketChannel
所以这里可以通过 final Channel child = (Channel) msg 这种方式拿到NioSocketChannel
其中childGroup是我们最初初始化的work线程, 这里的register()方法跟boss线程一样, 通过next()方法获选择一个线程进行注册, 这里不再赘述
我们紧跟调用链, 跟到SingleThreadEventLoop的register()方法:
|
1 2 3 4 5 |
public ChannelFuture register( final ChannelPromise promise) { ObjectUtil.checkNotNull(promise, "promise" ); promise.channel().unsafe().register( this , promise); return promise; } |
这里的unsafe(), 根据我们之前的剖析, 是NioByteUnsafe, 这里的register最终会调用AbstractUnsafe的register()方法, 并NioSocketChannel
不知道同学们是否记得, 当初NioServerSocketChannel注册的时候也走的这个方法
我们跟到register()这个方法中:
|
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 |
public final void register(EventLoop eventLoop, final ChannelPromise promise) { //省略验证代码 //所有的复制操作, 都交给eventLoop处理 AbstractChannel. this .eventLoop = eventLoop;
if (eventLoop.inEventLoop()) { //做实际主注册 register0(promise); } else { try { eventLoop.execute( new Runnable() { @Override public void run() { register0(promise); } }); } catch (Throwable t) { //代码省略 } } } |
我们学习过NioEventLoop相关知识之后, 应该对这部分代码不太陌生, 首先判断是不是当前NioEventLoop线程, 如果是, 则直接进行注册操作, 如果不是, 则封装成task在当前NioEventLoop中执行
走到这里不难明白, 这里并不是当前NioEventLoop线程, 这是boss线程执行的, 所以这里会走到else, 如果是第一次的连接操作, work线程的NioEventLoop并没有启动, 所以这里也会启动NioEventLoop, 并开始轮询操作
跟到register0(promise)中看其是如何做实际操作的:
|
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 |
private void register0(ChannelPromise promise) { try { //省略代码 //做实际的注册 doRegister(); neverRegistered = false ; registered = true ; //触发事件 pipeline.invokeHandlerAddedIfNeeded(); safeSetSuccess(promise); //触发注册成功事件 pipeline.fireChannelRegistered(); if (isActive()) { if (firstRegistration) { //传播active事件(4) pipeline.fireChannelActive(); } else if (config().isAutoRead()) { beginRead(); } } } catch (Throwable t) { //省略代码 } } |
这段代码我们同样并不陌生, 因为NioServerSokectChannel中也走这一部分, 我们继续关注doRegister()方法:
|
1 2 3 4 5 6 7 8 9 10 11 12 13 |
protected void doRegister() throws Exception { boolean selected = false ; for (;;) { try { //jdk底层的注册方法 //第一个参数为selector, 第二个参数表示不关心任何事件 selectionKey = javaChannel().register(eventLoop().selector, 0 , this ); return ; } catch (CancelledKeyException e) { //省略代码 } } } |
这部分也是我们之前剖析过的jdk底层的注册, 只是不同的是, 这里的javaChannel()是SocketChanel而不是ServerSocketChannel
同样, 这里也是表示不关心任何事件, 只是在当前NioEventLoop绑定的selector上注册
至此, NioSocketChannel完成注册
以上就是Netty代码跟踪NioSocketChannel注册到selector的详细内容，更多关于NioSocketChannel注册到selector的资料请关注其它相关文章！
原文链接：https://HdhCmsTestcnblogs测试数据/xiangnan6122/p/10204023.html
查看更多关于Netty分布式NioSocketChannel注册到selector方法解析的详细内容...