Netty服務端啓動流程分析

時間  2019-11-06
標籤 netty 服務 啓動 流程 分析 欄目 Netty 简体版
原文   原文鏈接

事例代碼

源碼閱讀過程當中，咱們使用下面這個簡單的示例代碼作參考；java

EventLoopGroup parentGroup = new NioEventLoopGroup(1);
    EventLoopGroup childGroup = new NioEventLoopGroup();
    try {
        ServerBootstrap serverBootstrap = new ServerBootstrap();
        serverBootstrap
                .group(parentGroup, childGroup)
                .channel(NioServerSocketChannel.class)
                .handler(new LoggingHandler(LogLevel.INFO))
                .childOption(ChannelOption.TCP_NODELAY, true)
                .childAttr(AttributeKey.newInstance("childAttr"), "childAttrVal")
                .childHandler(new ChannelInitializer<SocketChannel>() {
                    @Override
                    protected void initChannel(SocketChannel ch) {
                        ChannelPipeline pipeline = ch.pipeline();
                        pipeline.addLast("httpServerCodec", new HttpServerCodec());
                        pipeline.addLast("testHttpServerHandler", new TestHttpServerHandler());
                        pipeline.addLast(new LoggingHandler(LogLevel.INFO));
                    }
                });
        ChannelFuture channelFuture = serverBootstrap.bind(8080).sync();
        channelFuture.channel().closeFuture().sync();
    } catch (Exception e) {
        e.printStackTrace();
    } finally {
        parentGroup.shutdownGracefully();
        childGroup.shutdownGracefully();
    }
複製代碼

建立服務端Channel

步驟

  • bind()
  • initAndRegister()
  • newChannel() 反射建立服務端Channel
    • newSocket() 經過jdk來建立底層的Channel
    • NioServerSocketChannelConfig 設置TCP相關參數
    • AbstractNioChannel
      • configureBlocking
      • AbStractChannel() (id unsafe pipeline)

分析

咱們從bind(port)開始閱讀Netty服務端的建立初始化過程算法

serverBootstrap.bind(8080)
複製代碼

順着bind方法咱們一直跟到AbstractBootStrap.doBind(final SocketAddress localAddress)promise

private ChannelFuture doBind(final SocketAddress localAddress) {
            // 初始化和註冊
            final ChannelFuture regFuture = initAndRegister();
            ...
    }
複製代碼

initAndRegister()bash

final ChannelFuture initAndRegister() {
        Channel channel = null;
        try {
            // 建立NioServerChannel
            channel = channelFactory.newChannel();
            init(channel);
        } catch (Throwable t) {
            ...
        }
        ...
        return regFuture;
    }
複製代碼

建立NioServerChannel的channelFactory是什麼時候被初始化的呢？在事例代碼中對啓動引導類有以下設置：app

serverBootstrap
            .channel(NioServerSocketChannel.class)
複製代碼

看下channel(Class<? extends C> channelClass)方法作了什麼socket

public B channel(Class<? extends C> channelClass) {
        if (channelClass == null) {
            throw new NullPointerException("channelClass");
        }
        // 返回一個建立channel的工廠
        return channelFactory(new ReflectiveChannelFactory<C>(channelClass));
    }
複製代碼

咱們來看下ReflectiveChannelFactory類的實現ide

public class ReflectiveChannelFactory<T extends Channel> implements ChannelFactory<T> {
    private final Constructor<? extends T> constructor;
    public ReflectiveChannelFactory(Class<? extends T> clazz) {
        ObjectUtil.checkNotNull(clazz, "clazz");
        try {
            // 設置傳進來的Channel的構造器
            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 {
            // 經過構造器實例化對象
            return constructor.newInstance();
        } catch (Throwable t) {
            throw new ChannelException("Unable to create Channel from class " + constructor.getDeclaringClass(), t);
        }
    }
    ...
}
複製代碼

跟到這咱們發現channelFactory.newChannel()實際上調用的就是NioServerSocketChannel的無參構造方法：oop

  • NioServerSocketChannel無參構造方法建立Channel時包括如下步驟:
    • newSocket建立ServerSocketChannel
    • 調用父類構造器 註冊SelectionKey.OP_ACCEPT事件 初始化 id unsafe pipeline
public class NioServerSocketChannel extends AbstractNioMessageChannel
                             implements io.netty.channel.socket.ServerSocketChannel {
    ....                             
    private static final SelectorProvider DEFAULT_SELECTOR_PROVIDER = SelectorProvider.provider();
    private final ServerSocketChannelConfig config;
    private static ServerSocketChannel newSocket(SelectorProvider provider) {
        ...
        return provider.openServerSocketChannel();
        ...
    }
    public NioServerSocketChannel() {
        // 1.newSocket建立ServerSocketChannel
        this(newSocket(DEFAULT_SELECTOR_PROVIDER));
    }
    public NioServerSocketChannel(ServerSocketChannel channel) {
        // 2.調用父類構造器 賦值SelectionKey.OP_ACCEPT事件 初始化 id unsafe pipeline
        super(null, channel, SelectionKey.OP_ACCEPT);
        // 3.設置config
        config = new NioServerSocketChannelConfig(this, javaChannel().socket());
    }
    ...
}
複製代碼
  • 父類AbstractNioChannel的構造方法
// 設置channel和readInterestOp以及把channel設置爲非阻塞的讀
    protected AbstractNioChannel(Channel parent, SelectableChannel ch, int readInterestOp) {
        super(parent);
        this.ch = ch; // 之後javaChannel()獲取到的serverSocketChannel
        this.readInterestOp = readInterestOp;
        try {
            ch.configureBlocking(false);
        } catch (IOException e) {
            ...
        }
    }
    protected AbstractChannel(Channel parent) {
        this.parent = parent;
        id = newId();
        // netty內部的對
        unsafe = newUnsafe();
        pipeline = newChannelPipeline();
    }
    // 初始化ChannelPipeline
    protected DefaultChannelPipeline(Channel channel) {
        ...
        tail = new TailContext(this);
        head = new HeadContext(this);
        head.next = tail;
        tail.prev = head;
        ...
    }
複製代碼

初始化服務端Channel

步驟

  • set(channelOptions, channelAttrs)
  • set(childOption, childAttrs)
  • config handler 配置服務端pipeline
  • new ServerBootstrapAcceptor

分析

  • ServerBootstrap.init()
void init(Channel channel) throws Exception {
        // 1.1 設置ServerSocketChannel的options
        final Map<ChannelOption<?>, Object> options = options0();
        synchronized (options) {
            setChannelOptions(channel, options, logger);
        }
        // 1.2 設置ServerSocketChannel的attrs
        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());
            }
        }
        ChannelPipeline p = channel.pipeline();
        final EventLoopGroup currentChildGroup = childGroup;
        final ChannelHandler currentChildHandler = childHandler;
        final Entry<ChannelOption<?>, Object>[] currentChildOptions;
        final Entry<AttributeKey<?>, Object>[] currentChildAttrs;
        // 2.1 設置childoption 如ChannelOption.TCP_NODELAY
        synchronized (childOptions) {
            currentChildOptions = childOptions.entrySet().toArray(newOptionArray(0));
        }
        // 2.2 設置childAttrs
        synchronized (childAttrs) {
            currentChildAttrs = childAttrs.entrySet().toArray(newAttrArray(0));
        }
        // 3.初始化acceptor
        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));
                    }
                });
            }
        });
    }
複製代碼
  • new ServerBootstrapAcceptor的核心在於它的channelRead方法，咱們先來簡單看下：
public void channelRead(ChannelHandlerContext ctx, Object msg) {
        // 獲取到SocketChannel，也就是連結接入的客戶端Channel
        final Channel child = (Channel) msg;
        ...
        try {
            // 將child註冊給其中一個childEventLoop的selector上。
            childGroup.register(child).addListener(new ChannelFutureListener() {
                ...
            });
        } catch (Throwable t) {
            forceClose(child, t);
        }
    }
複製代碼

註冊Selector

步驟

  • AbstractChannel.register(channel)入口
    • this.eventLoop = eventLoop
    • register0 實際註冊
      • doRegister() 調用jdk底層註冊
      • invoke HandlerAddedIfNeeded
      • fireChannelRegistered() 傳播事件

分析

  • 仍是看AbstactBootstrap.initAndRegister()
final ChannelFuture initAndRegister() {
        ...
        ChannelFuture regFuture = config().group().register(channel);
        ...
        return regFuture;
    }
複製代碼
  • 跟着register方法一直到AbstractUnsafe.register0(..)方法
private void register0(ChannelPromise promise) {
        try {
            ...
            // 1.doRegister() 調用jdk底層註冊
            doRegister();
            // 2.invoke HandlerAddedIfNeeded
            pipeline.invokeHandlerAddedIfNeeded();
            // 3.調用ServerSocketChannle的fireChannelRegistered
            pipeline.fireChannelRegistered();
            ...
        } catch (Throwable t) {
            ...
        }
    }
複製代碼
  • AbstractNioChannel.doRegister() 調用jdk底層註冊
protected void doRegister() throws Exception {
        ...
        for (;;) {
            try {
                // 將ServerSocketChannel註冊給selector，這塊0表明不關注任何事件，由於會在綁定的時候監聽OP_ACCEPT
                selectionKey = javaChannel().register(eventLoop().unwrappedSelector(), 0, this);
                return;
            } catch (CancelledKeyException e) {
                ...
            }
        }
    }
複製代碼
  • javaChannel(),在以前步驟AbstractNioChannel初始化的時候，javaChannel()其實在當前狀況下獲取到的就是ServerSocketChannnel
  • 這個註冊其實你能夠發現是服務端ServelSocketChannel的註冊，而咱們上邊發現，對於接入的客戶端調用acceptor的channelRead方法時，其實對應的過程是SocketChannel的註冊，對於客戶端讀寫事件的selector咱們一般起多個EventLoop，因此這塊會有個簡單的輪訓算法找到EventLoop，而後在將channel註冊到它的selector上，後面的文章會更加詳細的說着點。

完成端口綁定

步驟

  • AbstractUnsafe.bind()
    • doBind()
      • javaChannel().bind() jdk底層綁定
    • pipeline.fireChannelActive()
      • HeadContext.readIfIsAutoRead() 將selector綁定事件爲OP_ACCEPT事件

介紹

  • AbstractUnsafe.bind()
public final void bind(final SocketAddress localAddress, final ChannelPromise promise) {
        ...
        // 這個時候channel並非active因此爲false
        boolean wasActive = isActive();
        try {
            // 1.jdk ServerSocketServer綁定端口
            doBind(localAddress);
        } catch (Throwable t) {
            ...
            return;
        }
        // 由於上邊已經完成端口綁定因此 isActive()此時爲true
        if (!wasActive && isActive()) {
            invokeLater(new Runnable() {
                @Override
                public void run() {
                    // 2. 一直跟蹤最後發現調用的是 HeadContext.readIfIsAutoRead()
                    pipeline.fireChannelActive();
                }
            });
        }
        ...
    }
複製代碼
  • isActive()其實調用的與ServerSocketChannel綁定的ServerSocket的isBound方法
public boolean isActive() {
        return javaChannel().socket().isBound();
    }
    public boolean isBound() {
        // Before 1.3 ServerSockets were always bound during creation
        return bound || oldImpl;
    }
複製代碼
  • HeadContext.readIfIsAutoRead()再一直跟蹤 AbstractNioChannel.doBeginRead()完成的OP_ACCEPT事件綁定
protected void doBeginRead() throws Exception {
        final SelectionKey selectionKey = this.selectionKey;
        if (!selectionKey.isValid()) {
            return;
        }
        readPending = true;
        // interestOps 完成OP_ACCEPT事件綁定
        final int interestOps = selectionKey.interestOps();
        if ((interestOps & readInterestOp) == 0) {
            selectionKey.interestOps(interestOps | readInterestOp);
        }
    }
複製代碼

總結

寫到這，經過本文了解到了：ui

  • 服務端channel的建立過程
  • 服務端channel的初始化過程
  • 服務端channel向selector的註冊
  • 完成端口綁定，和底層的serverSocket激活
相關文章
相關標籤/搜索
每日一句
    每一个你不满意的现在，都有一个你没有努力的曾经。
本站公眾號
   歡迎關注本站公眾號,獲取更多信息
聯繫我們 最近搜索 最新文章 沪ICP备13005482号-10 MyBatis教程 SQL 教程 MySQL教程 Java 教程 Thymeleaf 教程 Hibernate教程 Spring教程 Redis教程