溫馨提示×

您好,登錄后才能下訂單哦!

密碼登錄×
登錄注冊(cè)×
其他方式登錄
點(diǎn)擊 登錄注冊(cè) 即表示同意《億速云用戶(hù)服務(wù)條款》

netty 5 alph1源碼分析(服務(wù)端創(chuàng)建過(guò)程)

發(fā)布時(shí)間:2020-07-10 18:38:17 來(lái)源:網(wǎng)絡(luò) 閱讀:2569 作者:randy_shandong 欄目:移動(dòng)開(kāi)發(fā)

參照《Netty系列之Netty 服務(wù)端創(chuàng)建》,研究了netty的服務(wù)端創(chuàng)建過(guò)程。至于netty的優(yōu)勢(shì),可以參照網(wǎng)絡(luò)其他文章?!禢etty系列之Netty 服務(wù)端創(chuàng)建》是 李林鋒撰寫(xiě)的netty源碼分析的一篇好文,絕對(duì)是技術(shù)干貨。但拋開(kāi)技術(shù)來(lái)說(shuō),也存在一些瑕疵。

缺點(diǎn)如下

  1. 代碼銜接不連貫,上下不連貫。

  2. 代碼片段是截圖,對(duì)閱讀代理不便(可能和閱讀習(xí)慣有關(guān))

本篇主要內(nèi)容,參照《Netty系列之Netty 服務(wù)端創(chuàng)建》,梳理出自己喜歡的閱讀風(fēng)格。

1.整體邏輯圖

整體將服務(wù)端創(chuàng)建分為2部分:(1)綁定端口,提供服務(wù)過(guò)程;(2)輪詢(xún)網(wǎng)絡(luò)請(qǐng)求

netty 5 alph1源碼分析(服務(wù)端創(chuàng)建過(guò)程)


1.1 綁定端口序列圖

netty 5 alph1源碼分析(服務(wù)端創(chuàng)建過(guò)程)

1.2 類(lèi)圖

netty 5 alph1源碼分析(服務(wù)端創(chuàng)建過(guò)程)

類(lèi)圖僅僅涵蓋了綁定過(guò)程中比較重要的幾個(gè)組件

1.3 代碼分析

step 2 doBind 綁定本地端口,啟動(dòng)服務(wù)

private ChannelFuture doBind(final SocketAddress localAddress) {
    final ChannelFuture regFuture = initAndRegister();//1
    final Channel channel = regFuture.channel();
    if (regFuture.cause() != null) {
        return regFuture;
    }

    final ChannelPromise promise;
    if (regFuture.isDone()) {
        promise = channel.newPromise();
        doBind0(regFuture, channel, localAddress, promise);//2
    } else {
        // Registration future is almost always fulfilled already, but just in case it's not.
        promise = new DefaultChannelPromise(channel, GlobalEventExecutor.INSTANCE);
        regFuture.addListener(new ChannelFutureListener() {
            @Override
            public void operationComplete(ChannelFuture future) throws Exception {
                doBind0(regFuture, channel, localAddress, promise);//2
            }
        });
    }

    return promise;
}
主要分為2個(gè)處理單元

step3 initAndRegister

final ChannelFuture initAndRegister() {
    Channel channel;
    try {
        channel = createChannel();
    } catch (Throwable t) {
        return VoidChannel.INSTANCE.newFailedFuture(t);
    }

    try {
        init(channel);
    } catch (Throwable t) {
        channel.unsafe().closeForcibly();
        return channel.newFailedFuture(t);
    }
//注冊(cè)NioServerSocketChannel到Reactor線程的多路復(fù)用器上
    ChannelPromise regFuture = channel.newPromise();
    channel.unsafe().register(regFuture);
    if (regFuture.cause() != null) {
        if (channel.isRegistered()) {
            channel.close();
        } else {
            channel.unsafe().closeForcibly();
        }
    }

    return regFuture;
}

createChannel由子類(lèi)ServerBootstrap實(shí)現(xiàn),創(chuàng)建新的NioServerSocketChannel,并完成Channel初始化,以及注冊(cè)。

4.ServerBootstrap.createChannel

Channel createChannel() {
    EventLoop eventLoop = group().next();
    return channelFactory().newChannel(eventLoop, childGroup);
}

它有兩個(gè)參數(shù),參數(shù)1是從父類(lèi)的NIO線程池中順序獲取一個(gè)NioEventLoop,它就是服務(wù)端用于監(jiān)聽(tīng)和接收客戶(hù)端連接的Reactor線程。第二個(gè)參數(shù)就是所謂的workerGroup線程池,它就是處理IO讀寫(xiě)的Reactor線程組


5.ServerBootstrap.init

void init(Channel channel) throws Exception {
//設(shè)置Socket參數(shù)和NioServerSocketChannel的附加屬性
    final Map<ChannelOption<?>, Object> options = options();
    synchronized (options) {
        channel.config().setOptions(options);
    }
    final Map<AttributeKey<?>, Object> attrs = attrs();
    synchronized (attrs) {
        for (Entry<AttributeKey<?>, Object> e: attrs.entrySet()) {
            @SuppressWarnings("unchecked")
            AttributeKey<Object> key = (AttributeKey<Object>) e.getKey();
            channel.attr(key).set(e.getValue());
        }
    }
//將AbstractBootstrap的Handler添加到NioServerSocketChannel的ChannelPipeline中
    ChannelPipeline p = channel.pipeline();
    if (handler() != null) {
        p.addLast(handler());
    }

    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()));
    }
//將用于服務(wù)端注冊(cè)的Handler ServerBootstrapAcceptor添加到ChannelPipeline中
    p.addLast(new ChannelInitializer<Channel>() {
        @Override
        public void initChannel(Channel ch) throws Exception {
            ch.pipeline().addLast(new ServerBootstrapAcceptor(currentChildHandler, currentChildOptions,
                    currentChildAttrs));
        }
    });
}

到此處,Netty服務(wù)端監(jiān)聽(tīng)的相關(guān)資源已經(jīng)初始化完畢

6.AbstractChannel.AbstractUnsafe.register

public final void register(final ChannelPromise promise) {
//首先判斷是否是NioEventLoop自身發(fā)起的操作,如果是,則不存在并發(fā)操作,直接執(zhí)行Channel注冊(cè);
    if (eventLoop.inEventLoop()) {
        register0(promise);
    } else {//如果由其它線程發(fā)起,則封裝成一個(gè)Task放入消息隊(duì)列中異步執(zhí)行。
        try {
            eventLoop.execute(new Runnable() {
                @Override
                public void run() {
                    register0(promise);
                }
            });
        } catch (Throwable t) {
            logger.warn(
                    "Force-closing a channel whose registration task was not accepted by an event loop: {}",
                    AbstractChannel.this, t);
            closeForcibly();
            closeFuture.setClosed();
            promise.setFailure(t);
        }
    }
}

7.register0

private void register0(ChannelPromise promise) {
    try {
        // check if the channel is still open as it could be closed in the mean time when the register
        // call was outside of the eventLoop
        if (!ensureOpen(promise)) {
            return;
        }
        doRegister();
        registered = true;
        promise.setSuccess();
        pipeline.fireChannelRegistered();
        
        if (isActive()) {//完成綁定時(shí),不會(huì)調(diào)用該代碼段
            pipeline.fireChannelActive();
        }
    } catch (Throwable t) {
        // Close the channel directly to avoid FD leak.
        closeForcibly();
        closeFuture.setClosed();
        if (!promise.tryFailure(t)) {
            logger.warn(
                    "Tried to fail the registration promise, but it is complete already. " +
                            "Swallowing the cause of the registration failure:", t);
        }
    }
}

觸發(fā)事件

8.doRegister

protected void doRegister() throws Exception {
    boolean selected = false;
    for (;;) {
        try {
        //將NioServerSocketChannel注冊(cè)到NioEventLoop的Selector上
            selectionKey = javaChannel().register(eventLoop().selector, 0, this);
            return;
        } catch (CancelledKeyException e) {
            if (!selected) {
                // Force the Selector to select now as the "canceled" SelectionKey may still be
                // cached and not removed because no Select.select(..) operation was called yet.
                eventLoop().selectNow();
                selected = true;
            } else {
                // We forced a select operation on the selector before but the SelectionKey is still cached
                // for whatever reason. JDK bug ?
                throw e;
            }
        }
    }
}

大伙兒可能會(huì)很詫異,應(yīng)該注冊(cè)O(shè)P_ACCEPT(16)到多路復(fù)用器上,怎么注冊(cè)0呢?0表示只注冊(cè),不監(jiān)聽(tīng)任何網(wǎng)絡(luò)操作。這樣做的原因如下:

  1. 注冊(cè)方法是多態(tài)的,它既可以被NioServerSocketChannel用來(lái)監(jiān)聽(tīng)客戶(hù)端的連接接入,也可以用來(lái)注冊(cè)SocketChannel,用來(lái)監(jiān)聽(tīng)網(wǎng)絡(luò)讀或者寫(xiě)操作;

  2. 通過(guò)SelectionKey的interestOps(int ops)方法可以方便的修改監(jiān)聽(tīng)操作位。所以,此處注冊(cè)需要獲取SelectionKey并給AbstractNioChannel的成員變量selectionKey賦值。



向AI問(wèn)一下細(xì)節(jié)

免責(zé)聲明:本站發(fā)布的內(nèi)容(圖片、視頻和文字)以原創(chuàng)、轉(zhuǎn)載和分享為主,文章觀點(diǎn)不代表本網(wǎng)站立場(chǎng),如果涉及侵權(quán)請(qǐng)聯(lián)系站長(zhǎng)郵箱:is@yisu.com進(jìn)行舉報(bào),并提供相關(guān)證據(jù),一經(jīng)查實(shí),將立刻刪除涉嫌侵權(quán)內(nèi)容。

AI