【Netty 源码】服务端启动流程源码分析 篇一
1.原生Java NIO服务端创建流程
使用Java NIO创建服务端时，通常我们需要先创建Channel，Selector两个对象，然后将Channel绑定端口并注册到Selector上，最后对事件轮询监听
//第一步：创建Channel
ServerSocketChannel channel = ServerSocketChannel.open();
channel.configureBlocking(false);
//第二步:创建selector
Selector selector = Selector.open();
//第三步：将Channel注册到selector
SelectionKey selectionKey = channel.register(selector, 0, new Object());
//第四步：Channel监听端口
channel.bind(new InetSocketAddress(8080));
//第五步：关注感兴趣的事件
selectionKey.interestOps(SelectionKey.OP_ACCEPT);
2.Netty 服务端创建流程
Netty的服务端创建流程都在 ServerBootstrap.bind方法中完成
private ChannelFuture doBind(final SocketAddress localAddress) {
//创建 channel并初始化
final ChannelFuture regFuture = initAndRegister();
final Channel channel = regFuture.channel();
if (regFuture.cause() != null) {
return regFuture;
}
//不能肯定register完成，因为register是丢到nio event loop里面执行去了。
if (regFuture.isDone()) {
// At this point we know that the registration was complete and successful.
ChannelPromise promise = channel.newPromise();
//进一步进行绑定操作（此处的绑定指的是将 channel 绑定 selector）
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);
//等着register完成后再通知再执行bind
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;
}
}
2.1 initAndRegister（） 初始化Channel并将Channel注册到Selector
final ChannelFuture initAndRegister() {
Channel channel = null;
try {
//通过工厂类创建Channel
channel = channelFactory.newChannel();
//完成Channel初始化
init(channel);
} catch (Throwable t) {
if (channel != null) {
// channel can be null if newChannel crashed (eg SocketException("too many open files"))
channel.unsafe().closeForcibly();
// as the Channel is not registered yet we need to force the usage of the GlobalEventExecutor
return new DefaultChannelPromise(channel, GlobalEventExecutor.INSTANCE).setFailure(t);
}
// as the Channel is not registered yet we need to force the usage of the GlobalEventExecutor
return new DefaultChannelPromise(new FailedChannel(), GlobalEventExecutor.INSTANCE).setFailure(t);
}
//开始register
ChannelFuture regFuture = config().group().register(channel);
if (regFuture.cause() != null) {
if (channel.isRegistered()) {
channel.close();
} else {
channel.unsafe().closeForcibly();
}
}
// If we are here and the promise is not failed, it's one of the following cases:
// 1) If we attempted registration from the event loop, the registration has been completed at this point.
// i.e. It's safe to attempt bind() or connect() now because the channel has been registered.
// 2) If we attempted registration from the other thread, the registration request has been successfully
// added to the event loop's task queue for later execution.
// i.e. It's safe to attempt bind() or connect() now:
// because bind() or connect() will be executed *after* the scheduled registration task is executed
// because register(), bind(), and connect() are all bound to the same thread.
return regFuture;
}
通过堆栈信息可以看到此方法由main线程进行调用
channelFactory.newChannel()通过DEBUG追踪，调用的 io.netty.channel.ReflectiveChannelFactory#newChannel,底层通过反射无参构造，创建的Channel
public T newChannel() {
try {
//反射创建channel
return constructor.newInstance();
} catch (Throwable t) {
throw new ChannelException("Unable to create Channel from class " + constructor.getDeclaringClass(), t);
}
}
·ChannelFactory 是一个接口类，只有一个抽象方法 newChannel() ，在ServerBootstrap.channel 方法执行时，赋值
io.netty.bootstrap.AbstractBootstrap#channel
io.netty.bootstrap.AbstractBootstrap#channelFactory(io.netty.bootstrap.ChannelFactory<? extends C>)
`NioServerSocketChannel 无参构造被调用的时候会执行 newSocket 方法
public NioServerSocketChannel() {
this(newSocket(DEFAULT_SELECTOR_PROVIDER));
}
参数 DEFAULT_SELECTOR_PROVIDER 被 SelectorProvider.provider() 赋值，不同的平台下 SelectorProvider 实现类不一样
最终通过 provider.openServerSocketChannel() 创建一个Channel
io.netty.channel.socket.nio.NioServerSocketChannel#newSocket
private static ServerSocketChannel newSocket(SelectorProvider provider) {
try {
/**
* Use the {@link SelectorProvider} to open {@link SocketChannel} and so remove condition in
* {@link SelectorProvider#provider()} which is called by each ServerSocketChannel.open() otherwise.
*
* See <a href="https://github.com/netty/netty/issues/2308">#2308</a>.
*/
return provider.openServerSocketChannel();
} catch (IOException e) {
throw new ChannelException(
"Failed to open a server socket.", e);
}
}
可以看到这里Netty的做法跟原生的Java NIO 是一样的
2.1.1 init（） 初始化Channel，并添加ServerBootstrapAcceptor 处理器
@Override
void init(Channel channel) {
//参数配置，后面再看 此时跳过·
setChannelOptions(channel, options0().entrySet().toArray(newOptionArray(0)), logger);
setAttributes(channel, attrs0().entrySet().toArray(newAttrArray(0)));
//此时Channel已经创建出来了，拿到pipeline，准备添加Handler
ChannelPipeline p = channel.pipeline();
final EventLoopGroup currentChildGroup = childGroup;
final ChannelHandler currentChildHandler = childHandler;
final Entry<ChannelOption<?>, Object>[] currentChildOptions =
childOptions.entrySet().toArray(newOptionArray(0));
final Entry<AttributeKey<?>, Object>[] currentChildAttrs = childAttrs.entrySet().toArray(newAttrArray(0));
//ChannelInitializer一次性、初始化handler:
//负责添加一个ServerBootstrapAcceptor handler，添加完后，自己就移除了:
//ServerBootstrapAcceptor handler： 负责接收客户端连接创建连接后，对连接的初始化工作。
p.addLast(new ChannelInitializer<Channel>() {
@Override
public void initChannel(final Channel ch) {
final ChannelPipeline pipeline = ch.pipeline();
//将ServerBootstrap.Handler（）方法设置的Handler添加到pipeline中
ChannelHandler handler = config.handler();
if (handler != null) {
pipeline.addLast(handler);
}
//获取一个EventLoop，并提交往pipeline中添加ServerBootstrapAcceptor Handler的任务
ch.eventLoop().execute(new Runnable() {
@Override
public void run() {
pipeline.addLast(new ServerBootstrapAcceptor(
ch, currentChildGroup, currentChildHandler, currentChildOptions, currentChildAttrs));
}
});
}
});
}
·ServerBootstrapAcceptor 继承自 ChannelInboundHandlerAdapter ，ServerBootstrapAcceptor作用就是当发生连接事件时与Channel建立连接。
需要注意的时，此时只是往pipeline中添加了一个Handler，并没有真正执行。
2.1.2 register(channel) 将Channel注册到EventLoop
ChannelFuture regFuture = config().group().register(channel);
io.netty.channel.SingleThreadEventLoop#register(io.netty.channel.ChannelPromise)
promise.channel().unsafe().register(this, promise);
io.netty.channel.AbstractChannel.AbstractUnsafe#register
@Override
public final void register(EventLoop eventLoop, final ChannelPromise promise) {
if (eventLoop == null) {
throw new NullPointerException("eventLoop");
}
if (isRegistered()) {
promise.setFailure(new IllegalStateException("registered to an event loop already"));
return;
}
if (!isCompatible(eventLoop)) {
promise.setFailure(
new IllegalStateException("incompatible event loop type: " + eventLoop.getClass().getName()));
return;
}
AbstractChannel.this.eventLoop = eventLoop;
//此时还是main线程，因此走false逻辑
if (eventLoop.inEventLoop()) {
register0(promise);
} else {
try {
//将任务交由EventLoop执行
eventLoop.execute(new Runnable() {
@Override
public void run() {
register0(promise);
}
});
} catch (Throwable t) {
logger.warn(
"Force-closing a channel whose registration task was not accepted by an event loop: {}",
AbstractChannel.this, t);
closeForcibly();
closeFuture.setClosed();
safeSetFailure(promise, t);
}
}
}
2.2.3 register0（）真正执行Channel注册
io.netty.channel.AbstractChannel.AbstractUnsafe#register0
private void register0(ChannelPromise promise) {
try {
// check if the channel is still open as it could be closed in the mean time when the register
// call was outside of the eventLoop
if (!promise.setUncancellable() || !ensureOpen(promise)) {
return;
}
boolean firstRegistration = neverRegistered;
//将Channel注册到Selector上
doRegister();
neverRegistered = false;
registered = true;
// Ensure we call handlerAdded(...) before we actually notify the promise. This is needed as the
// user may already fire events through the pipeline in the ChannelFutureListener.
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.
//server socket的注册不会走进下面if,server socket接受连接创建的socket可以走进去。因为accept后就active了。
if (isActive()) {
if (firstRegistration) {
pipeline.fireChannelActive();
} else if (config().isAutoRead()) {
// This channel was registered before and autoRead() is set. This means we need to begin read
// again so that we process inbound data.
//
// See https://github.com/netty/netty/issues/4805
beginRead();
}
}
} catch (Throwable t) {
// Close the channel directly to avoid FD leak.
closeForcibly();
closeFuture.setClosed();
safeSetFailure(promise, t);
}
}
doRegister 方法将调用父类AbstractNioChannel
io.netty.channel.nio.AbstractNioChannel#doRegister
protected void doRegister() throws Exception {
boolean selected = false;
//死循环注册Channel，0没有绑定事件，并将Channel作为附件，方便后续取出使用
for (;;) {
try {
logger.info("initial register： " + 0);
selectionKey = javaChannel().register(eventLoop().unwrappedSelector(), 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;
}
}
}
}
此处是一个死循环，确保一定会注册上 并且将Channel作为一个附件，方便取出使用。每个eventLoop都维护了一个Selector。
回到register0方法中，在将Channel注册到Selector上后，执行pipeline.invokeHandlerAddedIfNeeded();
io.netty.channel.DefaultChannelPipeline#callHandlerAddedForAllHandlers
private void callHandlerAddedForAllHandlers() {
final PendingHandlerCallback pendingHandlerCallbackHead;
synchronized (this) {
assert !registered;
// This Channel itself was registered.
registered = true;
pendingHandlerCallbackHead = this.pendingHandlerCallbackHead;
// Null out so it can be GC'ed.
this.pendingHandlerCallbackHead = null;
}
// This must happen outside of the synchronized(...) block as otherwise handlerAdded(...) may be called while
// holding the lock and so produce a deadlock if handlerAdded(...) will try to add another handler from outside
// the EventLoop.
PendingHandlerCallback task = pendingHandlerCallbackHead;
while (task != null) {
task.execute();
task = task.next;
}
}
此时将调用 init（）方法中，往pipeline中添加的ChannelInitializer.initChannel()。由此可知，ChannelInitializer.initChannel 只会被调用一次，就是在Channel实例化后，成功注册到Selector上。
此时Channel已经注册到Selector上了，因此重新获取一个EventLoop后，提交任务往pipeline中添加ServerBootstrapAcceptor Handler，负责Channel建立连接事件
在回到 register0 方法，执行完Handler后，将调用 safeSetSuccess(promise) 放promise中设置结果，进而触发promise的监听器，执行doBind0()方法
io.netty.bootstrap.AbstractBootstrap#doBind 
io.netty.bootstrap.AbstractBootstrap#doBind0
将任务交给EventLoop执行
io.netty.channel.AbstractChannel#bind(java.net.SocketAddress, io.netty.channel.ChannelPromise)
public ChannelFuture bind(SocketAddress localAddress, ChannelPromise promise) {
return pipeline.bind(localAddress, promise);
}
io.netty.channel.DefaultChannelPipeline#bind(java.net.SocketAddress, io.netty.channel.ChannelPromise)
@Override
public final ChannelFuture bind(SocketAddress localAddress, ChannelPromise promise) {
return tail.bind(localAddress, promise);
}
每个pipeline默认都有tail和head两个Handler，此时的bind方法将交由tail执行
io.netty.channel.AbstractChannelHandlerContext#bind(java.net.SocketAddress, io.netty.channel.ChannelPromise)
public ChannelFuture bind(final SocketAddress localAddress, final ChannelPromise promise) {
if (localAddress == null) {
throw new NullPointerException("localAddress");
}
if (isNotValidPromise(promise, false)) {
// cancelled
return promise;
}
final AbstractChannelHandlerContext next = findContextOutbound(MASK_BIND);
EventExecutor executor = next.executor();
//任务交由了EventLoop执行，因此走true逻辑
if (executor.inEventLoop()) {
next.invokeBind(localAddress, promise);
} else {
safeExecute(executor, new Runnable() {
@Override
public void run() {
next.invokeBind(localAddress, promise);
}
}, promise, null);
}
return promise;
}
invokeBind 方法调用链比较深，最终走到io.netty.channel.socket.nio.NioServerSocketChannel#doBind
`ServerSocketChannel继承自父类ServerSocketChannel，拿到的就是java的ServerSocketChannel，如果java版本大于7，最终通过ServerSocketChannelImpl.bind方法绑定端口号，并设置是否阻塞
执行doBind方法后，AbstractUnsafe将给EventLoop提交pipeline.fireChannelActive 任务。
io.netty.channel.AbstractChannel.AbstractUnsafe#bind
此时的pipeline中有三个Handler，分别是head，tail，acceptor。fireChannelActive 方法将调用他们的channelActive方法。
我们主要看 head.channelActive
io.netty.channel.DefaultChannelPipeline.HeadContext#channelActive
调用链比较深，最终会回到AbstractNioChannel
io.netty.channel.nio.AbstractNioChannel#doBeginRead
扫一扫
751
被折叠的 条评论
为什么被折叠?
到【灌水乐园】发言
