服务端启动bind过程分析
介绍之前先，给出服务端启动代码
public class NettyServer {
static EventLoopGroup boss = null;
static EventLoopGroup work = null;
public static void main(String[] args) throws Exception {
int port = 8080;
startServer(port);
}
private static void startServer(int port) throws Exception {
if (Epoll.isAvailable()) {
boss = new EpollEventLoopGroup();
work = new EpollEventLoopGroup();
} else {
boss = new NioEventLoopGroup();
work = new NioEventLoopGroup();
}
ServerBootstrap bootStrap = new ServerBootstrap();
bootStrap.group(boss, work)
.channel(NioServerSocketChannel.class)
.option(ChannelOption.SO_BACKLOG, 1024)// accept 队列长度
.option(ChannelOption.SO_REUSEADDR, true)
.option(ChannelOption.TCP_NODELAY, true)
.option(ChannelOption.SO_KEEPALIVE, true)
.childHandler(new ServerChannelInitializer());
bootStrap.bind(port).addListener(new ChannelFutureListener() {
@Override
public void operationComplete(ChannelFuture future) throws Exception {
if(future.isDone()) {
System.out.println("端口绑定成功:"+future.channel());
return ;
}
System.out.println("端口绑定失败");
}
});
Runtime.getRuntime().addShutdownHook(new Thread() {
@Override
public void run() {
boss.shutdownGracefully();
work.shutdownGracefully();
}
});
System.out.println(String.format("start start port:%s", port));
}
}
public class ServerChannelInitializer extends ChannelInitializer<SocketChannel> {
@Override
protected void initChannel(SocketChannel socketChannel) throws Exception {
ChannelPipeline pipeline = socketChannel.pipeline();
// 字符串解码和编码
pipeline.addLast("decoder", new StringDecoder());
pipeline.addLast("encoder", new StringEncoder());
// 服务器的逻辑
pipeline.addLast("handler", new DemoServerHandler());
}
}
大致解释上述步骤
1、创建boss 线程组和work线程组
boss：负责监听端口接收accept 事件。
work：处理read /write事件。
2、设置io模型NioServerSocketChannel。
3、设置服务端channel tcp 属性。
4、设置ChannelInitializer，这个主要是在每条连接过来之后，调用ChannelInitializer 里面的initChannel方法为每个
client 的channel 设置处理逻辑handler 。
接下来介绍bind 端口过程。也就是bootStrap.bind(port)过程
之后进入AbstractBootstrap bind(int inetPort )
public ChannelFuture bind(int inetPort) {
return bind(new InetSocketAddress(inetPort));
}
接下来看bind(new InetSocketAddress(inetPort));
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;
}
//忽略部分代码
}
这里我们只分析initAndRegister()方法，方法字面意思可以看出，初始化服务端channel 的参数和注册到reactor 线程，
这个过程还没有绑定端口，因此服务端口不可用。
final ChannelFuture initAndRegister() {
Channel channel = null;
try {
//创建服务端channel ，也就是最开始的.channel(NioServerSocketChannel.class)
// 然后通过反射方式初始化服务端channel ，这为什么要使用反射？不直接new 这样子
channel = channelFactory.newChannel();
// 初始化
init(channel);
} catch (Throwable t) {
if (channel != null) {
channel.unsafe().closeForcibly();
}
return new DefaultChannelPromise(channel, GlobalEventExecutor.INSTANCE).setFailure(t);
}
//选择boss 线程组里面的nioeventloop绑定到一个reactor 线程
ChannelFuture regFuture = config().group().register(channel);
if (regFuture.cause() != null) {
if (channel.isRegistered()) {
channel.close();
} else {
channel.unsafe().closeForcibly();
}
}
return regFuture;
}
1、 channelFactory.newChannel()：反射初始化serverchannel，也就是NIoServerSocketChannel.
2、 init(channel)：为创建好的channel 初始化参数
3、ChannelFuture regFuture = config().group().register(channel)：选择一个reactor 线程，把服务端channel和该线程
绑定。
NIoServerSocketChannel 初始化
private static final SelectorProvider DEFAULT_SELECTOR_PROVIDER = SelectorProvider.provider();
//忽略了其他代码
private static ServerSocketChannel newSocket(SelectorProvider provider) {
try {
//使用一个selectorprovider打开一个服务端channel
return provider.openServerSocketChannel();
} catch (IOException e) {
throw new ChannelException(
"Failed to open a server socket.", e);
}
}
public NioServerSocketChannel() {
// 反射过程，调用无参构造方法
this(newSocket(DEFAULT_SELECTOR_PROVIDER));
}
public NioServerSocketChannel(ServerSocketChannel channel) {
//调用父类保存要监听的事件类型SelectionKey.OP_ACCEPT
super(null, channel, SelectionKey.OP_ACCEPT);
// 把服务端的设置的参数封装在NioServerSocketChannelConfig
config = new NioServerSocketChannelConfig(this, javaChannel().socket());
}
最终我们直接看顶层类AbstractNioChannel
protected AbstractNioChannel(Channel parent, SelectableChannel ch, int readInterestOp) {
super(parent);
//传入服务端channel
this.ch = ch;
//保留服务端监听的事件类型SelectionKey.OP_ACCEPT
this.readInterestOp = readInterestOp;
try {
// 设置为no-block
ch.configureBlocking(false);
} catch (IOException e) {
try {
ch.close();
} catch (IOException e2) {
if (logger.isWarnEnabled()) {
logger.warn(
"Failed to close a partially initialized socket.", e2);
}
}
throw new ChannelException("Failed to enter non-blocking mode.", e);
}
}
在进入父类AbstractChannel，看构造方法
protected AbstractChannel(Channel parent) {
this.parent = parent;
id = newId();
unsafe = newUnsafe();
pipeline = newChannelPipeline();
}
AbstractChannel主要创建了netty 的几个重要的组件
1、Unsafe
2、Pipeline。
接下来进入init(channel)
void init(Channel channel) throws Exception {
final Map<ChannelOption<?>, Object> options = options0();
synchronized (options) {
channel.config().setOptions(options);
}
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(childOptions.size()));
}
synchronized (childAttrs) {
currentChildAttrs = childAttrs.entrySet().toArray(newAttrArray(childAttrs.size()));
}
p.addLast(new ChannelInitializer<Channel>() {
@Override
public void initChannel(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(
currentChildGroup, currentChildHandler, currentChildOptions, currentChildAttrs));
}
});
}
});
}
方法有点长，我们看主要的步骤：
p.addLast(new ChannelInitializer<Channel>() {
@Override
public void initChannel(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() {
//可以对比reactor 里面acceptor ，这里就是新连接到达之后，分发work 线程的链条
pipeline.addLast(new ServerBootstrapAcceptor(
currentChildGroup, currentChildHandler, currentChildOptions, currentChildAttrs));
}
});
}
});
这个时候，主要是往服务端的pipleline里面添加handler ，此时，服务端的handler 结构
image.png
接下来我们进入到initAndRegister里面
ChannelFuture regFuture = config().group().register(channel);
该方法就是启动work 线程组里面的reactor 线程，这个过程，只是启动，并没有绑定端口。
进入到MultithreadEventLoopGroup 的register 方法
publicChannelFuture register(Channel channel) {
returnnext().register(channel);
}
next() 方法就是选择一个NIoeventloop ，最后我们看NIOeventloop 里面的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;
}
//保留选择的NIOeventloop
AbstractChannel.this.eventLoop = eventLoop;
// 判断是否已经启动并且是不是当前线程在运行
if (eventLoop.inEventLoop()) {
register0(promise);
} else {
try {
1.因为是刚刚启动，该线程还没有运行，
2.直接执行 eventLoop.execute
eventLoop.execute(new Runnable() {
@Override
public void run() {
register0(promise);
}
});
} catch (Throwable t) {
}
}
}
分析一下eventLoop.execute(),也就是NIOeventloop ，这个方法在他的父类SingleThreadEventExecutor里面
public void execute(Runnable task) {
if (task == null) {
throw new NullPointerException("task");
}
//还是判断是否在当前线程运行，刚刚开始是false
boolean inEventLoop = inEventLoop();
if (inEventLoop) {
addTask(task);
} else {
// 启动reactor 线程，也就是nioeventloop 的run 方法。
startThread();
// 向启动好的reactor 线程提交非io task
addTask(task);
if (isShutdown() && removeTask(task)) {
reject();
}
}
if (!addTaskWakesUp && wakesUpForTask(task)) {
wakeup(inEventLoop);
}
}
```
执行上诉else 方法，首先看startThread();
```java
private void doStartThread() {
assert thread == null;
// executor 是Java 的线程池。初始化的地方在NIoeventLoopGroup 里面,
//ThreadPerTaskExecutor
//
executor.execute(new Runnable() {
@Override
public void run() {
thread = Thread.currentThread();
if (interrupted) {
thread.interrupt();
}
boolean success = false;
updateLastExecutionTime();
try {
// 开始reactor 线程，死循环
SingleThreadEventExecutor.this.run();
success = true;
} catch (Throwable t) {
logger.warn("Unexpected exception from an event executor: ", t);
} finally {
//忽略其他代码
}
});
}
到上面之后，服务端reactor 已经启动，这个时候在执行 addTask(task)，往reactor 线程里面提交非io task
，这个时候执行
eventLoop.execute(new Runnable() {
@Override
public void run() {
register0(promise);
}
});
里面的run方法，这个run'方法的启动是reactor线程触发的。
接下来我们分析 register0(promise);，这个方法完成端口绑定和事件的回调。
private void register0(ChannelPromise promise) {
try {
//忽略其他代码
boolean firstRegistration = neverRegistered;
// 完成channel 和selector 的注册绑定
doRegister();
neverRegistered = false;
registered = true;
// 回调init 方法里面ChannelInitializer
// ChannelInitializer 执行方法initChannel，把handler ServerBootstrapAcceptor 加入服务端链条
// 这个时候服务端链条为head--ServerBootstrapAcceptor--tail
pipeline.invokeHandlerAddedIfNeeded();
safeSetSuccess(promise);
// 通知链条里面handler 的方法channelRegistered
pipeline.fireChannelRegistered();
//服务端还没有绑定端口完成false
if (isActive()) {
if (firstRegistration) {
pipeline.fireChannelActive();
} else if (config().isAutoRead()) {
beginRead();
}
}
} catch (Throwable t) {
// 忽略其他代码
}
doRegister()，完成channel 和selector 的绑定。
protected void doRegister() throws Exception {
boolean selected = false;
for (;;) {
try {
//完成selector 和服务端channel 的绑定注册，这里的注册的感兴趣的为0，代表没有什么需要关心的事件，并加上附属对象this
//也就是NioeventLoop
selectionKey = javaChannel().register(eventLoop().selector, 0, this);
return;
} catch (CancelledKeyException e) {
//忽略其他代码
}
}
}
接下来我们看看此时服务端的链条情况
need-to-insert-img
该链条在新连接到达之后，会执行Accept 为初始化客户端channel 和参数，并把新连接和work 线程绑定selector 和注册read 、write 事件。
经过上面的步骤，reactor 线程已经启动完成，并且完成了服务端channel 和reactor 的绑定，但是此时还没有端口绑定和注册监听事件
我们回到AbstractBootstrap 的doBind()
private ChannelFuture doBind(final SocketAddress localAddress) {
// 忽然其他代码
//绑定selector 完成
if (regFuture.isDone()) {
ChannelPromise promise = channel.newPromise();
// 绑定端口
doBind0(regFuture, channel, localAddress, promise);
return promise;
}
}
private static void doBind0(
final ChannelFuture regFuture, final Channel channel,
final SocketAddress localAddress, final ChannelPromise promise) {
// 往reactor 线程里面提交非io 任务
channel.eventLoop().execute(new Runnable() {
@Override
public void run() {
if (regFuture.isSuccess()) {
// 进行端口绑定,最后到达链条tair 里面
channel.bind(localAddress, promise).addListener(ChannelFutureListener.CLOSE_ON_FAILURE);
} else {
promise.setFailure(regFuture.cause());
}
}
});
}
我们直接看 channel.bind(localAddress, promise).addListener(ChannelFutureListener.CLOSE_ON_FAILURE);
1、先到达AbstractChannel 里面的bind方法
public ChannelFuture bind(SocketAddress localAddress) {
// 执行服务端的pipeline,最后到tail里面
return pipeline.bind(localAddress);
}
// 链条tail 里面
public ChannelFuture bind(final SocketAddress localAddress, final ChannelPromise promise) {
if (localAddress == null) {
throw new NullPointerException("localAddress");
}
if (!validatePromise(promise, false)) {
// cancelled
return promise;
}
// 找到下一个outbound 方法，目前只有head
final AbstractChannelHandlerContext next = findContextOutbound();
EventExecutor executor = next.executor();
if (executor.inEventLoop()) {
// 调用head 方法invokeBind
next.invokeBind(localAddress, promise);
} else {
safeExecute(executor, new Runnable() {
@Override
public void run() {
next.invokeBind(localAddress, promise);
}
}, promise, null);
}
return promise;
}
查看链条里面Head 的方法invokeBind()
private void invokeBind(SocketAddress localAddress, ChannelPromise promise) {
if (invokeHandler()) {
try {
//调用链条里面bind方法
((ChannelOutboundHandler) handler()).bind(this, localAddress, promise);
} catch (Throwable t) {
notifyOutboundHandlerException(t, promise);
}
} else {
bind(localAddress, promise);
}
}
最后达到链条head 里面的bind方法，之后进入到unsafe 方法
public final void bind(final SocketAddress localAddress, final ChannelPromise promise) {
assertEventLoop();
if (!promise.setUncancellable() || !ensureOpen(promise)) {
return;
}
// 忽略其他代码
boolean wasActive = isActive();
try {
// 进入服务端channel :NIoServerSocketChannel
doBind(localAddress);
} catch (Throwable t) {
safeSetFailure(promise, t);
closeIfClosed();
return;
}
if (!wasActive && isActive()) {
// 提交非io 事件到reactor 线程
invokeLater(new Runnable() {
@Override
public void run() {
// 注册Accept 事件
pipeline.fireChannelActive();
}
});
}
safeSetSuccess(promise);
}
NioServerSocketChannel 方法doBind()
protected void doBind(SocketAddress localAddress) throws Exception {
if (PlatformDependent.javaVersion() >= 7) {
// 这里看着就比较熟悉了，Java 原生的绑定端口的方法
javaChannel().bind(localAddress, config.getBacklog());
} else {
javaChannel().socket().bind(localAddress, config.getBacklog());
}
}
经过上面最终到服务端channel doBind 方法完成端口绑定，接下来看如何注册Accept 事件，也就是代码
if (!wasActive && isActive()) {
// 提交非io 事件到reactor 线程
invokeLater(new Runnable() {
@Override
public void run() {
// 注册Accept 事件，pipeline 也就是服务端的目前里面链条结构为 head--accept-tail
pipeline.fireChannelActive();
}
});
}
最后到达AbstractNioChannel里面
protected void doBeginRead() throws Exception {
// Channel.read() or ChannelHandlerContext.read() was called
// 注册服务端channel 到对应reactor 里面的selector
final SelectionKey selectionKey = this.selectionKey;
if (!selectionKey.isValid()) {
return;
}
readPending = true;
final int interestOps = selectionKey.interestOps();
//最开始没有任何感兴趣事情==0
if ((interestOps & readInterestOp) == 0) {
//readInterestOp 是我们初始化的事件类型Accept 也就是16，这儿绑定accept 事件
selectionKey.interestOps(interestOps | readInterestOp);
}
}
总结：
服务端channel 初始化
NIoServerSocketChannel初始化，创建netty 主要组建unsafe 、pipeline
启动work reactor 线程注册channel
AbstractBootstrap 类里面initAndRegister方法对应代码块 ChannelFuture regFuture = config().group().register(channel);
绑定端口和注册accept 事件到reactor 线程
AbstractBootstrap 类里面 方法 doBind(finalSocketAddress localAddress)
对应代码块
doBind0(regFuture, channel, localAddress, promise);
网友评论