前言
在源码分析的第一部分Netty源码（一）Netty架构解析里面提到了netty的几个关键组件
-
EventLoop
EventLoop是Netty中最重要的组件，一个单线程事件循环，监听IO事件、处理IO事件和任务队列。 -
EventLoopGroup
EventLoopGroup顾名思义，管理多个EventLoop。 -
Channel
Channel是能够进行IO操作组件的抽象，如读、写、连接和绑定。 -
ChannelHandler
ChannelHandler是我们使用Netty开发最关心的一个组件，它是对IO事件的具体处理逻辑。 -
ChannelPipeline
ChannelPipeline则是ChannelHandler的容器，本质是一个双向链表，将所有的ChannelHandler按照顺序连接起。
在本篇文章将会分析一个netty服务器的启动流程，以及是如何将这个组件给串联起来的。
服务端代码
下面是一段服务端的启动代码，来自netty官网
private int port;
public DiscardServer(int port) {
this.port = port;
}
public void run() throws Exception {
EventLoopGroup bossGroup = new NioEventLoopGroup(); // (1)
EventLoopGroup workerGroup = new NioEventLoopGroup();
try {
ServerBootstrap b = new ServerBootstrap(); // (2)
b.group(bossGroup, workerGroup)
.channel(NioServerSocketChannel.class) // (3)
.childHandler(new ChannelInitializer<SocketChannel>() { // (4)
@Override
public void initChannel(SocketChannel ch) {
ch.pipeline().addLast(new DiscardServerHandler());
}
})
.option(ChannelOption.SO_BACKLOG, 128) // (5)
.childOption(ChannelOption.SO_KEEPALIVE, true); // (6)
// Bind and start to accept incoming connections.
ChannelFuture f = b.bind(port).sync(); // (7)
// Wait until the server socket is closed.
// In this example, this does not happen, but you can do that to gracefully
// shut down your server.
f.channel().closeFuture().sync();
} finally {
workerGroup.shutdownGracefully();
bossGroup.shutdownGracefully();
}
}
- EventLoopGroup 之前以及详细介绍过了，处理IO操作的多线程事件循环，分为bossGroup和workGroup，分别处理连接事件和读写事件。可以通过构造参数指定EventLoopGroup 的线程数量，默认为当前cup数量的两倍；
- ServerBootstrap 为了方便构建服务器的辅助类；
- channel：由于是服务器端，使用NioServerSocketChannel构建一个Channel用于接收传入的连接；
- childHandler：添加用户自定义的ChannelHandler，比如解码、编码处理器和业务处理的Handler。
- option：设置Channel的一些参数，比如这里正在实现一个TCP/IP服务器，可以指定一些TCP参数比如tcpNoDelay和keepAlive。
- childOption：option与childOption不同在于，option指定的是传入channel的参数如现在的NioServerSocketChannel，而childOption指定的是NioServerSocketChannel接受的Channel。
- 将服务绑定到指定的端口，并启动服务。
可以看见netty构建一个服务器代码是很简单明了，大家如果看过用NIO构建一个服务器的代码是非常复杂和抽象的，而netty用极短的代码就能构建出一个服务器，可见netty框架对底层细节的抽象和对框架的设计的能力。
Channel初始化
接下来开始分析服务端的启动流程，上面代码构造的服务器最关键的一步就是调用bind()，其他的group、childHandler方法都是做一些简单的赋值操作，这里就不分析了，而bind()是将上一步赋值的参数全部串起来，然后启动服务。
public ChannelFuture bind(int inetPort) {
return bind(new InetSocketAddress(inetPort));
}
跟到关键实现的地方doBind()
private ChannelFuture doBind(final SocketAddress localAddress) {
final ChannelFuture regFuture = initAndRegister();
final Channel channel = regFuture.channel();
// ......
doBind0(regFuture, channel, localAddress, promise);
// ......
}
总结起来这个方法做了两个事件：
-
initAndRegister(): 初始化传入的Channel并且注册到EventLoopGroup -
doBind0()：绑定端口
initAndRegister()
由名字可知初始化并且注册，但是初始化什么、注册到哪里?虽然已经告诉了答案，
初始化传入的Channel并且注册到EventLoopGroup，还是要具体分析一下是怎么初始化和注册的。
initAndRegister()
final ChannelFuture initAndRegister() {
Channel channel = null;
try {
//初始化Channel
channel = channelFactory.newChannel();
init(channel);
} catch (Throwable t) {
// ......
}
//注册到EventLoopGroup
ChannelFuture regFuture = config().group().register(channel);
return regFuture;
}
初始化
channel = channelFactory.newChannel();
看到这行代码要channelFactory是什么，做什么作用?分析源码的过程中很容易就迷失，找不到关键的地方，带着问题去分析是最好的方法。比如这里当我们不知道这个channelFactory是什么的时候，首先看他的定义是什么
public interface ChannelFactory<T extends Channel> {
/**
* Creates a new channel.
*/
T newChannel();
}
定义了一个接口，初始化Channel，可以知道channelFactory的职责是初始化一个Channel。那这里channelFactory的具体实现是什么呢？
最简单的方法就是使用debug的方式找到channelFactory的实现类，或者使用编译器看这个变量这哪里被调用，做的复制操作。通过这些方法我们定位到了代码
public B channel(Class<? extends C> channelClass) {
return channelFactory(new ReflectiveChannelFactory<C>(
ObjectUtil.checkNotNull(channelClass, "channelClass")
));
}
上诉代码可以知道我们在调用.channel(NioServerSocketChannel.class)的时候就对channelFactory做了初始化，默认为ReflectiveChannelFactory，顾名思义通过反射初始化Channel，下面代码就展示了ReflectiveChannelFactory的实现。通过反射拿到Channel的构造器，然后调用初始化方法。
ReflectiveChannelFactory
//.....
this.constructor = clazz.getConstructor();
//.....
public T newChannel() {
try {
return constructor.newInstance();
} catch (Throwable t) {
throw new ChannelException("Unable to create Channel from class " + constructor.getDeclaringClass(), t);
}
}
所以分析到这里，我们就可以知道channel = channelFactory.newChannel();
这里的初始化的Channel就是第一步构造ServerBootstrap传入的NioServerSocketChannel。
接下来分析第二步init(channel)
abstract void init(Channel channel) throws Exception;
可以看见是一个抽象方法，找到服务端的具体实现
void init(Channel channel) {
//设置属性
setChannelOptions(channel, newOptionsArray(), logger);
setAttributes(channel, attrs0().entrySet().toArray(EMPTY_ATTRIBUTE_ARRAY));
ChannelPipeline p = channel.pipeline();
// ......
p.addLast(new ChannelInitializer<Channel>() {
// 将启动类传入的ChannelHandler 加入pipeline
@Override
public void initChannel(final Channel ch) {
final ChannelPipeline pipeline = ch.pipeline();
ChannelHandler handler = config.handler();
if (handler != null) {
pipeline.addLast(handler);
}
// acceptor角色
ch.eventLoop().execute(new Runnable() {
@Override
public void run() {
pipeline.addLast(new ServerBootstrapAcceptor(
ch, currentChildGroup, currentChildHandler, currentChildOptions, currentChildAttrs));
}
});
}
});
}
这段代码比较长，个人认为有两个点需要重点分析
- ChannelPipeline p = channel.pipeline();
- ServerBootstrapAcceptor
ChannelPipeline netty中的一个关键的组件在这里出现了，ChannelPipeline是ChannelHandler 的容器，这里是调用channel的pipeline()返回的ChannelPipeline对象。所以需要找一下ChannelPipeline 是什么、在什么时候初始化的，按照之前的方法很轻松的在AbstractChannel找到代码
protected AbstractChannel(Channel parent) {
this.parent = parent;
id = newId();
unsafe = newUnsafe();
pipeline = newChannelPipeline();
}
protected DefaultChannelPipeline newChannelPipeline() {
return new DefaultChannelPipeline(this);
}
可见是在Channel初始化的时候创建的ChannelPipeline默认为DefaultChannelPipeline，所以在这里也可以知道每个Channel都对应着一个ChannelPipeline。
接下来分析很关键的代码
ch.eventLoop().execute(new Runnable() {
@Override
public void run() {
pipeline.addLast(new ServerBootstrapAcceptor(
ch, currentChildGroup, currentChildHandler, currentChildOptions, currentChildAttrs));
}
});
在pipeline中添加了名为ServerBootstrapAcceptor的处理器，在第一篇文章中我们提到过Reactor模式，里面就有一个acceptor角色用于将mainReactor角色介绍的连接转给subReactor，这里看一下ServerBootstrapAcceptor是不是也在做这个事情。
private static class ServerBootstrapAcceptor extends ChannelInboundHandlerAdapter
继承自ChannelInboundHandlerAdapter，可见是一个ChannelHandler而且处理的是入站事件，下面分析它是怎么处理的入站事件。
public void channelRead(ChannelHandlerContext ctx, Object msg) {
// 父Channel监听的是连接事件，所有直接转为Channel
final Channel child = (Channel) msg;
//添加 启动类传入的childHandler
child.pipeline().addLast(childHandler);
//设置启动类传入的属性
setChannelOptions(child, childOptions, logger);
setAttributes(child, childAttrs);
try {
// 将boosWorup获得的连接注册到childGroup上
childGroup.register(child).addListener(new ChannelFutureListener() {
// ......
});
} catch (Throwable t) {
// ......
}
}
- 将父Channel(NioServerSocketChannel)读取的信息强转为Channel
- 添加 启动类传入的childHandler
- 设置启动类传入的属性
- 将boosWorup获得的连接注册到childGroup上
分析到这里可以知道netty是通过ChannelHandler的形式将bossGroup与workGroup联系起来，当bossGroup接受到新的连接就将其注册到childGroup上。
最后再回到initAndRegister()这个方法的最后一步
ChannelFuture regFuture = config().group().register(channel);
这个方法就是将我们实体化的NioServerSocketChannel注册到bossGroup上。
分析到这里对initAndRegister()做一个简单的总结
- 通过channelFactory使用反射的形式将启动类传入的Channel(NioServerSocketChannel)实例化；
- 实例化Channel的时候会创建一个ChannelPipeline对象，以Channel一一对应；
- 通过ServerBootstrapAcceptor将bossGroup接入的连接转给workGroup；
- 将实例化的Channel(NioServerSocketChannel)注册到boosGroup上。
所以initAndRegister()实例化的是启动类传入的Channel对象并注册在boosGroup上。
绑定地址
这里分析服务启动第二步doBind0绑定端口地址
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
public void run() {
if (regFuture.isSuccess()) {
channel.bind(localAddress, promise).addListener(ChannelFutureListener.CLOSE_ON_FAILURE);
} else {
promise.setFailure(regFuture.cause());
}
}
});
}
跟进去channel.bind(SocketAddress localAddress, ChannelPromise promise)找到AbstractChannel的bind方法
public ChannelFuture bind(SocketAddress localAddress, ChannelPromise promise) {
return pipeline.bind(localAddress, promise);
}
调用pipeline的bind方法，再跟进去
@Override
public final ChannelFuture bind(SocketAddress localAddress, ChannelPromise promise) {
return tail.bind(localAddress, promise);
}
之前说过ChannelPipeline是管理ChannelHandler的容器，DefaultChannelPipeline的实现就是用一个双向链表，初始化的是一个有一个头节点HeadContext和尾节点TailContext，这里不具体分析。
这里分析需要打上断点一步一步调用，最好找到实现在AbstractUnsafe中的bind方法
@Override
public final void bind(final SocketAddress localAddress, final ChannelPromise promise) {
//......
boolean wasActive = isActive();
try {
//绑定端口地址
doBind(localAddress);
} catch (Throwable t) {
//......
}
if (!wasActive && isActive()) {
invokeLater(new Runnable() {
@Override
public void run() {
//触发channelActive事件
pipeline.fireChannelActive();
}
});
}
//绑定成功
safeSetSuccess(promise);
}
简化之后做三件事情调用jdk底层绑定端口，触发channelActive事件，将promise设置为成功。
doBind 调用jdk底层绑定端口
protected void doBind(SocketAddress localAddress) throws Exception {
if (PlatformDependent.javaVersion() >= 7) {
javaChannel().bind(localAddress, config.getBacklog());
} else {
javaChannel().socket().bind(localAddress, config.getBacklog());
}
}
就此服务端启动的流程结束，当然这里分析不是所有细节都涉及到了，只是对netty服务端启动流程的一个简要分析。
总结
- 通过ServerBootstrap将主要的组件先添加进去，比如NioEventLoopGroup、Channel、ChannelHandler、option、childOption等；
- 通过反射创建出Channel，并且在创建Channel的时候也会初始化ChannelPipeline；
- 将ChannelHandler添加到Channel所对应的ChannelPipeline中；
- 通过ServerBootstrapAcceptor将Channel所接受的连接转交给workGroup处理；
- 将第2步初始化Channel注册到bossGroup上；
- 调用jdk底层绑定端口地址；
- 触发fireChannelActive事件；
netty服务端的启动流程主要就是上面的流程，越看netty的源码就越感觉netty设计的精妙。不管是底层细节、还是整体框架的设计，比如创建一个netty的客户端如下
public static void main(String[] args) throws Exception {
String host = args[0];
int port = Integer.parseInt(args[1]);
EventLoopGroup workerGroup = new NioEventLoopGroup();
try {
Bootstrap b = new Bootstrap(); // (1)
b.group(workerGroup); // (2)
b.channel(NioSocketChannel.class); // (3)
b.option(ChannelOption.SO_KEEPALIVE, true); // (4)
b.handler(new ChannelInitializer<SocketChannel>() {
@Override
public void initChannel(SocketChannel ch) throws Exception {
ch.pipeline().addLast(new TimeClientHandler());
}
});
// Start the client.
ChannelFuture f = b.connect(host, port).sync(); // (5)
// Wait until the connection is closed.
f.channel().closeFuture().sync();
} finally {
workerGroup.shutdownGracefully();
}
}
对比服务器只有少数不同的地方。或者不使用NioServerSocketChannel作为服务器的底层连接将NioServerSocketChannel替换就可以了比如EpollServerSocketChannel就可以，不需要调整其他地方。本篇的源码分析就到此为止。
网友评论