日韩无码专区无码一级三级片|91人人爱网站中日韩无码电影|厨房大战丰满熟妇|AV高清无码在线免费观看|另类AV日韩少妇熟女|中文日本大黄一级黄色片|色情在线视频免费|亚洲成人特黄a片|黄片wwwav色图欧美|欧亚乱色一区二区三区

本系列Netty源碼解析文章基于 4.1.56.Final版本。

大家第一眼看到這幅流程圖，是不是腦瓜子嗡嗡的呢?

大家先不要驚慌，問題不大，本文筆者的目的就是要讓大家清晰的理解這幅流程圖，從而深刻的理解Netty Reactor的啟動(dòng)全流程，包括其中涉及到的各種代碼設(shè)計(jì)實(shí)現(xiàn)細(xì)節(jié)。

在上篇文章??《聊聊Netty那些事兒之Reactor在Netty中的實(shí)現(xiàn)(創(chuàng)建篇)》??中我們?cè)敿?xì)介紹了Netty服務(wù)端核心引擎組件主從Reactor組模型 NioEventLoopGroup以及Reactor模型 NioEventLoop的創(chuàng)建過程。最終我們得到了netty Reactor模型的運(yùn)行骨架如下：

現(xiàn)在Netty服務(wù)端程序的骨架是搭建好了，本文我們就基于這個(gè)骨架來深入剖析下Netty服務(wù)端的啟動(dòng)過程。

我們繼續(xù)回到上篇文章提到的Netty服務(wù)端代碼模板中，在創(chuàng)建完主從Reactor線程組:bossGroup，workerGroup后，接下來就開始配置Netty服務(wù)端的啟動(dòng)輔助類ServerBootstrap了。

public final class EchoServer {
    static final int PORT = Integer.parseInt(System.getProperty("port", "8007"));
    public static void main(String[] args) throws Exception {
        // Configure the server.
        //創(chuàng)建主從Reactor線程組
        EventLoopGroup bossGroup = new NioEventLoopGroup(1);
        EventLoopGroup workerGroup = new NioEventLoopGroup();
        final EchoServerHandler serverHandler = new EchoServerHandler();
        try {
            ServerBootstrap b = new ServerBootstrap();
            b.group(bossGroup, workerGroup)//配置主從Reactor
             .channel(NioServerSocketChannel.class)//配置主Reactor中的channel類型
             .option(ChannelOption.SO_BACKLOG, 100)//設(shè)置主Reactor中channel的option選項(xiàng)
             .handler(new LoggingHandler(LogLevel.INFO))//設(shè)置主Reactor中Channel->pipline->handler
             .childHandler(new ChannelInitializer() {//設(shè)置從Reactor中注冊(cè)channel的pipeline
                 @Override
                 public void initChannel(SocketChannel ch) throws Exception {
                     ChannelPipeline p = ch.pipeline();
                     //p.addLast(new LoggingHandler(LogLevel.INFO));
                     p.addLast(serverHandler);
                 }
             });
            // Start the server. 綁定端口啟動(dòng)服務(wù)，開始監(jiān)聽accept事件
            ChannelFuture f = b.bind(PORT).sync();
            // Wait until the server socket is closed.
            f.channel().closeFuture().sync();
        } finally {
            // Shut down all event loops to terminate all threads.
            bossGroup.shutdownGracefully();
            workerGroup.shutdownGracefully();
        }
    }
}

在上篇文章中我們對(duì)代碼模板中涉及到ServerBootstrap的一些配置方法做了簡單的介紹，大家如果忘記的話，可以在返回去回顧一下。

ServerBootstrap類其實(shí)沒有什么特別的邏輯，主要是對(duì)Netty啟動(dòng)過程中需要用到的一些核心信息進(jìn)行配置管理，比如：

• Netty的核心引擎組件主從Reactor線程組：bossGroup，workerGroup。通過ServerBootstrap#group方法配置。
• Netty服務(wù)端使用到的Channel類型：NioServerSocketChannel ,通過ServerBootstrap#channel方法配置。以及配置NioServerSocketChannel時(shí)用到的SocketOption。SocketOption用于設(shè)置底層JDK NIO Socket的一些選項(xiàng)。通過ServerBootstrap#option方法進(jìn)行配置。

主ReactorGroup中的MainReactor管理的Channel類型為NioServerSocketChannel，如圖所示主要用來監(jiān)聽端口，接收客戶端連接，為客戶端創(chuàng)建初始化NioSocketChannel，然后采用round-robin輪詢的方式從圖中從ReactorGroup中選擇一個(gè)SubReactor與該客戶端NioSocketChannel進(jìn)行綁定。

從ReactorGroup中的SubReactor管理的Channel類型為NioSocketChannel，它是netty中定義客戶端連接的一個(gè)模型，每個(gè)連接對(duì)應(yīng)一個(gè)。如圖所示SubReactor負(fù)責(zé)監(jiān)聽處理綁定在其上的所有NioSocketChannel上的IO事件。

• 保存服務(wù)端NioServerSocketChannel和客戶端NioSocketChannel對(duì)應(yīng)pipeline中指定的ChannelHandler。用于后續(xù)Channel向Reactor注冊(cè)成功之后，初始化Channel里的pipeline。

不管是服務(wù)端用到的NioServerSocketChannel還是客戶端用到的NioSocketChannel，每個(gè)Channel實(shí)例都會(huì)有一個(gè)Pipeline，Pipeline中有多個(gè)ChannelHandler用于編排處理對(duì)應(yīng)Channel上感興趣的IO事件。

ServerBootstrap結(jié)構(gòu)中包含了netty服務(wù)端程序啟動(dòng)的所有配置信息，在我們介紹啟動(dòng)流程之前，先來看下ServerBootstrap的源碼結(jié)構(gòu)：

ServerBootstrap

ServerBootstrap的繼承結(jié)構(gòu)比較簡單，繼承層次的職責(zé)分工也比較明確。

ServerBootstrap主要負(fù)責(zé)對(duì)主從Reactor線程組相關(guān)的配置進(jìn)行管理，其中帶child前綴的配置方法是對(duì)從Reactor線程組的相關(guān)配置管理。從Reactor線程組中的Sub Reactor負(fù)責(zé)管理的客戶端NioSocketChannel相關(guān)配置存儲(chǔ)在ServerBootstrap結(jié)構(gòu)中。

父類AbstractBootstrap則是主要負(fù)責(zé)對(duì)主Reactor線程組相關(guān)的配置進(jìn)行管理，以及主Reactor線程組中的Main Reactor負(fù)責(zé)處理的服務(wù)端ServerSocketChannel相關(guān)的配置管理。

一、 配置主從Reactor線程組

ServerBootstrap b = new ServerBootstrap();
b.group(bossGroup, workerGroup)//配置主從Reactor

public class ServerBootstrap extends AbstractBootstrap {
     //Main Reactor線程組
    volatile EventLoopGroup group;
    //Sub Reactor線程組
    private volatile EventLoopGroup childGroup;
    public ServerBootstrap group(EventLoopGroup parentGroup, EventLoopGroup childGroup) {
        //父類管理主Reactor線程組
        super.group(parentGroup);
        if (this.childGroup != null) {
            throw new IllegalStateException("childGroup set already");
        }
        this.childGroup = ObjectUtil.checkNotNull(childGroup, "childGroup");
        return this;
    }
}

二、 配置服務(wù)端ServerSocketChannel

ServerBootstrap b = new ServerBootstrap();
b.channel(NioServerSocketChannel.class);

public class ServerBootstrap extends AbstractBootstrap {
    //用于創(chuàng)建ServerSocketChannel  ReflectiveChannelFactory
    private volatile ChannelFactory channelFactory;
    public B channel(Class channelClass) {
        return channelFactory(new ReflectiveChannelFactory(
                ObjectUtil.checkNotNull(channelClass, "channelClass")
        ));
    }
    @Deprecated
    public B channelFactory(ChannelFactory channelFactory) {
        ObjectUtil.checkNotNull(channelFactory, "channelFactory");
        if (this.channelFactory != null) {
            throw new IllegalStateException("channelFactory set already");
        }
        this.channelFactory = channelFactory;
        return self();
    }
}

在向ServerBootstrap配置服務(wù)端ServerSocketChannel的channel方法中，其實(shí)是創(chuàng)建了一個(gè)ChannelFactory工廠實(shí)例ReflectiveChannelFactory,在Netty服務(wù)端啟動(dòng)的過程中，會(huì)通過這個(gè)ChannelFactory去創(chuàng)建相應(yīng)的Channel實(shí)例。

我們可以通過這個(gè)方法來配置netty的IO模型，下面為ServerSocketChannel在不同IO模型下的實(shí)現(xiàn)：

EventLoopGroupReactor線程組在不同IO模型下的實(shí)現(xiàn)：

我們只需要將IO模型的這些核心接口對(duì)應(yīng)的實(shí)現(xiàn)類前綴改為對(duì)應(yīng)IO模型的前綴，就可以輕松在Netty中完成對(duì)IO模型的切換。

1、 ReflectiveChannelFactory

public class ReflectiveChannelFactory implements ChannelFactory {
    //NioServerSocketChannelde 構(gòu)造器
    private final Constructor constructor;
    public ReflectiveChannelFactory(Class clazz) {
        ObjectUtil.checkNotNull(clazz, "clazz");
        try {
            //反射獲取NioServerSocketChannel的構(gòu)造器
            this.constructor = clazz.getConstructor();
        } catch (NoSuchMethodException e) {
            throw new IllegalArgumentException("Class " + StringUtil.simpleClassName(clazz) +
                    " does not have a public non-arg constructor", e);
        }
    }
    @Override
    public T newChannel() {
        try {
            //創(chuàng)建NioServerSocketChannel實(shí)例
            return constructor.newInstance();
        } catch (Throwable t) {
            throw new ChannelException("Unable to create Channel from class " + constructor.getDeclaringClass(), t);
        }
    }
}

從類的簽名我們可以看出，這個(gè)工廠類是通過泛型加反射的方式來創(chuàng)建對(duì)應(yīng)的Channel實(shí)例。

• 泛型參數(shù)T extends Channel表示的是要通過工廠類創(chuàng)建的Channel類型，這里我們初始化的是NioServerSocketChannel。
• 在ReflectiveChannelFactory的構(gòu)造器中通過反射的方式獲取NioServerSocketChannel的構(gòu)造器。
• 在newChannel方法中通過構(gòu)造器反射創(chuàng)建NioServerSocketChannel實(shí)例。

注意這時(shí)只是配置階段，NioServerSocketChannel此時(shí)并未被創(chuàng)建。它是在啟動(dòng)的時(shí)候才會(huì)被創(chuàng)建出來。

三、為NioServerSocketChannel配置ChannelOption

ServerBootstrap b = new ServerBootstrap();
//設(shè)置被MainReactor管理的NioServerSocketChannel的Socket選項(xiàng)
b.option(ChannelOption.SO_BACKLOG, 100)

public abstract class AbstractBootstrap, C extends Channel> implements Cloneable {
    //serverSocketChannel中的ChannelOption配置
    private final Map, Object> options = new LinkedHashMap, Object>();
    public  B option(ChannelOption option, T value) {
        ObjectUtil.checkNotNull(option, "option");
        synchronized (options) {
            if (value == null) {
                options.remove(option);
            } else {
                options.put(option, value);
            }
        }
        return self();
    }
}

無論是服務(wù)端的NioServerSocketChannel還是客戶端的NioSocketChannel它們的相關(guān)底層Socket選項(xiàng)ChannelOption配置全部存放于一個(gè)Map類型的數(shù)據(jù)結(jié)構(gòu)中。

由于客戶端NioSocketChannel是由從Reactor線程組中的Sub Reactor來負(fù)責(zé)處理，所以涉及到客戶端NioSocketChannel所有的方法和配置全部是以child前綴開頭。

ServerBootstrap b = new ServerBootstrap();
.childOption(ChannelOption.TCP_NODELAY, Boolean.TRUE)

public class ServerBootstrap extends AbstractBootstrap {
   //客戶端SocketChannel對(duì)應(yīng)的ChannelOption配置
    private final Map, Object> childOptions = new LinkedHashMap, Object>();
    public  ServerBootstrap childOption(ChannelOption childOption, T value) {
        ObjectUtil.checkNotNull(childOption, "childOption");
        synchronized (childOptions) {
            if (value == null) {
                childOptions.remove(childOption);
            } else {
                childOptions.put(childOption, value);
            }
        }
        return this;
    }
}

相關(guān)的底層Socket選項(xiàng)，netty全部枚舉在ChannelOption類中，筆者這里就不一一列舉了，在本系列后續(xù)相關(guān)的文章中，筆者還會(huì)為大家詳細(xì)的介紹這些參數(shù)的作用。

public class ChannelOption extends AbstractConstant> {
    ..................省略..............
    public static final ChannelOption SO_BROADCAST = valueOf("SO_BROADCAST");
    public static final ChannelOption SO_KEEPALIVE = valueOf("SO_KEEPALIVE");
    public static final ChannelOption SO_SNDBUF = valueOf("SO_SNDBUF");
    public static final ChannelOption SO_RCVBUF = valueOf("SO_RCVBUF");
    public static final ChannelOption SO_REUSEADDR = valueOf("SO_REUSEADDR");
    public static final ChannelOption SO_LINGER = valueOf("SO_LINGER");
    public static final ChannelOption SO_BACKLOG = valueOf("SO_BACKLOG");
    public static final ChannelOption SO_TIMEOUT = valueOf("SO_TIMEOUT");
    ..................省略..............
}

四、為服務(wù)端NioServerSocketChannel中的Pipeline配置ChannelHandler

   //serverSocketChannel中pipeline里的handler(主要是acceptor)
    private volatile ChannelHandler handler;
    public B handler(ChannelHandler handler) {
        this.handler = ObjectUtil.checkNotNull(handler, "handler");
        return self();
    }

  ServerBootstrap b = new ServerBootstrap();
            b.group(bossGroup, workerGroup)//配置主從Reactor
             .channel(NioServerSocketChannel.class)//配置主Reactor中的channel類型
             .handler(new ChannelInitializer() {
                 @Override
                 protected void initChannel(NioServerSocketChannel ch) throws Exception {
                     ChannelPipeline p = ch.pipeline();
                     p.addLast(channelhandler1)
                      .addLast(channelHandler2)                    
                      ......                     
                      .addLast(channelHandler3);
                 }
             })

  final EchoServerHandler serverHandler = new EchoServerHandler();
            serverBootstrap.childHandler(new ChannelInitializer() {//設(shè)置從Reactor中注冊(cè)channel的pipeline
                 @Override
                 public void initChannel(SocketChannel ch) throws Exception {
                     ChannelPipeline p = ch.pipeline();            
                     p.addLast(new LoggingHandler(LogLevel.INFO));
                     p.addLast(serverHandler);
                 }
             });

    //socketChannel中pipeline中的處理handler
    private volatile ChannelHandler childHandler;
    public ServerBootstrap childHandler(ChannelHandler childHandler) {
        this.childHandler = ObjectUtil.checkNotNull(childHandler, "childHandler");
        return this;
    }

public abstract class ChannelInitializer extends ChannelInboundHandlerAdapter {
    @Override
    @SuppressWarnings("unchecked")
    public final void channelRegistered(ChannelHandlerContext ctx) throws Exception {
        //當(dāng)channelRegister事件發(fā)生時(shí)，調(diào)用initChannel初始化pipeline
        if (initChannel(ctx)) {
                 .................省略...............
        } else {
                 .................省略...............
        }
    }
    private boolean initChannel(ChannelHandlerContext ctx) throws Exception {
        if (initMap.add(ctx)) { // Guard against re-entrance.
            try {
                //此時(shí)客戶單NioSocketChannel已經(jīng)創(chuàng)建并初始化好了
                initChannel((C) ctx.channel());
            } catch (Throwable cause) {
                 .................省略...............
            } finally {
                  .................省略...............
            }
            return true;
        }
        return false;
    }
    protected abstract void initChannel(C ch) throws Exception;    
    .................省略...............
}

   final EchoServerHandler serverHandler = new EchoServerHandler();
            serverBootstrap.childHandler(new ChannelInitializer() {//設(shè)置從Reactor中注冊(cè)channel的pipeline
                 @Override
                 public void initChannel(SocketChannel ch) throws Exception {
                     ChannelPipeline p = ch.pipeline();           
                     p.addLast(new LoggingHandler(LogLevel.INFO));
                     p.addLast(serverHandler);
                 }
             });

// Start the server. 綁定端口啟動(dòng)服務(wù)，開始監(jiān)聽accept事件
ChannelFuture f = serverBootStrap.bind(PORT).sync();

    public ChannelFuture bind(int inetPort) {
        return bind(new InetSocketAddress(inetPort));
    }
    public ChannelFuture bind(SocketAddress localAddress) {
        //校驗(yàn)Netty核心組件是否配置齊全
        validate();
        //服務(wù)端開始啟動(dòng)，綁定端口地址，接收客戶端連接
        return doBind(ObjectUtil.checkNotNull(localAddress, "localAddress"));
    }
   private ChannelFuture doBind(final SocketAddress localAddress) {
        //異步創(chuàng)建，初始化，注冊(cè)ServerSocketChannel到main reactor上
        final ChannelFuture regFuture = initAndRegister();
        final Channel channel = regFuture.channel();
        if (regFuture.cause() != null) {
            return regFuture;
        }
        if (regFuture.isDone()) {   
           ........serverSocketChannel向Main Reactor注冊(cè)成功后開始綁定端口....,                            
        } else {
            //http://www.5511xx.com/article/cdsohce.html