netty源碼分析 - register分析

系列

Netty源碼分析 - Bootstrap服務(wù)端
Netty源碼分析 - Bootstrap客戶端
netty源碼分析 - ChannelHandler
netty源碼分析 - EventLoop類關(guān)系
netty源碼分析 - register分析
Netty源碼分析 - NioEventLoop事件處理
netty源碼分析 - accept過程分析
Netty源碼分析 - ByteBuf
Netty源碼分析 - 粘包和拆包問題


開篇

  • 這篇文章主要用于分析NioSocketChannel注冊到NioEventLoop對象的過程，順帶和Nio的例子進行下對比。
  • 通過這篇文章能夠了解到整個注冊過程的脈絡(luò)為NioEventLoopGroup => NioEventLoop => NioSocketChannel的劉哦吃。


Nio的例子

public class NioEchoServer {
    private static final int BUF_SIZE = 256;
    private static final int TIMEOUT = 3000;

    public static void main(String args[]) throws Exception {
        // 打開服務(wù)端 Socket,ServerSocketChannel.open()
        // 內(nèi)部執(zhí)行的是 SelectorProvider.provider().openServerSocketChannel();
        ServerSocketChannel serverSocketChannel = ServerSocketChannel.open();

        // 打開 Selector
        Selector selector = Selector.open();

        // 服務(wù)端 Socket 監(jiān)聽8080端口, 并配置為非阻塞模式
        serverSocketChannel.socket().bind(new InetSocketAddress(8080));
        serverSocketChannel.configureBlocking(false);

        // 將 channel 注冊到 selector 中.
        // 通常我們都是先注冊一個 OP_ACCEPT 事件, 然后在 OP_ACCEPT 到來時, 再將這個 Channel 的 OP_READ
        // 注冊到 Selector 中.
        serverSocketChannel.register(selector, SelectionKey.OP_ACCEPT);

        while (true) {
            // 通過調(diào)用 select 方法, 阻塞地等待 channel I/O 可操作
            if (selector.select(TIMEOUT) == 0) {
                System.out.print(".");
                continue;
            }

            // 獲取 I/O 操作就緒的 SelectionKey, 通過 SelectionKey 可以知道哪些 Channel 的哪類 I/O 操作已經(jīng)就緒.
            Iterator<SelectionKey> keyIterator = selector.selectedKeys().iterator();

            while (keyIterator.hasNext()) {

                SelectionKey key = keyIterator.next();

                // 當獲取一個 SelectionKey 后, 就要將它刪除, 表示我們已經(jīng)對這個 IO 事件進行了處理.
                keyIterator.remove();

                if (key.isAcceptable()) {
                    // 當 OP_ACCEPT 事件到來時, 我們就有從 ServerSocketChannel 中獲取一個 SocketChannel,
                    // 代表客戶端的連接
                    // 注意, 在 OP_ACCEPT 事件中, 從 key.channel() 返回的 Channel 是 ServerSocketChannel.
                    // 而在 OP_WRITE 和 OP_READ 中, 從 key.channel() 返回的是 SocketChannel.
                    SocketChannel clientChannel = ((ServerSocketChannel) key.channel()).accept();
                    clientChannel.configureBlocking(false);
                    //在 OP_ACCEPT 到來時, 再將這個 Channel 的 OP_READ 注冊到 Selector 中.
                    // 注意, 這里我們?nèi)绻麤]有設(shè)置 OP_READ 的話, 即 interest set 仍然是 OP_CONNECT 的話, 那么 select 方法會一直直接返回.
                    clientChannel.register(key.selector(), OP_READ, ByteBuffer.allocate(BUF_SIZE));
                }

                if (key.isReadable()) {
                    SocketChannel clientChannel = (SocketChannel) key.channel();
                    ByteBuffer buf = (ByteBuffer) key.attachment();
                    long bytesRead = clientChannel.read(buf);
                    if (bytesRead == -1) {
                        clientChannel.close();
                    } else if (bytesRead > 0) {
                        key.interestOps(OP_READ | SelectionKey.OP_WRITE);
                        System.out.println("Get data length: " + bytesRead);
                    }
                }

                if (key.isValid() && key.isWritable()) {
                    ByteBuffer buf = (ByteBuffer) key.attachment();
                    buf.flip();
                    SocketChannel clientChannel = (SocketChannel) key.channel();

                    clientChannel.write(buf);

                    if (!buf.hasRemaining()) {
                        key.interestOps(OP_READ);
                    }
                    buf.compact();
                }
            }
        }
    }
}


Netty NioServer

NioServerSocketChannel

NioServerSocketChannel
  • NioServerSocketChannel的類繼承關(guān)系如上圖，核心關(guān)注NioServerSocketChannel和AbstraceNioChannel。
  • NioServerSocketChannel和AbstraceNioChannel類關(guān)注初始化順序。


public class NioServerSocketChannel extends AbstractNioMessageChannel
                             implements io.netty.channel.socket.ServerSocketChannel {

    private static final ChannelMetadata METADATA = new ChannelMetadata(false, 16);
    // SelectorProvider.provider()返回的是單例對象
    private static final SelectorProvider DEFAULT_SELECTOR_PROVIDER = SelectorProvider.provider();
    private final ServerSocketChannelConfig config;

    private static ServerSocketChannel newSocket(SelectorProvider provider) {
        try {
            return provider.openServerSocketChannel();
        } catch (IOException e) {
        }
    }

    public NioServerSocketChannel() {
        this(newSocket(DEFAULT_SELECTOR_PROVIDER));
    }

    public NioServerSocketChannel(ServerSocketChannel channel) {
        super(null, channel, SelectionKey.OP_ACCEPT);
        config = new NioServerSocketChannelConfig(this, javaChannel().socket());
    }
}
 


public abstract class AbstractNioChannel extends AbstractChannel {

    private final SelectableChannel ch;
    protected final int readInterestOp;
    volatile SelectionKey selectionKey;

    protected AbstractNioChannel(Channel parent, SelectableChannel ch, int readInterestOp) {
        super(parent);
        this.ch = ch;
        this.readInterestOp = readInterestOp;
        try {
            // 如果一個Channel要注冊到Selector中, 那么這個Channel必須是非阻塞的
            // 即channel.configureBlocking(false);
            ch.configureBlocking(false);
        } catch (IOException e) {
        }
    }

    protected SelectableChannel javaChannel() {
        return ch;
    }
}
  • NioServerSocketChannel核心關(guān)注SelectableChannel和readInterestOp。
  • DEFAULT_SELECTOR_PROVIDER是通過SelectorProvider.provider()返回單例對象。
  • SelectableChannel對象由provider.openServerSocketChannel()。
  • readInterestOp在NioServerSocketChannel中監(jiān)聽事件為SelectionKey.OP_ACCEPT。
  • 核心關(guān)注ServerSocketChannel，繼續(xù)關(guān)注分析。
  • 核心關(guān)注SelectorProvider，用于創(chuàng)建ServerSocketChannel對象，繼續(xù)關(guān)注分析。


ServerSocketChannel

ServerSocketChannel
  • ServerSocketChannel的類關(guān)系圖如上圖所示，突出父類SelectableChannel，解答AbstractNioChannel的參數(shù)對象SelectableChannel的緣由。


SelectorProvider

SelectorProvider
  • SelectorProvider的類圖如上所示，生成ServerSocketChannel對象，分析下原理。


public abstract class SelectorProvider {

    private static final Object lock = new Object();
    private static SelectorProvider provider = null;

    public static SelectorProvider provider() {
        synchronized (lock) {
            if (provider != null)
                return provider;
            return AccessController.doPrivileged(
                new PrivilegedAction<SelectorProvider>() {
                    public SelectorProvider run() {
                            if (loadProviderFromProperty())
                                return provider;
                            if (loadProviderAsService())
                                return provider;
                            provider = sun.nio.ch.DefaultSelectorProvider.create();
                            return provider;
                        }
                    });
        }
    }
}


public abstract class SelectorProviderImpl extends SelectorProvider {

    public abstract AbstractSelector openSelector() throws IOException;

    public ServerSocketChannel openServerSocketChannel() throws IOException {
        return new ServerSocketChannelImpl(this);
    }

    public SocketChannel openSocketChannel() throws IOException {
        return new SocketChannelImpl(this);
    }
}
  • SelectorProvider#provider用于生成單例的SelectorProvider對象。
  • SelectorProviderImpl#openServerSocketChannel負責創(chuàng)建ServerSocketChannel對象。
  • SelectorProviderImpl#openSocketChannel負責創(chuàng)建SocketChannel對象。
  • 在Netty的體系里面channel對象都是由SelectorProvider來完成的，相當于靈魂。


register分析

注冊調(diào)用鏈
  • channel注冊調(diào)用鏈如上圖，按照NioEventLoopGroup => NioEventLoop => NioSocketChannel進行注冊。
  • 最終的注冊是將NioSocketChannel注冊到NioEventLoop的selector對象當中。


NioEventLoopGroup注冊

public abstract class AbstractBootstrap<B extends AbstractBootstrap<B, C>, C extends Channel> implements Cloneable {

    final ChannelFuture initAndRegister() {
        Channel channel = null;
        try {
            channel = channelFactory.newChannel();
            init(channel);
        } catch (Throwable t) {
        }
        // 進入NioEventLoopGroup的register過程，進入NioEventLoop的過程
        ChannelFuture regFuture = config().group().register(channel);
        if (regFuture.cause() != null) {
            if (channel.isRegistered()) {
                channel.close();
            } else {
                channel.unsafe().closeForcibly();
            }
        }

        return regFuture;
    }
}

public abstract class MultithreadEventLoopGroup extends MultithreadEventExecutorGroup implements EventLoopGroup {

    public ChannelFuture register(Channel channel) {
        return next().register(channel);
    }
}
  • config().group().register(channel)的方法中從NioEventLoopGroup執(zhí)行register方法。
  • next().register(channel)的方法實現(xiàn)從NioEventLoopGroup到NioEventLoop的調(diào)用傳遞。


NioEventLoop注冊

public abstract class SingleThreadEventLoop extends SingleThreadEventExecutor implements EventLoop {

    public ChannelFuture register(Channel channel) {
        // DefaultChannelPromise(channel, this)的this為NioEventLoop對象
        return register(new DefaultChannelPromise(channel, this));
    }

    public ChannelFuture register(final ChannelPromise promise) {

        ObjectUtil.checkNotNull(promise, "promise");
        // 由NioEventLoop進入NioSocketChannel的register過程
        promise.channel().unsafe().register(this, promise);
        return promise;
    }
}
  • promise.channel().unsafe().register(this, promise)實現(xiàn)由NioEventLoop到NioSocketChannel的調(diào)用傳遞。


NioSocketChannel注冊

public abstract class AbstractNioChannel extends AbstractChannel {

    protected abstract class AbstractUnsafe implements Unsafe {
        public final void register(EventLoop eventLoop, final ChannelPromise promise) {
            AbstractChannel.this.eventLoop = eventLoop;

            if (eventLoop.inEventLoop()) {
                register0(promise);
            } else {
                try {
                    eventLoop.execute(new Runnable() {
                        @Override
                        public void run() {
                            register0(promise);
                        }
                    });
                } catch (Throwable t) {
                }
            }
        }

        private void register0(ChannelPromise promise) {
            try {
                // 省略代碼
                doRegister();
            } catch (Throwable t) {
            }
        }

    }

    protected void doRegister() throws Exception {
        boolean selected = false;
        for (;;) {
            try {
                // 綁定NioSocketChannel到NioEventLoop對象當中，并且附加NioSocketChannel對象。
                selectionKey = javaChannel().register(eventLoop().selector, 0, this);
                return;
            } catch (CancelledKeyException e) {
            }
        }
    }
}
  • AbstractUnsafe#register執(zhí)行的NioSocketChannel的注冊動作，最終會執(zhí)行AbstractNioChannel#doRegister。
  • AbstractNioChannel#doRegister內(nèi)部通過javaChannel().register(eventLoop().selector, 0, this)實現(xiàn)綁定Channel到NioEventLoop對象當中。
最后編輯于 ：
?著作權(quán)歸作者所有,轉(zhuǎn)載或內(nèi)容合作請聯(lián)系作者
【社區(qū)內(nèi)容提示】社區(qū)部分內(nèi)容疑似由AI輔助生成，瀏覽時請結(jié)合常識與多方信息審慎甄別。
平臺聲明：文章內(nèi)容（如有圖片或視頻亦包括在內(nèi)）由作者上傳并發(fā)布，文章內(nèi)容僅代表作者本人觀點，簡書系信息發(fā)布平臺，僅提供信息存儲服務(wù)。

相關(guān)閱讀更多精彩內(nèi)容

友情鏈接更多精彩內(nèi)容