Netty源码阅读（一） ServerBootstrap启动
2016-10-23 23:16
666 查看
Netty源码阅读（一） ServerBootstrap启动
转自我的GithubNetty是由JBOSS提供的一个java开源框架。Netty提供异步的、事件驱动的网络应用程序框架和工具，用以快速开发高性能、高可靠性的网络服务器和客户端程序。本文讲会对Netty服务启动的过程进行分析，主要关注启动的调用过程，从这里面进一步理解Netty的线程模型，以及Reactor模式。
这是我画的一个Netty启动过程中使用到的主要的类的概要类图，当然是用到的类比这个多得多，而且我也忽略了各个类的继承关系，关于各个类的细节，可能以后会写单独的博客进行分析。在这里主要注意那么几个地方：
1. ChannelPromise关联了Channel和Executor，当然channel中也会有EventLoop的实例。 2. 每个channel有自己的pipeline实例。 3. 每个NioEventLoop中有自己的Executor实例和Selector实例。
网络请求在NioEventLoop中进行处理，当然accept事件也是如此，它会把接收到的channel注册到一个EventLoop的selector中，以后这个channel的所有请求都由所注册的EventLoop进行处理，这也是Netty用来处理竞态关系的机制，即一个channel的所有请求都在一个线程中进行处理，也就不会存在跨线程的冲突，因为这些调用都线程隔离了。
下面我们先看一段Netty源码里面带的example代码，直观感受一下Netty的使用：
1 // Configure the server.
2 EventLoopGroup bossGroup = new NioEventLoopGroup(1);
3 EventLoopGroup workerGroup = new NioEventLoopGroup();
4 try {
5 ServerBootstrap b = new ServerBootstrap();
6 b.group(bossGroup, workerGroup)
7 .channel(NioServerSocketChannel.class)
8 .option(ChannelOption.SO_BACKLOG, 100) // 设置tcp协议的请求等待队列
9 .handler(new LoggingHandler(LogLevel.INFO))
10 .childHandler(new ChannelInitializer<SocketChannel>() {
11 @Override
12 public void initChannel(SocketChannel ch) throws Exception {
13 ChannelPipeline p = ch.pipeline();
14 if (sslCtx != null) {
15 p.addLast(sslCtx.newHandler(ch.alloc()));
16 }
17 p.addLast(new EchoServerHandler());
18 }
19 });
20
21 // Start the server.
22 ChannelFuture f = b.bind(PORT).sync();
23
24 // Wait until the server socket is closed.
25 f.channel().closeFuture().sync();
26 } finally {
27 // Shut down all event loops to terminate all threads.
28 bossGroup.shutdownGracefully();
29 workerGroup.shutdownGracefully();
30 }首先我们先来了解Netty的主要类：
EventLoop 这个相当于一个处理线程，是Netty接收请求和处理IO请求的线程。
EventLoopGroup 可以理解为将多个EventLoop进行分组管理的一个类，是EventLoop的一个组。
ServerBootstrap 从命名上看就可以知道，这是一个对服务端做配置和启动的类。
ChannelPipeline 这是Netty处理请求的责任链，这是一个ChannelHandler的链表，而ChannelHandler就是用来处理网络请求的内容的。
ChannelHandler 用来处理网络请求内容，有ChannelInboundHandler和ChannelOutboundHandler两种，ChannlPipeline会从头到尾顺序调用ChannelInboundHandler处理网络请求内容，从尾到头调用ChannelOutboundHandler处理网络请求内容。这也是Netty用来灵活处理网络请求的机制之一，因为使用的时候可以用多个decoder和encoder进行组合，从而适应不同的网络协议。而且这种类似分层的方式可以让每一个Handler专注于处理自己的任务而不用管上下游，这也是pipeline机制的特点。这跟TCP/IP协议中的五层和七层的分层机制有异曲同工之妙。
现在看上面的代码，首先创建了两个EventLoopGroup对象，作为group设置到ServerBootstrap中，然后设置Handler和ChildHandler，最后调用bind()方法启动服务。下面按照Bootstrap启动顺序来看代码。
1 public ServerBootstrap group(EventLoopGroup parentGroup, EventLoopGroup childGroup) {
2 super.group(parentGroup);
3 if (childGroup == null) {
4 throw new NullPointerException("childGroup");
5 }
6 if (this.childGroup != null) {
7 throw new IllegalStateException("childGroup set already");
8 }
9 this.childGroup = childGroup;
10 return this;
11 }首先是设置EverLoopGroup，parentGroup一般用来接收accpt请求，childGroup用来处理各个连接的请求。不过根据开发的不同需求也可以用同一个group同时作为parentGroup和childGroup同时处理accpt请求和其他io请求。
1 public B channel(Class<? extends C> channelClass) {
2 if (channelClass == null) {
3 throw new NullPointerException("channelClass");
4 }
5 return channelFactory(new ReflectiveChannelFactory<C>(channelClass));
6 }接下来的channel()方法设置了ServerBootstrap的ChannelFactory，这里传入的参数是NioServerSocketChannel.class，也就是说这个ReflectiveChannelFactory创建的就是NioServerSocketChannel的实例。
后面的option()，handler()和childHandler()分别是设置Socket连接的参数，设置parentGroup的Handler，设置childGroup的Handler。childHandler()传入的ChannelInitializer实现了一个initChannel方法，用于初始化Channel的pipeline，以处理请求内容。
之前都是在对ServerBootstrap做设置，接下来的ServerBootstrap.bind()才是启动的重头戏。我们继续按照调用顺序往下看。
1 public ChannelFuture bind(int inetPort) {
2 return bind(new InetSocketAddress(inetPort));
3 }
4
5 /**
6 * Create a new {@link Channel} and bind it.
7 */
8 public ChannelFuture bind(SocketAddress localAddress) {
9 validate();
10 if (localAddress == null) {
11 throw new NullPointerException("localAddress");
12 }
13 return doBind(localAddress);
14 }
15
16 // AbstractBootstrap
17 private ChannelFuture doBind(final SocketAddress localAddress) {
18 final ChannelFuture regFuture = initAndRegister();
19 final Channel channel = regFuture.channel();
20 if (regFuture.cause() != null) {
21 return regFuture;
22 }
23
24 if (regFuture.isDone()) {
25 // At this point we know that the registration was complete and successful.
26 ChannelPromise promise = channel.newPromise();
27 doBind0(regFuture, channel, localAddress, promise);
28 return promise;
29 } else {
30 // Registration future is almost always fulfilled already, but just in case it's not.
31 final PendingRegistrationPromise promise = new PendingRegistrationPromise(channel);
32 regFuture.addListener(new ChannelFutureListener() {
33 @Override
34 public void operationComplete(ChannelFuture future) throws Exception {
35 Throwable cause = future.cause();
36 if (cause != null) {
37 // Registration on the EventLoop failed so fail the ChannelPromise directly to not cause an
38 // IllegalStateException once we try to access the EventLoop of the Channel.
39 promise.setFailure(cause);
40 } else {
41 // Registration was successful, so set the correct executor to use.
42 // See https://github.com/netty/netty/issues/2586 43 promise.registered();
44
45 doBind0(regFuture, channel, localAddress, promise);
46 }
47 }
48 });
49 return promise;
50 }
51 }我们可以看到bind()的调用最终调用到了doBind(final SocketAddress)，在这里我们看到先调用了initAndRegister()方法进行初始化和register操作。了解JavaNIO框架的同学应该能看出来是在这个方法中将channel注册到selector中的。最后程序再调用了doBind0()方法进行绑定，先按照顺序看initAndRegister方法做了什么操作。
1 // AbstractBootstrap
2 final ChannelFuture initAndRegister() {
3 Channel channel = null;
4 try {
5 channel = channelFactory.newChannel();
6 init(channel);
7 } catch (Throwable t) {
8 // ...
9 }
10
11 ChannelFuture regFuture = config().group().register(channel);
12 // ...
13 return regFuture;
14 }为了简单其间，我忽略了处理异常分支的代码，同学们有兴趣可以自行下载Netty源码对照。在这里终于看到channel的创建了，调用的是ServerBootstrap的channelFactory，之前的代码我们也看到了这里的工厂是一个ReflectChannelFactory，在构造函数中传入的是NioServerSocketChannel.class，所以这里创建的是一个NioServerSocketChannel的对象。接下来init(channel)对channel进行初始化。
1 // ServerBootstrap
2 void init(Channel channel) throws Exception {
3 final Map<ChannelOption<?>, Object> options = options0();
4 synchronized (options) {
5 channel.config().setOptions(options);
6 }
7
8 // 设置channel.attr
9 final Map<AttributeKey<?>, Object> attrs = attrs0();
10 synchronized (attrs) {
11 for (Entry<AttributeKey<?>, Object> e: attrs.entrySet()) {
12 @SuppressWarnings("unchecked")
13 AttributeKey<Object> key = (AttributeKey<Object>) e.getKey();
14 channel.attr(key).set(e.getValue());
15 }
16 }
17
18 ChannelPipeline p = channel.pipeline();
19
20 final EventLoopGroup currentChildGroup = childGroup;
21 // childGroup的handler
22 final ChannelHandler currentChildHandler = childHandler;
23 final Entry<ChannelOption<?>, Object>[] currentChildOptions;
24 final Entry<AttributeKey<?>, Object>[] currentChildAttrs;
25 synchronized (childOptions) {
26 currentChildOptions = childOptions.entrySet().toArray(newOptionArray(childOptions.size()));
27 }
28 synchronized (childAttrs) {
29 currentChildAttrs = childAttrs.entrySet().toArray(newAttrArray(childAttrs.size()));
30 }
31 // 给channelpipeline添加handler
32 p.addLast(new ChannelInitializer<Channel>() {
33 @Override
34 public void initChannel(Channel ch) throws Exception {
35 final ChannelPipeline pipeline = ch.pipeline();
36 // group的handler
37 ChannelHandler handler = config.handler();
38 if (handler != null) {
39 pipeline.addLast(handler);
40 }
41
42 // We add this handler via the EventLoop as the user may have used a ChannelInitializer as handler.
43 // In this case the initChannel(...) method will only be called after this method returns. Because
44 // of this we need to ensure we add our handler in a delayed fashion so all the users handler are
45 // placed in front of the ServerBootstrapAcceptor.
46 ch.eventLoop().execute(new Runnable() {
47 @Override
48 public void run() {
49 pipeline.addLast(new ServerBootstrapAcceptor(
50 currentChildGroup, currentChildHandler, currentChildOptions, currentChildAttrs));
51 }
52 });
53 }
54 });
55 }先是设置了channel的option和attr，然后将handler加入到channelpipleline的handler链中，这里大家请特别注意ServerBootstrapAcceptor这个Handler，因为接下来对于客户端请求的处理以及工作channl的注册可全是这个Handler处理的。不过由于现在channel还没有注册，所以还不会调用initChannel()方法，而是将这个handler对应的context加入到一个任务队列中，等到channel注册成功了再执行。关于ChannelPipeline的内容我们以后再说。然后在initAndRegister()方法中调用config().group().register(channel)对channel进行注册。config().group()获取到的其实就是bossGroup，在这个例子中就是一个NioEventLoopGroup，由于它继承了MultithreadEventLoopGroup所以这里调用的其实是这个类的方法。
1 // MultithreadEventLoopGroup
2 public ChannelFuture register(Channel channel) {
3 return next().register(channel);
4 }
5
6 public EventLoop next() {
7 return (EventLoop) super.next();
8 }
9
10 // SingleThreadEventLoop
11 public ChannelFuture register(Channel channel) {
12 return register(new DefaultChannelPromise(channel, this));
13 }
14
15 @Override
16 public ChannelFuture register(final ChannelPromise promise) {
17 ObjectUtil.checkNotNull(promise, "promise");
18 promise.channel().unsafe().register(this, promise);
19 return promise;
20 }这里会获取EventLoopGroup中的一个EventLoop，其实我们用的是NioEventLoopGroup所以这里获取到的其实是NioEventLoop，而NioEventLoop继承了SingleThreadEventLoop，这里register方法调用的就是SingleThreadEventLoop中的方法。我们重遇来到了channel最终注册的地方，这里其实是调用了channel的unsafe对象中的register方法，也就是NioServerSocketChannel的方法，这个方法是在AbstractChannel祖先类中实现的，代码如下：
1 public final void register(EventLoop eventLoop, final ChannelPromise promise) {
2 if (eventLoop == null) {
3 throw new NullPointerException("eventLoop");
4 }
5 if (isRegistered()) {
6 promise.setFailure(new IllegalStateException("registered to an event loop already"));
7 return;
8 }
9 if (!isCompatible(eventLoop)) {
10 promise.setFailure(
11 new IllegalStateException("incompatible event loop type: " + eventLoop.getClass().getName()));
12 return;
13 }
14 // 设置eventLoop
15 AbstractChannel.this.eventLoop = eventLoop;
16 // 这里是跟Netty的线程模型有关的，注册的方法只能在channel的工作线程中执行
17 if (eventLoop.inEventLoop()) {
18 register0(promise);
19 } else {
20 try {
21 eventLoop.execute(new Runnable() {
22 @Override
23 public void run() {
24 register0(promise);
25 }
26 });
27 } catch (Throwable t) {
28 logger.warn(
29 "Force-closing a channel whose registration task was not accepted by an event loop: {}",
30 AbstractChannel.this, t);
31 closeForcibly();
32 closeFuture.setClosed();
33 safeSetFailure(promise, t);
34 }
35 }
36 }
37
38 // AbstractNioChannel
39 protected void doRegister() throws Exception {
40 boolean selected = false;
41 for (;;) {
42 try {
43 selectionKey = javaChannel().register(eventLoop().selector, 0, this);
44 return;
45 } catch (CancelledKeyException e) {
46 // ...
47 }
48 }
49 }
50
51 // AbstractSelectableChannel
52 public final SelectionKey register(Selector sel, int ops,Object att)
53 throws ClosedChannelException
54 {
55 synchronized (regLock) {
56 if (!isOpen())
57 throw new ClosedChannelException();
58 if ((ops & ~validOps()) != 0)
59 throw new IllegalArgumentException();
60 if (blocking)
61 throw new IllegalBlockingModeException();
62 SelectionKey k = findKey(sel);
63 if (k != null) {
64 k.interestOps(ops);
65 k.attach(att);
66 }
67 if (k == null) {
68 // New registration
69 synchronized (keyLock) {
70 if (!isOpen())
71 throw new ClosedChannelException();
72 k = ((AbstractSelector)sel).register(this, ops, att);
73 addKey(k);
74 }
75 }
76 return k;
77 }
78 }这里先设置了channel的eventLoop属性，然后在接下来的一段代码中判断当前线程是否是channel的处理线程，也就是是不是eventLoop的线程，如果不是那么就将注册作为一个任务用EventLoop.execute执行。按照这里的执行顺序，当前线程肯定不是eventLoop的线程，所以会执行else分支，其实eventLoop的线程也是在这个调用中启动的。最后的注册是在AbstractSelectableChannel类的register()方法中执行的。这里有个很奇怪的地方，这里注册的ops是0，也就是没有感兴趣的事件。这个地方我们后面在分析。
将channel注册到selector的代码就是这些了，我们回头分析EventLoop.execute(…)，其实注册的代码是在这里面被调用的。
1 // SingleThreadEventExecutor
2 public void execute(Runnable task) {
3 if (task == null) {
4 throw new NullPointerException("task");
5 }
6
7 boolean inEventLoop = inEventLoop();
8 if (inEventLoop) {
9 addTask(task);
10 } else {
11 startThread();
12 addTask(task);
13 if (isShutdown() && removeTask(task)) {
14 reject();
15 }
16 }
17
18 if (!addTaskWakesUp && wakesUpForTask(task)) {
19 wakeup(inEventLoop);
20 }
21 }如果当前线程是EventLoop的线程，就把task加到任务队列中去，如果不是，那么启动线程，然后再把task加入到任务队列。
1 // SingleThreadEventLoop
2 private void startThread() {
3 if (STATE_UPDATER.get(this) == ST_NOT_STARTED) {
4 if (STATE_UPDATER.compareAndSet(this, ST_NOT_STARTED, ST_STARTED)) {
5 doStartThread();
6 }
7 }
8 }
9
10 private void doStartThread() {
11 assert thread == null;
12 executor.execute(new Runnable() {
13 @Override
14 public void run() {
15 thread = Thread.currentThread();
16 if (interrupted) {
17 thread.interrupt();
18 }
19
20 boolean success = false;
21 updateLastExecutionTime();
22 try {
23 SingleThreadEventExecutor.this.run();
24 success = true;
25 } catch (Throwable t) {
26 logger.warn("Unexpected exception from an event executor: ", t);
27 } finally {
28 // Some clean work
29 }
30 }
31 });
32 }其实最后的线程还是要落到EventLoop中的executor里面，而NioEventLoop初始化的时候executor属性设置的是一个ThreadPerTaskExecutor，顾名思义也就是每个任务新建一个线程去执行，而在这个Task里面对EventLoop的thread属性进行了设置，并且最后执行SingleThreadEventExecutor.this.run()，这个run方法在NioEventLoop中实现。
1 protected void run() {
2 for (;;) {
3 try {
4 switch (selectStrategy.calculateStrategy(selectNowSupplier, hasTasks())) {
5 case SelectStrategy.CONTINUE:
6 continue;
7 case SelectStrategy.SELECT:
8 select(wakenUp.getAndSet(false));
9 if (wakenUp.get()) {
10 selector.wakeup();
11 }
12 default:
13 // fallthrough
14 }
15
16 cancelledKeys = 0;
17 needsToSelectAgain = false;
18 final int ioRatio = this.ioRatio;
19 if (ioRatio == 100) {
20 processSelectedKeys();
21 runAllTasks();
22 } else {
23 final long ioStartTime = System.nanoTime();
24
25 processSelectedKeys();
26
27 final long ioTime = System.nanoTime() - ioStartTime;
28 runAllTasks(ioTime * (100 - ioRatio) / ioRatio);
29 }
30
31 if (isShuttingDown()) {
32 closeAll();
33 if (confirmShutdown()) {
34 break;
35 }
36 }
37 } catch (Throwable t) {
38 logger.warn("Unexpected exception in the selector loop.", t);
39
40 // Prevent possible consecutive immediate failures that lead to
41 // excessive CPU consumption.
42 try {
43 Thread.sleep(1000);
44 } catch (InterruptedException e) {
45 // Ignore.
46 }
47 }
48 }
49 }
50
51 private void processSelectedKeysPlain(Set<SelectionKey> selectedKeys) {
52 if (selectedKeys.isEmpty()) {
53 return;
54 }
55
56 Iterator<SelectionKey> i = selectedKeys.iterator();
57 for (;;) {
58 final SelectionKey k = i.next();
59 final Object a = k.attachment();
60 i.remove();
61
62 if (a instanceof AbstractNioChannel) {
63 // 处理ServerSocketChannl的事件，如accept
64 processSelectedKey(k, (AbstractNioChannel) a);
65 } else {
66 @SuppressWarnings("unchecked")
67 NioTask<SelectableChannel> task = (NioTask<SelectableChannel>) a;
68 processSelectedKey(k, task);
69 }
70
71 if (!i.hasNext()) {
72 break;
73 }
74
75 if (needsToSelectAgain) {
76 selectAgain();
77 selectedKeys = selector.selectedKeys();
78
79 // Create the iterator again to avoid ConcurrentModificationException
80 if (selectedKeys.isEmpty()) {
81 break;
82 } else {
83 i = selectedKeys.iterator();
84 }
85 }
86 }
87 }这个就是Netty最后的Reactor模式的事件循环了，在这个循环中调用selector的select方法查询需要处理的key，然后processSelectedKeys方法进行处理。在这里因为之前在注册NioServerSocketChannel的时候把channel当作attachment当做attachment，所以如果key的attachement是AbstractNioChannel说明这个是ServerSocketChannel的事件，如connect，read，accept。
其实还有一些问题没有写清楚，如下：
1. ServerSocketChannel的interestOps的注册 2. accept请求的处理 3. 线程模型 4. pipeline的链式调用 5. buffer 。。。 这些我会继续写文章进行说明～(希望可以=_=)
内容来自用户分享和网络整理，不保证内容的准确性，如有侵权内容，可联系管理员处理 
相关文章推荐
- Netty源码阅读（一） ServerBootstrap启动
- Netty源码阅读（一） ServerBootstrap启动
- Netty源码解读------------ServerBootstrap的启动(一)
- Netty源码－ServerBootStrap服务端启动流程
- Netty源码学习-ServerBootstrap启动及事件处理过程
- Netty3.5.9源码（一）Server端启动
- Java NIO框架Netty教程（四） – ServerBootStrap启动流程源码分析
- 【Netty源码学习】ServerBootStrap
- Java NIO框架Netty教程（四） – ServerBootStrap启动流程源码分析
- 【Netty源码学习】ServerBootStrap
- Netty源码阅读（一） ServerBootstrap启动
- Netty4启动ServerBootStrap源码分析
- netty3.2.3源码分析--ServerBootstrap启动分析
- netty3.2.3源码分析--ServerBootstrap启动分析
- Tomcat源码阅读之Bootstrap启动流程与classLoader设计
- Tomcat源码阅读之Server.xml文件的处理与Catalina启动流程
- java netty之ServerBootstrap的启动
- Java NIO框架Netty教程（四） – ServerBootStrap启动流程源码分析
- Java NIO框架Netty教程（四） – ServerBootStrap启动流程源码分析
- 【Netty源码学习】ServerBootStrap