Netty5的类层次结构相比于之前版本有了很大的变化，我准备写一系列文章来分析Netty5的源码。这篇讲讲服务器绑定的过程。
先总结一下，服务绑定主要做了几件事:
1. 创建NioEventLoopGroup线程池
2. 创建NioServerSocketChannel，并初始化，注册时没有直接把SelectionKey.OP_ACCEPT注册上，只是注册了一个0，把AbstractNioChannel作为attachment绑定到selectionkey了，但是传递了OP_ACCEPT到AbstractNioChannel的readInterestOp属性。
3. 初始化Pipeline以及相关的ChannelHandlerContext数据结构
4. 在读数据开始时，根据AbstractNioChannel的readInterestOp的值，把SelectionKey.OP_ACCEPT真正注册上。selectionKey.interestOps(interestOps | readInterestOp);
5. 调用ServerSocketChannel完成端口绑定。
一个典型的Netty5服务器端代码如下，首先需要提供两个EventLoopGroup线程池，在随后的文章中会分析Netty的线程模型，这里只需要知道两个线程池，bossGroup是用来接受客户端TCP连接，workGroup是用来处理IO事件。然后指定采用何种Channel，Netty抽象了一组Channel的结构，和Java本身的Channel概念基本一致，NioServerSocketChannel相当于ServerSocketChannel。最后指定一组ChannelHandler来处理业务功能。
public void bind(int port) throws Exception {
EventLoopGroup bossGroup = new NioEventLoopGroup();
EventLoopGroup workerGroup = new NioEventLoopGroup();
try {
ServerBootstrap b = new ServerBootstrap();
b.group(bossGroup, workerGroup)
.channel(NioServerSocketChannel.class)
.option(ChannelOption.SO_BACKLOG, 1024)
.childHandler(new ChildChannelHandler());
ChannelFuture f = b.bind(port).sync();
System.out.println("Netty time Server started at port " + port);
f.channel().closeFuture().sync();
} finally {
bossGroup.shutdownGracefully();
workerGroup.shutdownGracefully();
}
}
我们先看一下new NioEventLoopGroup()，它实际上创建了一个默认的线程池，线程的数量是Runtime.getRuntime().availableProcessors() * 2,线程的实现是NioEventLoop
static {
DEFAULT_EVENT_LOOP_THREADS = Math.max(1, SystemPropertyUtil.getInt(
"io.netty.eventLoopThreads", Runtime.getRuntime().availableProcessors() * 2));
if (logger.isDebugEnabled()) {
logger.debug("-Dio.netty.eventLoopThreads: {}", DEFAULT_EVENT_LOOP_THREADS);
}
}
/**
* @see {@link MultithreadEventExecutorGroup#MultithreadEventExecutorGroup(int, Executor, Object...)}
*/
protected MultithreadEventLoopGroup(int nThreads, Executor executor, Object... args) {
super(nThreads == 0 ? DEFAULT_EVENT_LOOP_THREADS : nThreads, executor, args);
}
服务器端从ServerBootstrap.bind(port)启动,我们来看看bind方法，在doBind方法中，首先通过initAndRegister()方法来创建一个Channel，这里创建的是NioServerSocketChannel
public ChannelFuture bind(SocketAddress localAddress) {
validate();
if (localAddress == null) {
throw new NullPointerException("localAddress");
}
return doBind(localAddress);
}
private ChannelFuture doBind(final SocketAddress localAddress) {
final ChannelFuture regFuture = initAndRegister();
final Channel channel = regFuture.channel();
if (regFuture.cause() != null) {
return regFuture;
}
final ChannelPromise promise;
if (regFuture.isDone()) {
promise = channel.newPromise();
doBind0(regFuture, channel, localAddress, promise);
} else {
// Registration future is almost always fulfilled already, but just in case it's not.
promise = new DefaultChannelPromise(channel, GlobalEventExecutor.INSTANCE);
regFuture.addListener(new ChannelFutureListener() {
@Override
public void operationComplete(ChannelFuture future) throws Exception {
doBind0(regFuture, channel, localAddress, promise);
}
});
}
return promise;
}
final ChannelFuture initAndRegister() {
Channel channel;
try {
channel = createChannel();
} catch (Throwable t) {
return VoidChannel.INSTANCE.newFailedFuture(t);
}
try {
init(channel);
} catch (Throwable t) {
channel.unsafe().closeForcibly();
return channel.newFailedFuture(t);
}
ChannelPromise regFuture = channel.newPromise();
channel.unsafe().register(regFuture);
if (regFuture.cause() != null) {
if (channel.isRegistered()) {
channel.close();
} else {
channel.unsafe().closeForcibly();
}
}
return regFuture;
}
Channel createChannel() {
EventLoop eventLoop = group().next();
return channelFactory().newChannel(eventLoop, childGroup);
}
// ServerBootstrapChannelFactory 根据NioServerSocketChannel.class来创建Channel实例
private static final class ServerBootstrapChannelFactory<T extends ServerChannel>
implements ServerChannelFactory<T> {
private final Class<? extends T> clazz;
ServerBootstrapChannelFactory(Class<? extends T> clazz) {
this.clazz = clazz;
}
@Override
public T newChannel(EventLoop eventLoop, EventLoopGroup childGroup) {
try {
Constructor<? extends T> constructor = clazz.getConstructor(EventLoop.class, EventLoopGroup.class);
return constructor.newInstance(eventLoop, childGroup);
} catch (Throwable t) {
throw new ChannelException("Unable to create Channel from class " + clazz, t);
}
}
@Override
public String toString() {
return StringUtil.simpleClassName(clazz) + ".class";
}
}
注册NioServerSocketChannel到selector的代码如下，把AbstractNioChannel作为Attachment绑定到selectionKey上
@Override
protected void doRegister() throws Exception {
boolean selected = false;
for (;;) {
try {
selectionKey = javaChannel().register(eventLoop().selector, 0, this);
return;
} catch (CancelledKeyException e) {
if (!selected) {
// Force the Selector to select now as the "canceled" SelectionKey may still be
// cached and not removed because no Select.select(..) operation was called yet.
eventLoop().selectNow();
selected = true;
} else {
// We forced a select operation on the selector before but the SelectionKey is still cached
// for whatever reason. JDK bug ?
throw e;
}
}
}
}
再看AbstractBootstrap的doBind0方法，它创建了一个Runnable对象，其中调用AbstractChannel的bind()方法
private static void doBind0(
final ChannelFuture regFuture, final Channel channel,
final SocketAddress localAddress, final ChannelPromise promise) {
// This method is invoked before channelRegistered() is triggered. Give user handlers a chance to set up
// the pipeline in its channelRegistered() implementation.
channel.eventLoop().execute(new Runnable() {
@Override
public void run() {
if (regFuture.isSuccess()) {
channel.bind(localAddress, promise).addListener(ChannelFutureListener.CLOSE_ON_FAILURE);
} else {
promise.setFailure(regFuture.cause());
}
}
});
}
再看AbstractChannel的bind()方法，把bind事件通过pipeline传递出去
@Override
public ChannelFuture bind(SocketAddress localAddress, ChannelPromise promise) {
return pipeline.bind(localAddress, promise);
}
在DefaultChannelPipleline的bind方法中，从尾部Tail的ChannelHandlerContext开始传递bind事件，会选取一组关心Bind事件的ChannelHandlerContext组成链,然后依次调用。
@Override
public ChannelFuture bind(SocketAddress localAddress, ChannelPromise promise) {
return tail.bind(localAddress, promise);
}
public ChannelFuture bind(final SocketAddress localAddress, final ChannelPromise promise) {
DefaultChannelHandlerContext next = findContextOutbound(MASK_BIND);
next.invoker.invokeBind(next, localAddress, promise);
return promise;
}
每个ChannelHandlerContext都会有一个handler对象，这里处理bind事件的handler是HeadHandler，调用HeadHandler的bind方法，在这个方法中，委托给Unsafe来进行实际的bind操作，这里调用的是AbstractUnsafe对象的bind方法，这是个模板方法，最终的doBind方法委托给了NioServerSocketChannel
public void bind(
ChannelHandlerContext ctx, SocketAddress localAddress, ChannelPromise promise)
throws Exception {
unsafe.bind(localAddress, promise);
}
public final void bind(final SocketAddress localAddress, final ChannelPromise promise) {
if (!ensureOpen(promise)) {
return;
}
// See: https://github.com/netty/netty/issues/576
if (!PlatformDependent.isWindows() && !PlatformDependent.isRoot() &&
Boolean.TRUE.equals(config().getOption(ChannelOption.SO_BROADCAST)) &&
localAddress instanceof InetSocketAddress &&
!((InetSocketAddress) localAddress).getAddress().isAnyLocalAddress()) {
// Warn a user about the fact that a non-root user can't receive a
// broadcast packet on *nix if the socket is bound on non-wildcard address.
logger.warn(
"A non-root user can't receive a broadcast packet if the socket " +
"is not bound to a wildcard address; binding to a non-wildcard " +
"address (" + localAddress + ") anyway as requested.");
}
boolean wasActive = isActive();
try {
doBind(localAddress);
} catch (Throwable t) {
promise.setFailure(t);
closeIfClosed();
return;
}
if (!wasActive && isActive()) {
invokeLater(new Runnable() {
@Override
public void run() {
pipeline.fireChannelActive();
}
});
}
promise.setSuccess();
}
NioServerSocketChannel封装了Java的ServerSocketChannel,完成最终的bind
@Override
protected void doBind(SocketAddress localAddress) throws Exception {
javaChannel().socket().bind(localAddress, config.getBacklog());
}