零、 整体流程
1、用户在main线程启动执行ServerSocketChannel的初始化
1）初始化一个NioServerSocketChannel channel
包括初始化Channel关联的ch、pipeline、unsafe、config。
nio.ServerSocketChannel ch = nio.SelectorProvider.openServerSocketChannel()；
ch.configBlocking(false);
2）将channel与ServerBootstrap的EventLoopGroup中的某个EventLoop child绑定
3）向child的任务执行队列中添加channel的register0事件
4）监听register0事件的完成状态，完成时向child的任务执行队列中添加channel的bind事件
2、EventLoop线程中执行register0事件
1）将NioServerSocketChannel的ch注册到NioEventLoop child的selector上
SelectionKey selectionKey = ch.register(eventLoop().selector, 0, this);
2）标识ChannelPromise状态为success，触发Listener将bind任务添加到EventLoop的执行任务队列
3）ServerSocketChannel channel产生ChannelRegisted事件
pipeline.fireChannelRegistered();
会导致ChannelInitialzer.channelRegisted()执行，将ServerBootstrapAcceptor添加到channel的pipeline中。
3、EventLoop线程中执行Channel.bind事件
1）ServerSocketChannel绑定对应服务端口，监听新的客户端连接
javaChannel().socket().bind(localAddress, config.getBacklog());
2）ServerSocketChannel channel产生ChannelActive事件
pipeline.fireChannelActive();
ChannelActive事件由HeadContext处理，向ch中添加OP_ACCEPT事件监听。
selectionKey.interestOps(OP_ACCEPT);
一、代码入口
ServerBootstrap b = new ServerBootstrap();
// Start the server.
ChannelFuture f = b.bind(port).sync();
服务端启动时都会执行上面的代码，用来启动ServerSocketChannel监听对应端口。
最终由AbstractBootstrap.doBind方法处理。
private ChannelFuture doBind(final SocketAddress localAddress) {
final ChannelFuture regFuture = initAndRegister();
final Channel channel = regFuture.channel();
if (regFuture.cause() != null) {
return regFuture;
}
if (regFuture.isDone()) {
// At this point we know that the registration was complete and successful.
ChannelPromise promise = channel.newPromise();
doBind0(regFuture, channel, localAddress, promise);
return promise;
} else {
// Registration future is almost always fulfilled already, but just in case it's not.
final PendingRegistrationPromise promise = new PendingRegistrationPromise(channel);
regFuture.addListener(new ChannelFutureListener() {
@Override
public void operationComplete(ChannelFuture future) throws Exception {
Throwable cause = future.cause();
if (cause != null) {
// Registration on the EventLoop failed so fail the ChannelPromise directly to not cause an
// IllegalStateException once we try to access the EventLoop of the Channel.
promise.setFailure(cause);
} else {
// Registration was successful, so set the correct executor to use.
// See https://github.com/netty/netty/issues/2586
promise.registered();
doBind0(regFuture, channel, localAddress, promise);
}
}
});
return promise;
}
}
二、用户线程初始化NioServerSocketChannel
final Channel channel = channelFactory.newChannel();
从此行代码开始，ReflectiveChannelFactory通过反射的方式调用NioServerSocketChannel的构造方法进行channel的初始化。
1、NioServerSocketChannel的实例化
** 1）创建一个nio.ServerSocketChannel**
newSocket(SelectorProvider provider);
使用nio.SelectorProvider.openServerSocketChannel()创建一个nio.ServerSocketChannel ch。
** 2）通过构造方法实例化NioServerSocketChannel，并初始化相关的field**
parent = null
unsafe = new NioMessageUnsafe()
pipeline = new DefaultChannelPipeline(this) ->此处初始化一个DefaultChannelPipeline，并将pipeline和channel互相绑定
ch = ch ->同时将ch设置为非阻塞模式
readInterestOp = OP_ACCEPT
config = new ServerSocketChannelConfig(this, javaChannel().socket())
3）向channel的pipeline中添加Inbound处理器ChannelInitializer
init(channel);
由其父类ServerBootstrap直接实现。
- 设置channel的attr和options**
- 添加ChannelInitializer到pipeline中**
p.addLast(new ChannelInitializer<Channel>() {
@Override
public void initChannel(Channel ch) throws Exception {
ChannelPipeline pipeline = ch.pipeline();
ChannelHandler handler = config.handler();
if (handler != null) {
pipeline.addLast(handler);
}
pipeline.addLast(new ServerBootstrapAcceptor(
currentChildGroup, currentChildHandler, currentChildOptions, currentChildAttrs));
}
}
该处理器的*channelRegisted事件*回调方法会将*ServerBootstrapAcceptor*加入channel的pipeline中，并将自己从pipeline中移除。
4）ServerBootstrapAcceptor
- ServerBootstrapAcceptor也是一个Inbound处理器，用于在Server端accept新的客户端连接时，向新生成的socketChannel中添加用户定义的业务处理器。
- 其channelRead事件回调方法会将业务方往ServerBootstrap中添加的childHandler添加到socketChannel对应的pipeline中。
- 对于Server端，channelRead事件被定义为server端accept到了新的socket连接。
2、将NioServerSocketChannel注册到ServerBootstrap的EventLoopGroup上
ChannelFuture regFuture = group().register(channel);
1）NioEventLoopGroup的chooser从其children中选出一个NioEventLoop child，调用其register()方法进行channel注册；
2）实际将NioEventLoop child传给NioServerSocketChannel的unsafe，调用其register(EventLoop eventLoop, final ChannelPromise promise)方法完成注册
- 将channel.eventLoop绑定为当前NioEventLoop child；
- 将AbstractUnsafe.register0(DefaultChannelPromise promise)任务加入EventLoop child的执行任务队列；
3、给标识“register0任务”完成状态ChannelFuture添加一个Listener
doBind0(regFuture, channel, localAddress, promise);
当ChannelFuture完成时，将NioServerSocketChannel.bind(SocketAddress localAddress, ChannelPromise promise)任务加入EventLoop child的执行任务队列
三、EventLoop线程中执行register0事件
AbstractUnsafe.register0(ChannelPromise promise);
1、将NioServerSocketChannel的ch注册到NioEventLoop child的selector上，同时将注册得到的SelectionKey绑定为NioServerSocketChannel的selectionKey
doRegister();
selectionKey = javaChannel().register(eventLoop().selector, 0, this);
netty的轮询注册机制其实是将AbstractNioChannel内部的jdk类对象SelectableChannel ch注册到jdk类对象Selector selector上去，并且将AbstractNioChannel channel作为SelectableChannel对象ch的一个attachment附属上，这样在jdk轮询出某个SelectableChannel有IO事件发生时，就可以直接取出AbstractNioChannel进行后续操作。
2、标识ChannelPromise状态为success，触发Listener将bind任务添加到EventLoop的执行任务队列
safeSetSuccess(promise);
3、ServerSocketChannel channel产生ChannelRegisted事件
pipeline.fireChannelRegistered();
会导致ChannelInitialzer.channelRegisted()执行，将ServerBootstrapAcceptor添加到channel的pipeline中。
四、EventLoop线程中执行Channel.bind事件
AbstractUnsafe.bind(final SocketAddress localAddress, final ChannelPromise promise);
1、ServerSocketChannel绑定对应服务端口，监听新的客户端连接
javaChannel().socket().bind(localAddress, config.getBacklog());
2、ServerSocketChannel channel产生ChannelActive事件
pipeline.fireChannelActive();
ChannelActive事件由HeadContext处理，最终调用了AbstractNioUnsafe.doBeginRead()方法，向ch中添加OP_READ事件监听。
selectionKey.interestOps(interestOps | readInterestOp);
