【死磕Netty】-----服务端启动过程分析

public class NettyServer {

public void bind(int port){
// 创建EventLoopGroup
EventLoopGroup bossGroup = new NioEventLoopGroup();        //创建BOSS线程组 用于服务端接受客户端的连接
EventLoopGroup workerGroup = new NioEventLoopGroup();      //创建WORK线程组 用于进行SocketChannel的网络读写

try {
// 创建ServerBootStrap实例
// ServerBootstrap 用于启动NIO服务端的辅助启动类，目的是降低服务端的开发复杂度
ServerBootstrap b = new ServerBootstrap();
// 绑定Reactor线程池
b.group(bossGroup, workerGroup)
// 设置并绑定服务端Channel
// 指定所使用的NIO传输的Channel
.channel(NioServerSocketChannel.class)
.option(ChannelOption.SO_BACKLOG, 1024)
.handler(new LoggingServerHandler())
.childHandler(new ChannelInitializer(){

@Override
protected void initChannel(Channel ch) throws Exception {
//do something
}
});

// 绑定端口，同步等待成功
ChannelFuture future = b.bind(port).sync();
// 等待服务端监听端口关闭
future.channel().closeFuture().sync();

} catch (InterruptedException e) {
e.printStackTrace();
} finally {
// 优雅地关闭
bossGroup.shutdownGracefully();
workerGroup.shutdownGracefully();
}
}

private class LoggingServerHandler extends ChannelInboundHandlerAdapter{
@Override
public void channelActive(ChannelHandlerContext ctx) throws Exception {
System.out.println("loggin-channelActive");
}

@Override
public void channelRegistered(ChannelHandlerContext ctx) throws Exception {
System.out.println("loggin-channelRegistered");
}

@Override
public void handlerAdded(ChannelHandlerContext ctx) throws Exception {
System.out.println("loggin-handlerAdded");
}
}

public static void main(String[] args){
new NettyServer().bind(8899);
}
}

服务端创建流程

服务端源码分析

EventLoopGroup bossGroup = new NioEventLoopGroup();
EventLoopGroup workerGroup = new NioEventLoopGroup();

ServerBootstrap b = new ServerBootstrap();

group

channel

option

childHandler

bind

group()

b.group(bossGroup, workerGroup)

public ServerBootstrap group(EventLoopGroup parentGroup, EventLoopGroup childGroup) {
super.group(parentGroup);        // 绑定boosGroup
if (childGroup == null) {
throw new NullPointerException("childGroup");
}
if (this.childGroup != null) {
throw new IllegalStateException("childGroup set already");
}
this.childGroup = childGroup;    // 绑定workerGroup
return this;
}

public B group(EventLoopGroup group) {
if (group == null) {
throw new NullPointerException("group");
}
if (this.group != null) {
throw new IllegalStateException("group set already");
}
this.group = group;
return (B) this;
}

.channel(NioServerSocketChannel.class)

ServerBootstrap.channel

public B channel(Class<? extends C> channelClass) {
if (channelClass == null) {
throw new NullPointerException("channelClass");
}
return channelFactory(new ReflectiveChannelFactory<C>(channelClass));
}

channelFactory()

public B channelFactory(io.netty.channel.ChannelFactory<? extends C> channelFactory) {
return channelFactory((ChannelFactory<C>) channelFactory);
}

public B channelFactory(ChannelFactory<? extends C> channelFactory) {
if (channelFactory == null) {
throw new NullPointerException("channelFactory");
}
if (this.channelFactory != null) {
throw new IllegalStateException("channelFactory set already");
}

this.channelFactory = channelFactory;
return (B) this;
}

public class ReflectiveChannelFactory<T extends Channel> implements ChannelFactory<T> {

private final Class<? extends T> clazz;

public ReflectiveChannelFactory(Class<? extends T> clazz) {
if (clazz == null) {
throw new NullPointerException("clazz");
}
this.clazz = clazz;
}
//需要创建 channel 的时候，该方法将被调用
@Override
public T newChannel() {
try {
// 反射创建对应 channel
return clazz.newInstance();
} catch (Throwable t) {
throw new ChannelException("Unable to create Channel from class " + clazz, t);
}
}

@Override
public String toString() {
return StringUtil.simpleClassName(clazz) + ".class";
}
}

option()

.option(ChannelOption.SO_BACKLOG, 1024)

option()

public <T> B option(ChannelOption<T> option, T value) {
if (option == null) {
throw new NullPointerException("option");
}
if (value == null) {
synchronized (options) {
options.remove(option);
}
} else {
synchronized (options) {
options.put(option, value);
}
}
return (B) this;
}

private final Map<ChannelOption<?>, Object> options = new LinkedHashMap<ChannelOption<?>, Object>();

.handler(new LoggingServerHandler())
.childHandler(new ChannelInitializer(){
//省略代码
})

handler()

childHandler()

// AbstractBootstrap
public B handler(ChannelHandler handler) {
if (handler == null) {
throw new NullPointerException("handler");
}
this.handler = handler;
return (B) this;
}

// ServerBootstrap
public ServerBootstrap childHandler(ChannelHandler childHandler) {
if (childHandler == null) {
throw new NullPointerException("childHandler");
}
this.childHandler = childHandler;
return this;
}

ChannelFuture future = b.bind(port).sync();

bind()

public ChannelFuture bind(int inetPort) {
return bind(new InetSocketAddress(inetPort));
}

public ChannelFuture bind(SocketAddress localAddress) {
validate();
if (localAddress == null) {
throw new NullPointerException("localAddress");
}
return doBind(localAddress);
}

validate()

doBind()

private ChannelFuture doBind(final SocketAddress localAddress) {
// 初始化并注册一个Channel
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绑定
doBind0(regFuture, channel, localAddress, promise);
return promise;
} else {
// Registration future is almost always fulfilled already, but just in case it's not.
final AbstractBootstrap.PendingRegistrationPromise promise = new AbstractBootstrap.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;
}
}

initAndRegister()

regFuture.cause()

regFuture.isDone()

initAndRegister()

doBind0

initAndRegister()

initAndRegister()

operationComplete()

operationComplete()

doBind0

doBind()

initAndRegister()

final ChannelFuture initAndRegister() {
Channel channel = null;
try {
// 新建一个Channel
channel = channelFactory.newChannel();
// 初始化Channel
init(channel);
} catch (Throwable t) {
if (channel != null) {
channel.unsafe().closeForcibly();
}
return new DefaultChannelPromise(channel, GlobalEventExecutor.INSTANCE).setFailure(t);
}

// /向EventLoopGroup中注册一个channel
ChannelFuture regFuture = config().group().register(channel);
if (regFuture.cause() != null) {
if (channel.isRegistered()) {
channel.close();
} else {
channel.unsafe().closeForcibly();
}
}
return regFuture;
}

newChannel()

.channel(NioServerSocketChannel.class)

public NioServerSocketChannel() {
this(newSocket(DEFAULT_SELECTOR_PROVIDER));
}

newSocket()

SelectorProvider.openServerSocketChannel()

public NioServerSocketChannel(ServerSocketChannel channel) {
super(null, channel, SelectionKey.OP_ACCEPT);
config = new NioServerSocketChannelConfig(this, javaChannel().socket());
}

// AbstractNioMessageChannel
protected AbstractNioMessageChannel(Channel parent, SelectableChannel ch, int readInterestOp) {
super(parent, ch, readInterestOp);
}

// AbstractNioChannel
protected AbstractNioChannel(Channel parent, SelectableChannel ch, int readInterestOp) {
super(parent);
this.ch = ch;
this.readInterestOp = readInterestOp;
try {
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();
}

super()

unsafe = newUnsafe()

pipeline = newChannelPipeline()

ch.configureBlocking(false);

this.readInterestOp = readInterestOp;

config = new NioServerSocketChannelConfig(this, javaChannel().socket())

channel = channelFactory.newChannel()

channelFactory.newChannel();

init()

void init(Channel channel) throws Exception {
// 设置配置的option参数
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());
}
}

// 获取绑定的pipeline
ChannelPipeline p = channel.pipeline();

// 准备child用到的4个part
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()));
}

// 为NioServerSocketChannel的pipeline添加一个初始化Handler,
// 当NioServerSocketChannel在EventLoop注册成功时，该handler的init方法将被调用
p.addLast(new ChannelInitializer<Channel>() {
@Override
public void initChannel(Channel ch) throws Exception {
final ChannelPipeline pipeline = ch.pipeline();
ChannelHandler handler = config.handler();
//如果用户配置过Handler
if (handler != null) {
pipeline.addLast(handler);
}

ch.eventLoop().execute(new Runnable() {
@Override
public void run() {
// 为NioServerSocketChannel的pipeline添加ServerBootstrapAcceptor处理器
// 该Handler主要用来将新创建的NioSocketChannel注册到EventLoopGroup中
pipeline.addLast(new ServerBootstrapAcceptor(
currentChildGroup, currentChildHandler, currentChildOptions, currentChildAttrs));
}
});
}
});
}

// /向EventLoopGroup中注册一个channel
ChannelFuture regFuture = config().group().register(channel);

initAndRegister()

group()

register()

public ChannelFuture register(Channel channel) {
return next().register(channel);
}

next()

register()

public ChannelFuture register(Channel channel) {
return register(new DefaultChannelPromise(channel, this));
}

public ChannelFuture register(ChannelPromise promise) {
ObjectUtil.checkNotNull(promise, "promise");
promise.channel().unsafe().register(this, promise);
return promise;
}

register()

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;

// 必须要保证注册是由该EventLoop发起的
if (eventLoop.inEventLoop()) {
register0(promise);        // 注册
} else {
// 如果不是单独封装成一个task异步执行
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);
}
}
}

isRegistered()

eventLoop.inEventLoop()

register0()

register0()

register0()

private void register0(ChannelPromise promise) {
try {
// 确保 Channel 处于 open
if (!promise.setUncancellable() || !ensureOpen(promise)) {
return;
}
boolean firstRegistration = neverRegistered;

// 真正的注册动作
doRegister();

neverRegistered = false;
registered = true;

pipeline.invokeHandlerAddedIfNeeded();
safeSetSuccess(promise);        //设置注册结果为成功

pipeline.fireChannelRegistered();

if (isActive()) {
//如果是首次注册,发起 pipeline 的 fireChannelActive
if (firstRegistration) {
pipeline.fireChannelActive();
} else if (config().isAutoRead()) {
beginRead();
}
}
} catch (Throwable t) {
closeForcibly();
closeFuture.setClosed();
safeSetFailure(promise, t);
}
}

doRegister()

fireChannelActive()

protected void doRegister() throws Exception {
boolean selected = false;
for (;;) {
try {
// 注册到NIOEventLoop的Selector上
selectionKey = javaChannel().register(eventLoop().selector, 0, this);
return;
} catch (CancelledKeyException e) {
if (!selected) {
eventLoop().selectNow();
selected = true;
} else {
throw e;
}
}
}
}

SelectionKey.interestOps(int ops)

doRegister()

pipeline.invokeHandlerAddedIfNeeded();

loggin-handlerAdded

safeSetSuccess(promise)

pipeline.fireChannelRegistered();

loggin-channelRegistered

public final ChannelPipeline fireChannelRegistered() {
AbstractChannelHandlerContext.invokeChannelRegistered(head);
return this;
}

static void invokeChannelRegistered(final AbstractChannelHandlerContext next) {
EventExecutor executor = next.executor();
if (executor.inEventLoop()) {
next.invokeChannelRegistered();
} else {
executor.execute(new Runnable() {
@Override
public void run() {
next.invokeChannelRegistered();
}
});
}
}

next.invokeChannelRegistered()

private void invokeChannelRegistered() {
if (invokeHandler()) {
try {
((ChannelInboundHandler) handler()).channelRegistered(this);
} catch (Throwable t) {
notifyHandlerException(t);
}
} else {
fireChannelRegistered();
}
}

invokeChannelRegistered()

handler(new LoggingServerHandler()

channelRegistered()

loggin-channelRegistered

initAndRegister() (final ChannelFuture regFuture = initAndRegister();)

init(channel)

initAndRegister()

doBind(final SocketAddress localAddress)

doBind(final SocketAddress localAddress)

initAndRegister()

regFuture.isDone()

doBind0()

initAndRegister()

operationComplete()

doBind0()

doBind0()

private static void doBind0(
final ChannelFuture regFuture, final Channel channel,
final SocketAddress localAddress, final ChannelPromise promise) {

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());
}
}
});
}

doBind0()

execute()

public void execute(Runnable task) {
if (task == null) {
throw new NullPointerException("task");
}

boolean inEventLoop = inEventLoop();
if (inEventLoop) {
addTask(task);
} else {
startThread();
addTask(task);
if (isShutdown() && removeTask(task)) {
reject();
}
}

if (!addTaskWakesUp && wakesUpForTask(task)) {
wakeup(inEventLoop);
}
}

inEventLoop()

addTask()

addTask()

addTask()

protected void addTask(Runnable task) {
if (task == null) {
throw new NullPointerException("task");
}
if (!offerTask(task)) {
reject(task);
}
}

offerTask()

final boolean offerTask(Runnable task) {
if (isShutdown()) {
reject();
}
return taskQueue.offer(task);
}

channel.bind(localAddress, promise).addListener(ChannelFutureListener.CLOSE_ON_FAILURE);

bind()

public ChannelFuture bind(SocketAddress localAddress, ChannelPromise promise) {
return pipeline.bind(localAddress, promise);
}

public final ChannelFuture bind(SocketAddress localAddress, ChannelPromise promise) {
return tail.bind(localAddress, promise);
}

final AbstractChannelHandlerContext tail;

bind()

bind()

public ChannelFuture bind(final SocketAddress localAddress, final ChannelPromise promise) {
if (localAddress == null) {
throw new NullPointerException("localAddress");
}
if (!validatePromise(promise, false)) {
// cancelled
return promise;
}

final AbstractChannelHandlerContext next = findContextOutbound();
EventExecutor executor = next.executor();
if (executor.inEventLoop()) {
next.invokeBind(localAddress, promise);
} else {
safeExecute(executor, new Runnable() {
@Override
public void run() {
next.invokeBind(localAddress, promise);
}
}, promise, null);
}
return promise;
}

findContextOutbound()

private AbstractChannelHandlerContext findContextOutbound() {
AbstractChannelHandlerContext ctx = this;
do {
ctx = ctx.prev;
} while (!ctx.outbound);
return ctx;
}

invokeBind()

private void invokeBind(SocketAddress localAddress, ChannelPromise promise) {
if (invokeHandler()) {
try {
((ChannelOutboundHandler) handler()).bind(this, localAddress, promise);
} catch (Throwable t) {
notifyOutboundHandlerException(t, promise);
}
} else {
bind(localAddress, promise);
}
}

handler()

bind()

public void bind(
ChannelHandlerContext ctx, SocketAddress localAddress, ChannelPromise promise)
throws Exception {
unsafe.bind(localAddress, promise);
}

unsafe = pipeline.channel().unsafe();

bind()

public final void bind(final SocketAddress localAddress, final ChannelPromise promise) {
assertEventLoop();

if (!promise.setUncancellable() || !ensureOpen(promise)) {
return;
}

if (Boolean.TRUE.equals(config().getOption(ChannelOption.SO_BROADCAST)) &&
localAddress instanceof InetSocketAddress &&
!((InetSocketAddress) localAddress).getAddress().isAnyLocalAddress() &&
!PlatformDependent.isWindows() && !PlatformDependent.isRoot()) {

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) {
safeSetFailure(promise, t);
closeIfClosed();
return;
}

if (!wasActive && isActive()) {
invokeLater(new Runnable() {
@Override
public void run() {
pipeline.fireChannelActive();
}
});
}

safeSetSuccess(promise);
}

doBind()

protected void doBind(SocketAddress localAddress) throws Exception {
if (PlatformDependent.javaVersion() >= 7) {
javaChannel().bind(localAddress, config.getBacklog());
} else {
javaChannel().socket().bind(localAddress, config.getBacklog());
}
}

javaChannel()

bind()

public ServerSocketChannel bind(SocketAddress var1, int var2) throws IOException {
Object var3 = this.lock;
synchronized(this.lock) {
if(!this.isOpen()) {
throw new ClosedChannelException();
} else if(this.isBound()) {
throw new AlreadyBoundException();
} else {
InetSocketAddress var4 = var1 == null?new InetSocketAddress(0):Net.checkAddress(var1);
SecurityManager var5 = System.getSecurityManager();
if(var5 != null) {
var5.checkListen(var4.getPort());
}

NetHooks.beforeTcpBind(this.fd, var4.getAddress(), var4.getPort());
Net.bind(this.fd, var4.getAddress(), var4.getPort());
Net.listen(this.fd, var2 < 1?50:var2);
Object var6 = this.stateLock;
synchronized(this.stateLock) {
this.localAddress = Net.localAddress(this.fd);
}

return this;
}
}
}

参考资料