Netty ServerBootstrap加载启动过程源码分析

Netty一般的server创建代码

public class NettyServer {
// 日志
private Logger log = LoggerFactory.getLogger(getClass());

// 端口号
@Value("${netty.port}")
private int port;

// 启动服务器方法
public void run() {
EventLoopGroup bossGroup = new NioEventLoopGroup();
EventLoopGroup workerGroup = new NioEventLoopGroup();
try {
// 实例化一个空的serverBootstrap
ServerBootstrap serverBootstrap = new ServerBootstrap();
// 配是group，内部做了不为null检查，
// 并将bossGroup赋给group属性，workerGroup赋给childGroup属性
serverBootstrap.group(bossGroup, workerGroup);
// 反射实例化NioServerSocketChannel类型的ChannelFactory
serverBootstrap.channel(NioServerSocketChannel.class);
// 服务端处理客户端连接请求是顺序处理的，所以同一时间只能处理一个客户端连接，
// 多个客户端来的时候，服务端将不能处理的客户端连接请求放在队列中等待处理，
// backlog参数指定了队列的大小
serverBootstrap.option(ChannelOption.SO_BACKLOG, 1024);
// 配置handler
serverBootstrap.handler(new LoggingHandler(LogLevel.INFO));
// 配置childOption，child开头的属性只有在ServerBootstrap中才有，
// 客户端Bootstrap没有
serverBootstrap.childOption(ChannelOption.TCP_NODELAY, true);
serverBootstrap.childOption(ChannelOption.SO_KEEPALIVE, true);
serverBootstrap.childHandler(new NettyServerInitializer());

// 绑定端口,开始接收进来的连接，这里才是真正使用上述配置的地方
ChannelFuture channelFuture = serverBootstrap.bind(port).sync();
log.info("netty服务启动: [port:" + port + "]");
// 等待服务器socket关闭
channelFuture.channel().closeFuture().sync();
} catch (Exception e) {
log.error("netty服务启动异常-" + e.getMessage());
} finally {
bossGroup.shutdownGracefully();
workerGroup.shutdownGracefully();
}
}

}

上面的一大半都是配置加载过程，启动实在bind方法中

// bind方法实际调用的如下方法，先进行参数验证，包括group，channelFactory等
public ChannelFuture bind(SocketAddress localAddress) {
validate();
if (localAddress == null) {
throw new NullPointerException("localAddress");
}
return doBind(localAddress);
}

// 最终的实现是doBind
private ChannelFuture doBind(final SocketAddress localAddress) {
// 初始化和注册
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();
// 最后一步是判断初始化成功，将结束监听器注册到channel上
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);
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方法
final ChannelFuture initAndRegister() {
Channel channel = null;
try {
// new一个NioServerSocketChannel类型的channel
channel = channelFactory.newChannel();
// 核心的init方法，见下面
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);
}

ChannelFuture regFuture = config().group().register(channel);
if (regFuture.cause() != null) {
if (channel.isRegistered()) {
channel.close();
} else {
channel.unsafe().closeForcibly();
}
}
return regFuture;
}

// init是调用ServerBootstrap的init，上半部分都是基于父类的属性为自身初始化属性
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();
// 将config中的handler添加到pipline
ChannelHandler handler = config.handler();
if (handler != null) {
pipeline.addLast(handler);
}
ch.eventLoop().execute(new Runnable() {
@Override
public void run() {
// 在server端收到请求时才会新起线程将child开头的的配置注册到pipline上
pipeline.addLast(new ServerBootstrapAcceptor(
currentChildGroup, currentChildHandler, currentChildOptions, currentChildAttrs));
}
});
}
});
}