深入浅出Netty源码剖析Java学习笔记

Netty源码分析:1.4服务器启动流程

2018-04-17  本文已影响32人  蓝汝丶琪

第一章节是主要是服务器启动的代码分析。章节目录有:
|---------1.1初始化NioEventLoopGroup
|---------1.2初始化NioEventLoop
|---------1.3初始化NioServerSocketChannel
|---------1.4服务器启动流程
为什么先从初始化开始了解服务器启动?
因为在我看服务器启动的相关源码的时候,有很多地方都是初始化的时候已经建立好的。所以我就从初始化的源码开始看起。这是我第一次看源码的笔记,仍有很多理解错误的地方和不解的地方。欢迎讨论。

本篇目录

启动服务器代码

 public void bind() {
 
         //1 
        EventLoopGroup boss = new NioEventLoopGroup();
        EventLoopGroup work = new NioEventLoopGroup();
        try {
//            引导绑定和启动服务器
            //2
             ServerBootstrap b = new ServerBootstrap();
             //3
             b.group(boss, work);
//            创建NioEventLoopGroup对象来处理事件,如接受新连接、接收数据、写数据等等
            //4
             b.channel(NioServerSocketChannel.class);
//            设置childHandler执行所有的连接请求
            //5
             b.childHandler(new ChildChannelHandler());
             //6
             b.option(ChannelOption.SO_BACKLOG, 100);
             
            //绑定端口
            //7
            ChannelFuture future = b.bind(8080).sync();

            //8
            future.channel().closeFuture().sync();
        } catch (InterruptedException e) {
            e.printStackTrace();
        }finally {
            //9
            boss.shutdownGracefully();
            work.shutdownGracefully();
        }

    }

3 Group

赋值两个线程池。
1 首先第一个boss线程池是赋值到父类AbstractBootstrap的group变量中。
2 work线程池就赋值在ServerBootstrap的childGroup变量中。

public ServerBootstrap group(EventLoopGroup parentGroup, EventLoopGroup childGroup) {
        //1
        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;
    }

4 channel()

此时channel 还没被创建,直到bind()方法中调用initAndRegister()方法的时候才会用该工厂类生成一个channel。下文会讲。

 public B channel(Class<? extends C> channelClass) {
        if (channelClass == null) {
            throw new NullPointerException("channelClass");
        }
        return channelFactory(new ReflectiveChannelFactory<C>(channelClass));
    }

5 childHandler()

在调用这个childHandler()这个方法的时候,你需要自己写一个方法继承ChannelInitializer类。而ChannelInitializer类也是继承ChannelInboundHandlerAdapter的。所以childHandler()方法的参数就是我们自己写的类。然后赋值到childHandler参数。

public ServerBootstrap childHandler(ChannelHandler childHandler) {
        if (childHandler == null) {
            throw new NullPointerException("childHandler");
        }
        this.childHandler = childHandler;
        return this;
    }

6 Option()

将一些配置存储到options变量中,该变量是一个LinkedHashMap

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;
    }

7 bind()

image.png

之前的一些方法都只是对ServerBootstrap的配置,说白了就是用来set参数的。
bind()则是开始启动服务器了。

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

doBind()

 private ChannelFuture doBind(final SocketAddress localAddress) {
        final ChannelFuture regFuture = initAndRegister(); //1
        final Channel channel = regFuture.channel();
        if (regFuture.cause() != null) {
            return regFuture;
        }
        //2
        if (regFuture.isDone()) {
            ChannelPromise promise = channel.newPromise();
            doBind0(regFuture, channel, localAddress, promise);
            return promise;
        } else { 
            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) {
                       
                        promise.setFailure(cause);
                    } else {
                      
                        promise.registered();

                        doBind0(regFuture, channel, localAddress, promise);
                    }
                }
            });
            return promise;
        }
    }


doBind0()

到这个方法。服务器启动就已经完成了

 private static void doBind0(
            final ChannelFuture regFuture, final Channel channel,
            final SocketAddress localAddress, final ChannelPromise promise) {
        //1
        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());
                }
            }
        });
    }

下文都是对initAndRegister()方法的代码解析

initAndRegister()

final ChannelFuture initAndRegister() {
        Channel channel = null;
        try {
           //1
            channel = channelFactory.newChannel();
            //2
            init(channel);
        } catch (Throwable t) {
          
        }
        //3
        ChannelFuture regFuture = config().group().register(channel);
        if (regFuture.cause() != null) {
            if (channel.isRegistered()) {
                channel.close();
            } else {
                channel.unsafe().closeForcibly();
            }
        }
        return regFuture;
    }

init()

void init(Channel channel) throws Exception {
        final Map<ChannelOption<?>, Object> options = options0();
        //1 
        synchronized (options) {
            setChannelOptions(channel, options, logger);
        }

        //2
        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());
            }
        }
        
        //3
        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()));
        }
        
        //4
        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));
                    }
                });
            }
        });
    }

config().group().register(channel)

 @Override
    public EventLoop next() {
        return (EventLoop) super.next();
    }
@Override
    public ChannelFuture register(Channel channel) {
        return next().register(channel); //1
    }

继续看下去。

@Override
    public ChannelFuture register(Channel channel) {
       //2
        return register(new DefaultChannelPromise(channel, this));
    }

继续往下看

 @Override
    public ChannelFuture register(final ChannelPromise promise) {
        //3
        ObjectUtil.checkNotNull(promise, "promise");
        promise.channel().unsafe().register(this, promise);
        return promise;
    }

继续往下分析,该register()方法是AbstractChannel类的。也就是初始化NioServerSocketChannel的时候,建立pipeline和unsafe的类。

 @Override
        public final void register(EventLoop eventLoop, final ChannelPromise promise) {
            //4
            if (eventLoop == null) {
                throw new NullPointerException("eventLoop");
            }
            if (isRegistered()) {
                promise.setFailure(new IllegalStateException("registered to an event loop already"));
                return;
            }
            if (!isCompatible(eventLoop)) {
                promise.setFailure(
                        new IllegalStateException("incompatible event loop type: " + eventLoop.getClass().getName()));
                return;
            }
            //5
            AbstractChannel.this.eventLoop = eventLoop;
            //6
            if (eventLoop.inEventLoop()) {
                register0(promise);
            } else {
                try {
                    eventLoop.execute(new Runnable() {
                        @Override
                        public void run() {
                        //7
                            register0(promise);
                        }
                    });
                } catch (Throwable t) {
                  
                }
            }
        }

首先在研究register0()之前有一个小知识。

在版本4.1.x 的时候,在初始化eventLoop的时候,还没有创建线程,而是保存了Executor这个变量。这个变量在4.0.x版本的时候是没有的。那么4.1.x版本在什么时候创建线程呢?在eventLoop调用execute()方法的时候创建线程。下文可以看到

 @Override
    public boolean inEventLoop(Thread thread) {
        return thread == this.thread;
    }
image.png
public void execute(Runnable task) {
        if (task == null) {
            throw new NullPointerException("task");
        }
        //1
        boolean inEventLoop = inEventLoop();
        if (inEventLoop) {
            addTask(task);
        } else {
             //2
            startThread();
            //3
            addTask(task);
            if (isShutdown() && removeTask(task)) {
                reject();
            }
        }
        if (!addTaskWakesUp && wakesUpForTask(task)) {
            wakeup(inEventLoop);
        }
    }

register0(ChannelPromise promise)

 private void register0(ChannelPromise promise) {
            try {
                //1
                if (!promise.setUncancellable() || !ensureOpen(promise)) {
                    return;
                }
                //2
                boolean firstRegistration = neverRegistered;
                doRegister();//3
                neverRegistered = false;
                registered = true;

      
                pipeline.invokeHandlerAddedIfNeeded();

                safeSetSuccess(promise);
                pipeline.fireChannelRegistered();
            
                if (isActive()) {
                    if (firstRegistration) {
                        pipeline.fireChannelActive();
                    } else if (config().isAutoRead()) {
                     
                        beginRead();
                    }
                }
            } catch (Throwable t) {
            
            }
        }
上一篇下一篇

猜你喜欢

热点阅读