Netty
服务端一般如下面代码模式,简化了 NIO
编程的复杂性同时,并且借助于 Pipeline
模型,可以很简单的就构建出高性能、可扩展的应用程序。
public class DemoServer {
public static void main(String[] args) throws InterruptedException {
EventLoopGroup bossGroup = new NioEventLoopGroup();
EventLoopGroup workerGroup = new NioEventLoopGroup();
try {
ServerBootstrap serverBootstrap = new ServerBootstrap();
serverBootstrap.group(bossGroup, workerGroup)
.channel(NioServerSocketChannel.class)
.childHandler(new TestServerInitializer());
ChannelFuture channelFuture = serverBootstrap.bind(8899).sync();
channelFuture.channel().closeFuture().sync();
}finally {
bossGroup.shutdownGracefully();
workerGroup.shutdownGracefully();
}
}
}
下面我们就通过源码分析下隐藏在这些代码背后的逻辑,对 Netty
可以有个更加深刻的认识。
NioEventLoopGroup
Netty
中每个 Channel
都会通过注册方式,绑定到一个具体的 NioEventLoop
实例上, NioEventLoop
继承抽象类 SingleThreadEventLoop
,内部通过单个线程模式管理所有注册到它上面的 Channel
,负责这些 Channel
事件监听、事件处理等。 NioEventLoopGroup
内部包含一组 NioEventLoop
,好比 NioEventLoop
是用于管理 Channel
的其中一个线程,而 NioEventLoopGroup
则对应的是管理所有 Channel
的线程池。
上面创建了两个 NioEventLoopGroup
对象,一个是用来管理 NioServerSocketChannel
的,而另一个是用来管理客户端连接进来时创建的客户端对应的 NioSocketChannel
的。
通过跟踪 NioEventLoopGroup
构造过程,本身逻辑是比较简单,但是调用栈比较深,这里就不太方便代码展示,其大概完成事情可以用如下图描述:
NioEventLoopGroup
创建时,同时会创建三个元素: executor
、 chooser
和 child
:
-
child NioEventLoop Group NioEventLoop CPU核数*2 NioEventLoop
-
executor NioEventLoop executor
-
taskQueue NioEventLoop NioEventLoop taskQueue NioEventLoop taskQueue
-
rejectedHandler addTask() taskQueue
-
Selector NioEventLoop Selector Channel Selector SelectionKey NioEventLoop
-
executor NioEventLoop Channel executor
-
chooser NioEventLoopGroup NioEventLoop Channel chooser NioEventLoop
ServerBootStrap配置
下面我们来分析下如下代码作用:
ServerBootstrap serverBootstrap = new ServerBootstrap();
serverBootstrap.group(bossGroup, workerGroup)
.channel(NioServerSocketChannel.class)
.childHandler(new TestServerInitializer());
ServerBootStrap
是 Netty
使用的一个启动引导类,上面的代码主要是为后续 Netty
启动提供配置数据,本身比较简单:
-
group(bossGroup, workerGroup) Netty bossGroup NioServerSocketChannel OP_ACCEPT workerGroup NioSocketChannel
-
channel(NioServerSocketChannel.class)
:用于指定网络模型。 -
childHandler(new TestServerInitializer()) NioSocketChannel channel pipeline channel ChannelInitializer handler pipeline handler
bind
当执行到 serverBootstrap.bind(8899)
,则表示 Netty
开始进入真正的启动阶段。一路跟踪下来,会进入到 doBind()
方法中:
private ChannelFuture doBind(final SocketAddress localAddress) {
//创建NioServerSocketChannel -> pipeline添加ServerBootstrapAcceptor -> channel进行register,分配NioEventLoop
final ChannelFuture regFuture = initAndRegister();
final Channel channel = regFuture.channel();
if (regFuture.cause() != null) {
return regFuture;
}
/**
* register完成,则执行doBind0()进行server端口绑定
*/
if (regFuture.isDone()) {
ChannelPromise promise = channel.newPromise();
doBind0(regFuture, channel, localAddress, promise);
return promise;
} else {
final PendingRegistrationPromise promise = new PendingRegistrationPromise(channel);
//register还未完成,则添加listener,待注册完成再执行doBind0()进行server端口绑定
regFuture.addListener(new ChannelFutureListener() {
@Override
public void operationComplete(ChannelFuture future) throws Exception {
Throwable cause = future.cause();
if (cause != null) {
promise.setFailure(cause);
} else {
promise.registered();
doBind0(regFuture, channel, localAddress, promise);
}
}
});
return promise;
}
}
这个方法主要完成两件事:
-
initAndRegister() Channel Channel Channel NioEventLoop NioEventLoop
-
doBind0() initAndRegister() doBind0() initAndRegister() doBind0() regFuture.isDone() Channel addListener() listener doBind0() channel channel
initAndRegister
final ChannelFuture initAndRegister() {
Channel channel = null;
try {
/**
*
* 通过反射方式创建IO模型类型,具体类型有serverBootstrap.channel()方法指定,比如:NioServerSocketChannel
* NioServerSocketChannel创建时,构造方法中会触发创建jdk channel创建
* 同时会创建对应的配置类:NioServerSocketChannelConfig(tcp参数配置)
*/
channel = channelFactory.newChannel();
/**
* 初始化channel,由子类bootstrap或者serverBootStrap进行实现,可视为一个模板方法
* ServerBootStrap逻辑:options、attrs等初始化,同时向pipeline中添加一个InboundHandler:ServerBootstrapAcceptor
*
* new ServerBootstrapAcceptor(serverSocketChannel, currentChildGroup, currentChildHandler, currentChildOptions, currentChildAttrs)
* 这样NioServerSocketChannel接收到OP_ACCEPT事件时,就可以利用这些参数给代表客户端连接的SocketChannel初始化
*/
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
//还没有注册到线程池。使用默认线程GlobalEventExecutor
return new DefaultChannelPromise(new FailedChannel(), GlobalEventExecutor.INSTANCE).setFailure(t);
}
/**
* 将NioServerSocketChannel注册到Reactor主线程池上 ,即给当前创建的Channel分配一个NioEventLoop线程
*/
ChannelFuture regFuture = config().group().register(channel);
if (regFuture.cause() != null) {
if (channel.isRegistered()) {
channel.close();
} else {
channel.unsafe().closeForcibly();
}
}
return regFuture;
}
这个方法主要完成3件事:
-
channelFactory.newChannel() Channel serverBootstrap.channel() .channel(NioServerSocketChannel.class)
-
init(channel) Channel options attrs pipeline handler ServerBootstrapAcceptor
new ServerBootstrapAcceptor(serverSocketChannel, currentChildGroup, currentChildHandler, currentChildOptions, currentChildAttrs)
ServerBootstrapAcceptor
连接处理器是 Server
端非常重要的一个 InBound
类型的 handler
,当 NioServerSocketChannel
轮询 OP_ACCEPT
事件接收到客户端连接进来时,客户端连接各种设置等工作就是由这个 Acceptor
连接器完成。
-
config().group().register(channel) Channel NioEventLoop NioEventLoop channel NioEventLoop Selector
newChannel()
比较简单,这里就不展开了,核心点主要在于 init(channel)
和 register(channel)
这两个方法。
init
void init(Channel channel) throws Exception {
final Map<ChannelOption<?>, Object> options = options0();
synchronized (options) {
setChannelOptions(channel, options, logger);
}
final Map<AttributeKey<?>, Object> attrs = attrs0();
synchronized (attrs) {
for (Entry<AttributeKey<?>, Object> e: attrs.entrySet()) {
@SuppressWarnings("unchecked")
AttributeKey<Object> key = (AttributeKey<Object>) e.getKey();
channel.attr(key).set(e.getValue());
}
}
ChannelPipeline p = channel.pipeline();
final EventLoopGroup currentChildGroup = childGroup;
final ChannelHandler currentChildHandler = childHandler;
final Entry<ChannelOption<?>, Object>[] currentChildOptions;
final Entry<AttributeKey<?>, Object>[] currentChildAttrs;
synchronized (childOptions) {
currentChildOptions = childOptions.entrySet().toArray(newOptionArray(0));
}
synchronized (childAttrs) {
currentChildAttrs = childAttrs.entrySet().toArray(newAttrArray(0));
}
/**
* 给NIOServerChannel绑定的pipeline添加一个ChannelInitializer
*/
p.addLast(new ChannelInitializer<Channel>() {
@Override
public void initChannel(final Channel ch) throws Exception {
final ChannelPipeline pipeline = ch.pipeline();
ChannelHandler handler = config.handler();
if (handler != null) {
pipeline.addLast(handler);
}
ch.eventLoop().execute(new Runnable() {
@Override
public void run() {
pipeline.addLast(new ServerBootstrapAcceptor(
ch, currentChildGroup, currentChildHandler, currentChildOptions, currentChildAttrs));
}
});
}
});
}
这个方法主要完成2件事:
-
NioServerSocketChannel option attr
-
NioServerSocketChannel Pipeline ChannelInitializer ChannelInitializer pipeline handler NioServerSocketChannel handler ServerBootstrapAcceptor
顺便我们来看下通过 pipeline.addLast()
方式向 pipeline
添加 handler
逻辑:
public final ChannelPipeline addLast(EventExecutorGroup group, String name, ChannelHandler handler) {
final AbstractChannelHandlerContext newCtx;
synchronized (this) {
checkMultiplicity(handler);
newCtx = newContext(group, filterName(name, handler), handler);
addLast0(newCtx);
//当前Channel还未注册,需要先封装成PendingHandlerAddedTask,并链表方式挂载到Pipeline的pendingHandlerCallbackHead变量下,待后续注册完成后再回调
if (!registered) {
newCtx.setAddPending();
callHandlerCallbackLater(newCtx, true);
return this;
}
//注册完成,且当前线程和Channel绑定线程不是同一个,则用Channel的绑定线程执行
EventExecutor executor = newCtx.executor();
if (!executor.inEventLoop()) {
callHandlerAddedInEventLoop(newCtx, executor);
return this;
}
}
callHandlerAdded0(newCtx);
return this;
}
pipeline
是一个双向链表,刚创建完成时默认有两个节点: head
和 tail
,如下图:
执行 p.addLast(new ChannelInitializer())
后是如下图:
如上面代码, pipeline
不是直接将 handler
添加进来,而是封装成 handlerContext
。执行 addLast0(newCtx)
将 handler
对应的 HandlerContext
添加进来后,正常情况下这时需要回调 handler#handlerAdded()
方法。 handler#handlerAdded()
执行是需要在 channel
注册的 NioEventLoop
线程中执行才行,所以有 if (!executor.inEventLoop())
这个判断。但是,当前是在主线程 main
中,且 channel
因为还没有注册完成,所以当前 channel
和 NioEventLoop
根本就还没有绑定到一起,所以是没法执行的,这里会进入 if (!registered)
流程:将 handlerContext
封装成一个 PendingHandlerAddedTask
实例,先挂载到 pipeline
的 pendingHandlerCallbackHead
全局变量下,待后续 channel
注册完成后再来处理 handler#handlerAdded()
。还有个问题,如果添加多个 handler
, PendingHandlerAddedTask
有个 next
,可以把它们串成一个链表即可。
register
这样,我们把 init()
方法的主要逻辑基本都分析完成了,现在我们再回过头看下 initAndRegister
方法中另外一个重要逻辑: config().group().register(channel)
。这里的 config().group()
就是获取的是之前传入的用于处理 server
端线程组: EventLoopGroup bossGroup = new NioEventLoopGroup()
。
NioEventLoopGroup#register()
第一步就是使用 chooser
选取一个其管理的 NioEventLoop
,默认选取策略很简单,就是使用一个递增序列 idx
,然后和数组长度取模即可:
executors[idx.getAndIncrement() & executors.length - 1]
选取好 NioEventLoop
后,调用 NioEventLoop#register(channel)
方法, NioEventLoop#register(channel)
方法又会调用 channle
的 Unsafe
对象的 register
进行处理,并把自己即 NioEventLoop
作为参数传入:
promise.channel().unsafe().register(this, promise);
Channel
创建时同时创建一个 UnSafe
对象,主要用于处理与 java
底层 socket
相关操作。
所以, register()
方法跑了一圈最后还是在 channel
中的 Unsafe#register()
方法中进行处理, NioEventLoopGroup
只是利用 chooser
选取一个 NioEventLoop
作为参数传入到 register()
方法中。
我们来看下 Unsafe#register()
方法做了哪些事情。
1、后面这个 eventLoop
就是将刚才利用 chooser
选取的 NioEventLoop
,通过赋值给 channel
的 eventLoop
字段上,即完成了 channel
和 NioEventLoop
的关联;
AbstractChannel.this.eventLoop = eventLoop;
2、调用 register0()
方法, register0()
方法需要在 NioEventLoop
线程中执行才行,所以这里也使用 if (eventLoop.inEventLoop())
判断下,当前是主线程 main
,所以会进入到 else
逻辑处理中,把执行逻辑封装成任务提交到 NioEventLoop
的任务队列 taskQueue
中:
if (eventLoop.inEventLoop()) {
register0(promise);
} else {
try {
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);
}
}
eventLoop.execute()
方法中处理不只是简单将 task
放入到 taskQueue
中,我们来看下其还做了哪些事:
public void execute(Runnable task) {
boolean inEventLoop = inEventLoop();
addTask(task);
if (!inEventLoop) {
startThread();
if (isShutdown()) {
boolean reject = false;
try {
if (removeTask(task)) {
reject = true;
}
} catch (UnsupportedOperationException e) {
}
if (reject) {
reject();
}
}
}
if (!addTaskWakesUp && wakesUpForTask(task)) {
wakeup(inEventLoop);
}
}
这个方法主要完成3件事:
-
addTask(task) register0() taskQueue taskQueue
-
startThread() NioEventLoop channel startThread() if (!inEventLoop) NioEventLoop inEventLoop false NioEventLoop selector.select() processSelectedKeys() runAllTasks()
-
wakeup() NioEventLoop NioEventLoop selector.select(timeout) wakeup() NioEventLoop select()
eventLoop.execute()
分析完成后, register0()
方法任务已被添加到 taskQueue
中,然后启动 NioEventLoop
线程开始干活,最后通过 wakeup()
唤醒 NioEventLoop
让其去处理 taskQueue
中的任务,所以,这时我们需要再回头看下 register0()
方法。
private void register0(ChannelPromise promise) {
try {
if (!promise.setUncancellable() || !ensureOpen(promise)) {
return;
}
boolean firstRegistration = neverRegistered;
doRegister();
neverRegistered = false;
registered = true;
pipeline.invokeHandlerAddedIfNeeded();
safeSetSuccess(promise);
pipeline.fireChannelRegistered();
if (isActive()) {//这里实际返回false,channelActive()不会在这里触发
if (firstRegistration) {
pipeline.fireChannelActive();
} else if (config().isAutoRead()) {
beginRead();
}
}
} catch (Throwable t) {
closeForcibly();
closeFuture.setClosed();
safeSetFailure(promise, t);
}
}
这个方法主要完成3件事:
-
doRegister() java api channel selector javaChannel().register(eventLoop().unwrappedSelector(), 0, this)
-
channel NioEventLoop Selector
-
SelectionKey=0 channel selector OP_ACCEPT
-
this attachment this server NioServerSocketChannel selector OP_ACCEPT NioServerSocketChannel accept()
-
pipeline.invokeHandlerAddedIfNeeded() pipeline handler channel handler PendingHandlerAddedTask pipeline.pendingHandlerCallbackHead channel pendingHandlerCallbackHead handler#handlerAdded() ChannelInitializer#handlerAdded() initChannel() handler ServerBootstrapAcceptor pipeline ChannelInitializer pipeline
p.addLast(new ChannelInitializer<Channel>() {
@Override
public void initChannel(final Channel ch) throws Exception {
final ChannelPipeline pipeline = ch.pipeline();
ChannelHandler handler = config.handler();
if (handler != null) {
pipeline.addLast(handler);
}
ch.eventLoop().execute(new Runnable() {
@Override
public void run() {
pipeline.addLast(new ServerBootstrapAcceptor(
ch, currentChildGroup, currentChildHandler, currentChildOptions, currentChildAttrs));
}
});
}
});
-
pipeline.fireChannelRegistered()
:回调handler#channelRegistered()
方法; -
pipeline.fireChannelActive() channel if (isActive()) channel isActive() true
总结
分析到这里, doBind()
方法中两个重要方法: initAndRegister()
和 doBind0()
的第一个方法已全部分析完成, initAndRegister()
还是完成了相当多的任务,其核心逻辑总结下:创建 NioServerSocketChannel
,然后进行各种配置初始化,最重要的一步是把 channel
注册到 NioEventLoop
上, NioEventLoop
采用单线程模式轮询事件、处理事件。 handler
回调方法: handlerAdded()
和 channelRegistered()
也会在上面执行过程中被触发调用。
长按识别关注, 持续输出原创
以上就是本文的全部内容,希望对大家的学习有所帮助,也希望大家多多支持 码农网
猜你喜欢:- kubelet 分析源码:启动流程
- 【zookeeper源码】启动流程详解
- View绘制流程源码分析
- gorm查询流程源码分析
- ReactNative源码解析-启动流程
- Android 系统源码-1:Android 系统启动流程源码分析
本站部分资源来源于网络,本站转载出于传递更多信息之目的,版权归原作者或者来源机构所有,如转载稿涉及版权问题,请联系我们。