java netty之ServerBootstrap的啓動

時間 2019-11-10

標籤 java netty serverbootstrap 啓動欄目 Java 简体版

原文原文鏈接

經過前面的幾篇文章，對整個netty部分的架構已經運行原理都有了必定的瞭解，那麼這篇文章來分析一個常常用到的類：ServerBootstrap，通常對於服務器端的編程它用到的都還算是比較的多。。看一看它的初始化，以及它的運行原理。。。java

首先咱們仍是引入一段代碼，經過分析這段代碼來分析ServerBootstrap的運行。。。git

[java] view plain copy

EventLoopGroup bossGroup = new NioEventLoopGroup(); //這個是用於serversocketchannel的eventloop
EventLoopGroup workerGroup = new NioEventLoopGroup(); //這個是用於處理accept到的channel
try {
ServerBootstrap b = new ServerBootstrap(); //構建serverbootstrap對象
b.group(bossGroup, workerGroup); //設置時間循環對象，前者用來處理accept事件，後者用於處理已經創建的鏈接的io
b.channel(NioServerSocketChannel.class); //用它來創建新accept的鏈接，用於構造serversocketchannel的工廠類
b.childHandler(new ChannelInitializer<SocketChannel>(){ //爲accept channel的pipeline預添加的inboundhandler
@Override //當新鏈接accept的時候，這個方法會調用
protected void initChannel(SocketChannel ch) throws Exception {
// TODO Auto-generated method stub
ch.pipeline().addLast(new MyChannelHandler()); //爲當前的channel的pipeline添加自定義的處理函數
}
});
//bind方法會建立一個serverchannel，而且會將當前的channel註冊到eventloop上面，
//會爲其綁定本地端口，並對其進行初始化，爲其的pipeline加一些默認的handler
ChannelFuture f = b.bind(80).sync();
f.channel().closeFuture().sync(); //至關於在這裏阻塞，直到serverchannel關閉
} finally {
bossGroup.shutdownGracefully();
workerGroup.shutdownGracefully();
}

這段代碼在前面的文章也有用到，基本上其意思也都在上面的註釋中說的比較清楚了，那麼咱們接下來具體的分析其中的方法調用，首先是ServerBootstrap的group方法：github

[java] view plain copy

//這裏parent用於執行server的accept時間事件，child纔是用於執行獲取的channel鏈接的事件
public ServerBootstrap group(EventLoopGroup parentGroup, EventLoopGroup childGroup) {
super.group(parentGroup);
if (childGroup == null) {
throw new NullPointerException("childGroup");
}
if (this.childGroup != null) {
throw new IllegalStateException("childGroup set already");
}
this.childGroup = childGroup;
return this;
}

這個方法是用來設置eventloopgroup，首先調用了父類的group方法（abstractbootstrap），就不將父類的方法列出來了，其實意思都差很少，eventloopgroup屬性的值。。。編程

好了，接下來咱們再來看一下channel方法：bootstrap

[java] view plain copy

//構造serversocketchannel factory
public B channel(Class<? extends C> channelClass) {
if (channelClass == null) {
throw new NullPointerException("channelClass");
}
return channelFactory(new BootstrapChannelFactory<C>(channelClass)); //構造工廠類
}
/**
* {@link ChannelFactory} which is used to create {@link Channel} instances from
* when calling {@link #bind()}. This method is usually only used if {@link #channel(Class)}
* is not working for you because of some more complex needs. If your {@link Channel} implementation
* has a no-args constructor, its highly recommend to just use {@link #channel(Class)} for
* simplify your code.
*/
@SuppressWarnings("unchecked")
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;
}

該方法主要是用於構造用於產生channel的工廠類，在咱們這段代碼說白了就是用於實例化serversocketchannel的工廠類。。。promise

接下來咱們再來看一下childHandler方法：服務器

[java] view plain copy

//設置childHandler，這個是當有channel accept以後爲其添加的handler
public ServerBootstrap childHandler(ChannelHandler childHandler) {
if (childHandler == null) {
throw new NullPointerException("childHandler");
}
this.childHandler = childHandler;
return this;
}

這個很簡單吧，就是一個賦值，具體說他有什麼用，前面的註釋有說明，不過之後的分析會說明它有什麼用的。。。架構

接下來咱們來看一下bind方法，這個比較重要吧：socket

[java] view plain copy

//最終將會建立serverchannel，而後會將其綁定到這個地址，而後對其進行初始化
public ChannelFuture bind(int inetPort) {
return bind(new InetSocketAddress(inetPort));
}

好吧，接下來再來看bind方法：ide

[java] view plain copy

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

好吧，再來看看doBind方法：

[java] view plain copy

private ChannelFuture doBind(final SocketAddress localAddress) {
final ChannelFuture regPromise = initAndRegister(); //在這裏建立serverchanel，並對其進行初始化，並將其註冊到eventloop當中去
final Channel channel = regPromise.channel();
final ChannelPromise promise = channel.newPromise();
if (regPromise.isDone()) {
doBind0(regPromise, channel, localAddress, promise); //將當前的serverchannel綁定地址
} else {
regPromise.addListener(new ChannelFutureListener() {
@Override
public void operationComplete(ChannelFuture future) throws Exception {
doBind0(future, channel, localAddress, promise);
}
});
}
return promise;
}

這裏調用了一個比較重要的方法：initAndRegister，咱們來看看它的定義：

[java] view plain copy

//建立初始化以及註冊serverchanel
final ChannelFuture initAndRegister() {
//利用工廠類建立channel
final Channel channel = channelFactory().newChannel();
try {
init(channel); //init函數留給了後面來實現，用於初始化channel，例如爲其的pipeline加上handler
} catch (Throwable t) {
channel.unsafe().closeForcibly();
return channel.newFailedFuture(t);
}
ChannelPromise regPromise = channel.newPromise();
group().register(channel, regPromise); //將當前建立的serverchannel註冊到eventloop上面去
if (regPromise.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 beause 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 regPromise;
}

代碼仍是很簡單，並且也相對比較好理解，無非就是利用前面說到過的channel工廠類來建立一個serversocketchannel，而後調用init方法對這個剛剛生成的channel進行一些初始化的操做，而後在調用eventloopgroup的register方法，將當前這個channel的註冊到group上，那麼之後這個channel的事件都在這個group上面執行，說白了也就是一些accept。、。。

好，咱們先來看看這個init方法吧：

[java] view plain copy

@Override
//初始化chanel，當用channel factory構造channel之後，會調用這個函數來初始化，說白了就是爲當前的channel的pipeline加入一些handler
void init(Channel channel) throws Exception {
//先初始化一些配置
final Map<ChannelOption<?>, Object> options = options();
synchronized (options) {
channel.config().setOptions(options);
}
//初始化一些屬性
final Map<AttributeKey<?>, Object> attrs = attrs();
synchronized (attrs) {
for (Entry<AttributeKey<?>, Object> e: attrs.entrySet()) {
@SuppressWarnings("unchecked")
AttributeKey<Object> key = (AttributeKey<Object>) e.getKey();
channel.attr(key).set(e.getValue());
}
}
//獲取當前channel的pipeline
ChannelPipeline p = channel.pipeline();
if (handler() != null) {
p.addLast(handler());
}
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 {
//這是一個inboundher，將其加入到serverchannel的pipeline上面去
ch.pipeline().addLast(new ServerBootstrapAcceptor(
currentChildGroup, currentChildHandler, currentChildOptions, currentChildAttrs));
}
});
}

代碼仍是相對很簡單，首先初始化一些配置參數，而後初始化屬性，最後還要爲當前的channel的pipeline添加一個handler，這個handler用來當channel註冊到eventloop上面以後對其進行一些初始化，咱們仍是來看看channelInitalizer的定義吧：

[java] view plain copy

public abstract class ChannelInitializer<C extends Channel> extends ChannelStateHandlerAdapter {
private static final InternalLogger logger = InternalLoggerFactory.getInstance(ChannelInitializer.class);
/**
* This method will be called once the {@link Channel} was registered. After the method returns this instance
* will be removed from the {@link ChannelPipeline} of the {@link Channel}.
*
* @param ch the {@link Channel} which was registered.
* @throws Exception is thrown if an error occours. In that case the {@link Channel} will be closed.
*/
protected abstract void initChannel(C ch) throws Exception;
@SuppressWarnings("unchecked")
@Override
public final void channelRegistered(ChannelHandlerContext ctx)
throws Exception {
boolean removed = false;
boolean success = false;
try {
//調用用戶定義的init函數對當前的channel進行初始化
initChannel((C) ctx.channel());
ctx.pipeline().remove(this);
removed = true;
ctx.fireChannelRegistered();
success = true;
} catch (Throwable t) {
logger.warn("Failed to initialize a channel. Closing: " + ctx.channel(), t);
} finally {
if (!removed) {
ctx.pipeline().remove(this);
}
if (!success) {
ctx.close();
}
}
}
@Override
public void inboundBufferUpdated(ChannelHandlerContext ctx) throws Exception {
ctx.fireInboundBufferUpdated();
}
}

它有一個channelRegistered方法，這個方法是在當前pipeline所屬的channel註冊到eventloop上面以後會激活的方法，它則是調用了用戶自定義的函數來初始化channel，而後在將當前handler移除。。。也就是執行

[java] view plain copy

ch.pipeline().addLast(new ServerBootstrapAcceptor(
currentChildGroup, currentChildHandler, currentChildOptions, currentChildAttrs));

這裏又爲當前的serversocketchannel添加了另一個handler，來看看該類型的定義吧：

[java] view plain copy

private static class ServerBootstrapAcceptor
extends ChannelStateHandlerAdapter implements ChannelInboundMessageHandler<Channel> {
private final EventLoopGroup childGroup;
private final ChannelHandler childHandler;
private final Entry<ChannelOption<?>, Object>[] childOptions;
private final Entry<AttributeKey<?>, Object>[] childAttrs;
@SuppressWarnings("unchecked")
ServerBootstrapAcceptor(
EventLoopGroup childGroup, ChannelHandler childHandler,
Entry<ChannelOption<?>, Object>[] childOptions, Entry<AttributeKey<?>, Object>[] childAttrs) {
this.childGroup = childGroup; //這個是用於管理accept的channel的eventloop
this.childHandler = childHandler;
this.childOptions = childOptions;
this.childAttrs = childAttrs;
}
@Override
public MessageBuf<Channel> newInboundBuffer(ChannelHandlerContext ctx) throws Exception {
return Unpooled.messageBuffer();
}
@Override
@SuppressWarnings("unchecked")
//當有數據進來的時候，會調用這個方法來處理數據，這裏進來的數據就是accept的channel
public void inboundBufferUpdated(ChannelHandlerContext ctx) {
MessageBuf<Channel> in = ctx.inboundMessageBuffer(); //獲取buf
for (;;) {
Channel child = in.poll();
if (child == null) {
break;
}
child.pipeline().addLast(childHandler); //爲accept的channel的pipeline加入用戶定義的初始化handler
for (Entry<ChannelOption<?>, Object> e: childOptions) {
try {
if (!child.config().setOption((ChannelOption<Object>) e.getKey(), e.getValue())) {
logger.warn("Unknown channel option: " + e);
}
} catch (Throwable t) {
logger.warn("Failed to set a channel option: " + child, t);
}
}
for (Entry<AttributeKey<?>, Object> e: childAttrs) {
child.attr((AttributeKey<Object>) e.getKey()).set(e.getValue());
}
try {
childGroup.register(child); //將當前accept的channel註冊到eventloop
} catch (Throwable t) {
child.unsafe().closeForcibly();
logger.warn("Failed to register an accepted channel: " + child, t);
}
}
}
@Override
public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
final ChannelConfig config = ctx.channel().config();
if (config.isAutoRead()) {
// stop accept new connections for 1 second to allow the channel to recover
// See https://github.com/netty/netty/issues/1328
config.setAutoRead(false);
ctx.channel().eventLoop().schedule(new Runnable() {
@Override
public void run() {
config.setAutoRead(true);
}
}, 1, TimeUnit.SECONDS);
}
// still let the exceptionCaught event flow through the pipeline to give the user
// a chance to do something with it
ctx.fireExceptionCaught(cause);
}
}

主要是有一個比較重要的方法，inboundBufferUpdated，這個方法是在有數據進來的時候會調用的，用於處理進來的數據，也就是accept到的channel，這裏就知道咱們定義的chidHandler的用處了吧，netty會將這個handler直接加入到剛剛accept到的channel的pipeline上面去。。。最後還要講當前accept到的channel註冊到child eventloop上面去，這裏也就完徹底全的明白了最開始定義的兩個eventloopgroup的做用了。。。

好了，serversocketchannel的init以及register差很少了，而後會調用doBind0方法，將當前的serversocketchannel綁定到一個本地端口，

[java] view plain copy

//將chanel綁定到一個本地地址
private static void doBind0(
final ChannelFuture regFuture, final Channel channel,
final SocketAddress localAddress, final ChannelPromise promise) {
// This method is invoked before channelRegistered() is triggered. Give user handlers a chance to set up
// the pipeline in its channelRegistered() implementation.
channel.eventLoop().execute(new Runnable() {
@Override
//匿名內部類想要訪問外面的參數，那麼外面的參數必須是要final的才行
public void run() {
if (regFuture.isSuccess()) {
//調用channel的bind方法，將當前的channl綁定到一個本地地址，實際上是調用的是pipeline的bind方法，可是最終又是調用的當前
//channel的unsafe對象的bind方法
channel.bind(localAddress, promise).addListener(ChannelFutureListener.CLOSE_ON_FAILURE);
} else {
promise.setFailure(regFuture.cause());
}
}
});
}

其實這裏調用bind方法最終仍是調用serversocketchannel的unsafe對象的bind方法。。。。

到這裏，整個serverbootstrap 就算初始化完成了，並且也能夠開始運行了。。。

[java] view plain copy

b.childHandler(new ChannelInitializer<SocketChannel>(){ //爲accept channel的pipeline預添加的inboundhandler
@Override //當新鏈接accept的時候，這個方法會調用
protected void initChannel(SocketChannel ch) throws Exception {
// TODO Auto-generated method stub
ch.pipeline().addLast(new MyChannelHandler()); //爲當前的channel的pipeline添加自定義的處理函數
}
});