我们回到最初的niomessageunsafe的read()方法:
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()方法中:
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()方法:
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()这个方法中:
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)中看其是如何做实际操作的:
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()方法:
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的资料请关注其它相关文章！
