Netty的启动执行过程分析(一)

开如启动我们有Netty的服务端，用的方法对象是ServerBootstrap对象

具体的Netty的服务端的代码如下

public class MyNettyServer {

public static void main(String[] args) {
// 因为我们的Netty用的是主从模式的开发模式，所以要设置二个主从组给这个服务器引擎,
// 同时这个二个group可以在创建的时候可以指定这个组当中可以有多少个线程，默认是1个
NioEventLoopGroup parent = new NioEventLoopGroup();
NioEventLoopGroup chlid = new NioEventLoopGroup();
try {
// 先进行构建我们的服务器引擎ServerBootStrap,直接new也就可以了
ServerBootstrap serverBootStrap = new ServerBootstrap();
serverBootStrap.group(parent, chlid);
// 进行配置这个服务器引擎ServerBootStrap，刚创建时是不能直接启动这个这个服务器引擎的，要配置三个基本的配置的
// 这个三个基本的配置如下:channel,handler,option

// 这个表示的是如果我们的服务器如果如果不过来这么多的通道，最多只能让1024个通道时进行阻塞等待
serverBootStrap.option(ChannelOption.SO_BACKLOG, 1024);
serverBootStrap.option(ChannelOption.SO_KEEPALIVE, true);

// 给我们的引擎设置一下我们的通道
serverBootStrap.channel(NioServerSocketChannel.class);
// 最后再注册一个handler处理器对象，进行注册一个通道初始化器对象
serverBootStrap.childHandler(new ChannelInitializer<SocketChannel>() {

@Override
protected void initChannel(SocketChannel ch) throws Exception {
// TODO Auto-generated method stub
System.out.println("进行通道的初始化操作，这个方法是在我们的通道向Selector注册时，就会执行这个方法");

ch.pipeline().addLast(new DelimiterBasedFrameDecoder(Integer.MAX_VALUE,
Delimiters.lineDelimiter()[0]));

ch.pipeline().addLast(new StringDecoder(Charset.defaultCharset()));
//进行数据的编码，然后发送给客户端
ch.pipeline().addLast(new StringEncoder(Charset.defaultCharset()));
//进行具体的一个管道调用处理操作
ch.pipeline().addFirst(new MyHandlerTest());
}
});
//现在进行绑定端口,同时用的是同步的方法
ChannelFuture future=serverBootStrap.bind(8888).sync();
//进行给我们的客户端发送响应数据
//          future.channel().writeAndFlush("开明乾坤，照大地...\r\n");

ByteBuf bu=PooledByteBufAllocator.DEFAULT.buffer(1024);
bu.writeBytes("开明乾坤，照大地".getBytes(Charset.defaultCharset()));
future.channel().writeAndFlush(bu);

//进行关闭操作，这个方法是一个阻塞的方法，线程会被阻塞到这里
future.channel().closeFuture().sync();
//进行那么操作

// 最后进行关闭这个组
} catch (Exception e) {
// TODO Auto-generated catch block
e.printStackTrace();
}finally{
parent.shutdownGracefully();
chlid.shutdownGracefully();
}

}

}

具体的服务器的handler的代码如下：

public class MyHandlerTest extends ChannelInboundHandlerAdapter {
//这个方法是在我们的channel向我们的Select注册时调用这个方法
@Override
public void channelRegistered(ChannelHandlerContext ctx) throws Exception {
super.channelRegistered(ctx);
}
//这个方法是在我们的通道进行有数据发送过来时进行调用这个方法去取得通道中数据
@Override
public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
super.channelRead(ctx, msg);
System.out.println("有数据过来了，你要接收这个数据了");
//进行判断一下我们的数据的类型
if(msg instanceof ByteBuf){
//进行具体数据处理过程如下：
ByteBuf buf=(ByteBuf) msg;
//进行数据的转换操作,也可以Charset.defaultCharset()
System.out.println("我们的服务器接收到客户端发过来的数据为:"+
buf.toString(Charset.forName("UTF-8")));
}
}

}

具体的客户端的代码如下：

public class MyNettyClient {
public static void main(String[] args) {
//进行创建事件组对象
NioEventLoopGroup group=new NioEventLoopGroup();

try {
//创建一个客户端启动引擎，用的是Bootstrap
Bootstrap bootStrap=new Bootstrap();
//进行增加组对象
bootStrap.group(group);
//进行客户端启动引擎的基本的配置如下
bootStrap.option(ChannelOption.SO_KEEPALIVE,true)//保持存活
.option(ChannelOption.SO_RCVBUF,1024) //设置接受缓冲区的大小
.channel(NioSocketChannel.class)       //设置突然我们客户端的通道为SocketChannel
.handler(new ChannelInitializer<NioSocketChannel>() {
@Override
protected void initChannel(NioSocketChannel ch) throws Exception {
//进行基本的客户端的初始化操作
System.out.println("我们的客户端开始工作了...");
//增加一些基本的处理器操作

//增加数据分割符，数据后面增加一个分割符，也就是
//Delimiters.lineDelimiter()[0],这个lineDelimiter返回有二个元素
//的数据，第一个表示的是\r\n，第二个是\n,这里用的是第一个
ch.pipeline().addLast(new DelimiterBasedFrameDecoder(Integer.MAX_VALUE,
Delimiters.lineDelimiter()[0]));
//进行增加一些数据的解码和编码的操作,这个Charset.defaultCharset()就是UTF-8
ch.pipeline().addLast(new StringEncoder());
//这个进行接收字符串的数据的操作
ch.pipeline().addLast(new StringDecoder());

ch.pipeline().addLast(new MyReadDataHandler());
}
});

//进行启动我们的客户端引擎对象,使用的是同步的启动方法
ChannelFuture future=bootStrap.connect("localhost",8888).sync();
//进行给服务器发送数据,Delimiters.lineDelimiter()这个表示数据有\r\n时就把这个数据发送给服务器
//          future.channel().write("杨海龙".getBytes(Charset.forName("UTF-8")));//这种方式不可以
//          future.channel().writeAndFlush(Delimiters.lineDelimiter()[0]);//不可以
future.channel().writeAndFlush("杨海龙");//注意这个不要增加回换行在数据的后面，因为在定义编码的时候默认增加上去了
//          future.channel().writeAndFlush(Delimiters.lineDelimiter()[0]);//不可以

//第二种数据写出的方式
ByteBuf buf=PooledByteBufAllocator.DEFAULT.buffer(1024);
//进行写入数据
buf.writeBytes("李四".getBytes(Charset.defaultCharset()));
//最后进行写出数据
future.channel().writeAndFlush(buf);

//进行通道等待关闭的操作
future.channel().closeFuture().sync();
} catch (Exception e) {
e.printStackTrace();
}finally{
group.shutdownGracefully();
}

}
}

我们有客户端的Handler的代码如下：

public class MyReadDataHandler extends ChannelInboundHandlerAdapter {
//这个方法是用来注册我们有Channnel时调用这的方法
@Override
public void channelRegistered(ChannelHandlerContext ctx) throws Exception {
// TODO Auto-generated method stub
super.channelRegistered(ctx);
}
//我们的服务器有发送数据过来时，就会调用这个方法进行读取这个这个数据
@Override
public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
super.channelRead(ctx, msg);
//进行数据的读取
System.out.println("读取服务器发送过来的数据...");

//进行接收数据了
if(msg instanceof ByteBuf){
ByteBuf buf=(ByteBuf) msg;
System.out.println("我们的从服务器接收到的数据为:"+buf.toString(Charset.defaultCharset()));
}
}

}

现在开始一步一步的调度我们有Netty的服务器端的代码

// 先进行构建我们的服务器引擎ServerBootStrap,直接new也就可以了
ServerBootstrap serverBootStrap = new ServerBootstrap();

上面这个调用创建一个服务器的启动辅助类，加载了这个ServerBootStrap当中的所有的静态的成员的方法，其中有重要的ServerBootstrap的配置类ServerBootstrapConfig

public class ServerBootstrap extends AbstractBootstrap<ServerBootstrap, ServerChannel> {

private static final InternalLogger logger = InternalLoggerFactory.getInstance(ServerBootstrap.class);

private final Map<ChannelOption<?>, Object> childOptions = new LinkedHashMap<ChannelOption<?>, Object>();
private final Map<AttributeKey<?>, Object> childAttrs = new LinkedHashMap<AttributeKey<?>, Object>();
private final ServerBootstrapConfig config = new ServerBootstrapConfig(this);//也会调用这个类的构造方法
public ServerBootstrap() { }
}

同时我们的这个ServerBootstrap继承于AbstractBootstrap，所以也会调用我们有AbstractBootstrap的构造方法

public abstract class AbstractBootstrap<B extends AbstractBootstrap<B, C>, C extends Channel> implements Cloneable {

volatile EventLoopGroup group;
@SuppressWarnings("deprecation")
private volatile ChannelFactory<? extends C> channelFactory;
private volatile SocketAddress localAddress;
private final Map<ChannelOption<?>, Object> options = new LinkedHashMap<ChannelOption<?>, Object>();
private final Map<AttributeKey<?>, Object> attrs = new LinkedHashMap<AttributeKey<?>, Object>();
private volatile ChannelHandler handler;

AbstractBootstrap() {
// Disallow extending from a different package.
}

执行到这个方法时会绑定我们有线程的模型

serverBootStrap.group(parent, chlid);

这个方法先把这个parent加入到当前类的父类AbstractBootstrap当中的group里面，然后再把child加入到serverBootstrap当中的group当中

/**
* The {@link EventLoopGroup} which is used to handle all the events for the to-be-created
* {@link Channel}
*/
@SuppressWarnings("unchecked")
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;
}

ServerBootstrap当中的group方法

/**
* Set the {@link EventLoopGroup} for the parent (acceptor) and the child (client). These
* {@link EventLoopGroup}'s are used to handle all the events and IO for {@link ServerChannel} and
* {@link Channel}'s.
*/
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;
}

然后再进行我们有TCP参数的绑定操作：

// 这个表示的是如果我们的服务器如果如果不过来这么多的通道，最多只能让1024个通道时进行阻塞等待
serverBootStrap.option(ChannelOption.SO_BACKLOG, 1024);
serverBootStrap.option(ChannelOption.SO_KEEPALIVE, true);

// 给我们的引擎设置一下我们的通道
serverBootStrap.channel(NioServerSocketChannel.class);

下面的方法调用了父类AbstractBootstrap的option的方法

/**
* Allow to specify a {@link ChannelOption} which is used for the {@link Channel} instances once they got
* created. Use a value of {@code null} to remove a previous set {@link ChannelOption}.
*/
@SuppressWarnings("unchecked")
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;
}

上面的这个options是我们父类AbstractBootstrap的静态的linkedHashMap的结构体

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

对应的结果为：options={SO_BACKLOG=1024, SO_KEEPALIVE=true}

接下来开始设置我们有服务器的通道的方法了

// 给我们的引擎设置一下我们的通道
serverBootStrap.channel(NioServerSocketChannel.class);

具体的调用代码如下：

一样还是调用我们有ServerBootstrap的channel的方法，下面这个channelFactory(通道工厂类)是专门用我们调用这个方法时传递进来的字节类名来进行反射创建出来具体的通道类，也就是创建出来了NioServerSocketChannel类对象

/**
* The {@link Class} which is used to create {@link Channel} instances from.
* You either use this or {@link #channelFactory(io.netty.channel.ChannelFactory)} if your
* {@link Channel} implementation has no no-args constructor.
*/
public B channel(Class<? extends C> channelClass) {
if (channelClass == null) {
throw new NullPointerException("channelClass");
}
return channelFactory(new ReflectiveChannelFactory<C>(channelClass));
}

接下来进行具体的handler的注册工作

// 最后再注册一个handler处理器对象，进行注册一个通道初始化器对象
serverBootStrap.childHandler(new ChannelInitializer<SocketChannel>() {

@Override
protected void initChannel(SocketChannel ch) throws Exception {
// TODO Auto-generated method stub
System.out.println("进行通道的初始化操作，这个方法是在我们的通道向Selector注册时，就会执行这个方法");

ch.pipeline().addLast(new DelimiterBasedFrameDecoder(Integer.MAX_VALUE,
Delimiters.lineDelimiter()[0]));

ch.pipeline().addLast(new StringDecoder(Charset.defaultCharset()));
//进行数据的编码，然后发送给客户端
ch.pipeline().addLast(new StringEncoder(Charset.defaultCharset()));
//进行具体的一个管道调用处理操作
ch.pipeline().addFirst(new MyHandlerTest());
}
});

* Set the {@link ChannelHandler} which is used to serve the request for the {@link Channel}'s.
*/
public ServerBootstrap childHandler(ChannelHandler childHandler) {
if (childHandler == null) {
throw new NullPointerException("childHandler");
}
this.childHandler = childHandler;
return this;
}

接下是开始绑定我们有数据的了

//现在进行绑定端口,同时用的是同步的方法
ChannelFuture future=serverBootStrap.bind(8888).sync();

还是调用了我们有AbstractBootstrap的方法

/**
* Create a new {@link Channel} and bind it.
*/
public ChannelFuture bind(InetAddress inetHost, int inetPort) {
return bind(new InetSocketAddress(inetHost, inetPort));
}
/**
* Create a new {@link Channel} and bind it.
*/
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();//这个方法具体的去调用我们有工厂的方法，然后调用在newChannel的方法当中调用了一个空构造方法创建出来我们ServerSocketChannel的通道类
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(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;
}
}

现在这个validate这个方法是我们ServerBootstrap类的方法的具体的实现如下

/**
* Validate all the parameters. Sub-classes may override this, but should
* call the super method in that case.
*/
@SuppressWarnings("unchecked")
public B validate() {
if (group == null) {
throw new IllegalStateException("group not set");
}
if (channelFactory == null) {
throw new IllegalStateException("channel or channelFactory not set");
}
return (B) this;
}

在我们有上面的这个DoBind方法当中调用了一个initAndRegister();

final ChannelFuture initAndRegister() {
Channel channel = null;
try {
channel = channelFactory.newChannel();
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();
}
}

// 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 because 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 regFuture;
}

//这个方法具体的去调用我们有工厂的方法，然后调用在newChannel的方法当中调用了一个空构造方法创建出来我们ServerSocketChannel的通道类

@Override
public T newChannel() {
try {
return clazz.getConstructor().newInstance();
} catch (Throwable t) {
throw new ChannelException("Unable to create Channel from class " + clazz, t);
}
}

调用这个初始化init的方法代码的操作的方法如下

@Override
void init(Channel channel) throws Exception {
final Map<ChannelOption<?>, Object> options = options0();//得到我们有在服务器当中设置的可选的参数
synchronized (options) {
setChannelOptions(channel, options, logger);
}
//我们没有设置服务器的att，所以这个可以不用看
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());
}
}
//得到一个默认的DefaultChannelPipeline
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()));
}
//下面这些操作都是我们有handler的基本的设置
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));
}
});
}
});
}

进行具体的设置我们有服务器的可选的参数

static void setChannelOptions(
Channel channel, Map<ChannelOption<?>, Object> options, InternalLogger logger) {//进行for取出我们有已经设置的option
for (Map.Entry<ChannelOption<?>, Object> e: options.entrySet()) {
setChannelOption(channel, e.getKey(), e.getValue(), logger);
}
}

@SuppressWarnings("unchecked")
private static void setChannelOption(
Channel channel, ChannelOption<?> option, Object value, InternalLogger logger) {
try {//修改我们的配置文件，把这些数据设置进行我们有配置类当中
if (!channel.config().setOption((ChannelOption<Object>) option, value)) {
logger.warn("Unknown channel option '{}' for channel '{}'", option, channel);
}
} catch (Throwable t) {
logger.warn(
"Failed to set channel option '{}' with value '{}' for channel '{}'", option, value, channel, t);
}
}

其他的看后续文章