ServerBootstrapAcceptor是用来处理当BossGroup(EventLoopGroup)里面有新的客户端连接产生时候 将新连接(NioSocketChannel)交给WorkerGroup(EventLoopGroup)
private static class ServerBootstrapAcceptor extends ChannelInboundHandlerAdapter {
public void channelRead(ChannelHandlerContext ctx, Object msg) {
final Channel child = (Channel) msg;
//将新的channelHandler(MySocketServerInitializer)添加到pipeline
//同时会添加一个pendingHandlerCallback里面包含了当前ChannelHandlerContext
//当后面调用注册方法的时候会触发pipeline上的HandlerAdded方法回调 pendingHandlerCallback里面的方法
child.pipeline().addLast(childHandler);
setChannelOptions(child, childOptions, logger);
try {
//将新的连接(NioSocketChannel)注册到WorkerGroup中的一个WorkerEventLoop(EventLoop)
//会从WorkerGroup获取一个WorkerEventLoop挑选方法可以看一下
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);
}
}
}
io.netty.channel.AbstractChannel.AbstractUnsafe#register0
private void register0(ChannelPromise promise) {
try {
//真正将channel注册到 WorkerEventLoop中的Selecter上
doRegister();
//调用pipeline.callHandlerAddedForAllHandlers()方法
// PendingHandlerCallback task
// ctx.callHandlerAdded()
// 触发了注册方法 @2
//注:这个时候pipline上只有三个head MySocketServerInitializer tail 三个handler
// 其中 head tail这两个handler是在创建NioSocketChannel时候创建pipeline默认生成的
// 一个NioSocketChannel只绑定一个pipeline 参见@1
// MySocketServerInitializer是在ServerBootstrapAcceptor中拿到连接时候直接加上的
pipeline.invokeHandlerAddedIfNeeded();
safeSetSuccess(promise);
pipeline.fireChannelRegistered();
// Only fire a channelActive if the channel has never been registered. This prevents firing
// multiple channel actives if the channel is deregistered and re-registered.
if (isActive()) {
if (firstRegistration) {
pipeline.fireChannelActive();
} else if (config().isAutoRead()) {
beginRead();
}
}
} catch (Throwable t) {
// Close the channel directly to avoid FD leak.
closeForcibly();
closeFuture.setClosed();
safeSetFailure(promise, t);
}
}
@1
public abstract class AbstractChannel extends DefaultAttributeMap implements Channel {
private final Channel parent;
private final ChannelId id;
private final Unsafe unsafe;
private final DefaultChannelPipeline pipeline;
}
@2
public abstract class ChannelInitializer<C extends Channel> extends ChannelInboundHandlerAdapter {
public void handlerAdded(ChannelHandlerContext ctx) throws Exception {
if (ctx.channel().isRegistered()) {
if (initChannel(ctx)) {
removeState(ctx);
}
}
}
}
网友评论