netty服務(wù)端處理請(qǐng)求聯(lián)合pipeline分析

更新時(shí)間：2023年04月25日 11:18:11 作者：寬仔的編程之路

這篇文章主要為大家介紹了netty服務(wù)端處理請(qǐng)求聯(lián)合pipeline示例分析，有需要的朋友可以借鑒參考下，希望能夠有所幫助，祝大家多多進(jìn)步，早日升職加薪

兩個(gè)問(wèn)題

在客戶(hù)端接入的時(shí)候, NioMessageUnsafe的read方法中pipeline.fireChannelRead(readBuf.get(i))為什么會(huì)調(diào)用到ServerBootstrap的內(nèi)部類(lèi)ServerBootstrapAcceptor中的channelRead()方法。
客戶(hù)端handler是什么時(shí)候被添加的?

先分析第一個(gè)問(wèn)題?；氐絥etty處理客戶(hù)端請(qǐng)求分析_1中服務(wù)端接收到accpet事件后，進(jìn)行讀取的方法NioMessageUnsafe.read()

NioMessageUnsafe.read()

public void read() {
    //必須是NioEventLoop方法調(diào)用的, 不能通過(guò)外部線程調(diào)用
    assert eventLoop().inEventLoop();
    //服務(wù)端channel的config
    final ChannelConfig config = config();
    //服務(wù)端channel的pipeline
    final ChannelPipeline pipeline = pipeline();
    //處理服務(wù)端接入的速率
    final RecvByteBufAllocator.Handle allocHandle = unsafe().recvBufAllocHandle();
    //設(shè)置配置
    allocHandle.reset(config);
    boolean closed = false;
    Throwable exception = null;
    try {
        try {
            do {
                //創(chuàng)建jdk底層的channel
                //readBuf用于臨時(shí)承載讀到鏈接
                int localRead = doReadMessages(readBuf);
                if (localRead == 0) {
                    break;
                }
                if (localRead < 0) {
                    closed = true;
                    break;
                }
                //分配器將讀到的鏈接進(jìn)行計(jì)數(shù)
                allocHandle.incMessagesRead(localRead);
                //連接數(shù)是否超過(guò)最大值
            } while (allocHandle.continueReading());
        } catch (Throwable t) {
            exception = t;
        }
        int size = readBuf.size();
        //遍歷每一條客戶(hù)端連接
        for (int i = 0; i < size; i ++) {
            readPending = false;
            //傳遞事件, 將創(chuàng)建NioSokectChannel進(jìn)行傳遞
            //最終會(huì)調(diào)用ServerBootstrap的內(nèi)部類(lèi)ServerBootstrapAcceptor的channelRead()方法
            pipeline.fireChannelRead(readBuf.get(i));
        }
        readBuf.clear();
        allocHandle.readComplete();
        pipeline.fireChannelReadComplete();
        //代碼省略
    } finally {
        //代碼省略
    }
}

重點(diǎn)看pipeline.fireChannelRead(readBuf.get(i))

首先, 這里pipeline是服務(wù)端channel的pipeline, 也就是NioServerSocketChannel的pipeline

我們學(xué)習(xí)過(guò)pipeline之后, 對(duì)這種寫(xiě)法并不陌生, 就是傳遞channelRead事件, 這里通過(guò)傳遞channelRead事件走到了ServerBootstrapAcceptor的channelRead()方法, 說(shuō)明在這步之前, ServerBootstrapAcceptor作為一個(gè)handler添加到了服務(wù)端channel的pipeline中, 那么這個(gè)handler什么時(shí)候添加的呢?

我們回顧下第一章, 初始化NioServerSocketChannel的時(shí)候, 調(diào)用了ServerBootstrap的init方法回顧下ServerBootstrap.init的調(diào)用鏈路:

ServerBootstrap.bind(8899) ---> AbstractBootstrap.doBind(final SocketAddress localAddress) ---> AbstractBootstrap.initAndRegister() ---> ServerBootstrap.init(Channel channel)

ServerBootstrap.init(Channel channel)

void init(Channel channel) throws Exception {
    //獲取用戶(hù)定義的選項(xiàng)(1)
    final Map<ChannelOption<?>, Object> options = options0();
    synchronized (options) {
        channel.config().setOptions(options);
    }
    //獲取用戶(hù)定義的屬性(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());
        }
    }
    //獲取channel的pipline(3)
    ChannelPipeline p = channel.pipeline();
    //work線程組(4)
    final EventLoopGroup currentChildGroup = childGroup;
    //用戶(hù)設(shè)置的Handler(5)
    final ChannelHandler currentChildHandler = childHandler;
    final Entry<ChannelOption<?>, Object>[] currentChildOptions;
    final Entry<AttributeKey<?>, Object>[] currentChildAttrs;
    //選項(xiàng)轉(zhuǎn)化為Entry對(duì)象(6)
    synchronized (childOptions) { 
        currentChildOptions = childOptions.entrySet().toArray(newOptionArray(childOptions.size()));
    }
    //屬性轉(zhuǎn)化為Entry對(duì)象(7)
    synchronized (childAttrs) { 
        currentChildAttrs = childAttrs.entrySet().toArray(newAttrArray(childAttrs.size()));
    }
    //添加服務(wù)端handler(8)
    p.addLast(new ChannelInitializer<Channel>() {
        //初始化channel
        @Override
        public void initChannel(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(
                            currentChildGroup, currentChildHandler, currentChildOptions, currentChildAttrs));
                }
            });
        }
    });
}

我們重點(diǎn)關(guān)注第8步, 添加服務(wù)端channel, 這里的pipeline, 是服務(wù)服務(wù)端channel的pipeline, 也就是NioServerSocketChannel綁定的pipeline, 這里添加了一個(gè)ChannelInitializer類(lèi)型的handler

ChannelInitializer的繼承關(guān)系

public abstract class ChannelInitializer&lt;C extends Channel&gt; extends ChannelInboundHandlerAdapter {
    //省略類(lèi)體
}

我們看到其繼承了ChannelInboundHandlerAdapter, 說(shuō)明是一個(gè)inbound類(lèi)型的handler

這里我們可能會(huì)想到, 添加完handler會(huì)執(zhí)行handlerAdded, 然后在handlerAdded方法中做了添加ServerBootstrapAcceptor這個(gè)handler

但是, 實(shí)際上并不是這樣的, 當(dāng)程序執(zhí)行到這里, 并沒(méi)有馬上執(zhí)行handlerAdded, 我們緊跟addLast方法

最后執(zhí)行到DefualtChannelPipeline.addLast(EventExecutorGroup group, String name, ChannelHandler handler)

public final ChannelPipeline addLast(EventExecutorGroup group, String name, ChannelHandler handler) {
    final AbstractChannelHandlerContext newCtx;
    synchronized (this) {
        //判斷handler是否被重復(fù)添加(1)
        checkMultiplicity(handler);
        //創(chuàng)建一個(gè)HandlerContext并添加到列表(2)
        newCtx = newContext(group, filterName(name, handler), handler);
        //添加HandlerContext(3)
        addLast0(newCtx);
        //是否已注冊(cè)
        if (!registered) {
            newCtx.setAddPending();
            callHandlerCallbackLater(newCtx, true);
            return this;
        }
        EventExecutor executor = newCtx.executor();
        if (!executor.inEventLoop()) {
            newCtx.setAddPending();
            //回調(diào)用戶(hù)事件
            executor.execute(new Runnable() {
                @Override
                public void run() {
                    callHandlerAdded0(newCtx);
                }
            });
            return this;
        }
    }
    //回調(diào)添加事件(4)
    callHandlerAdded0(newCtx);
    return this;
}

首先完成了handler的添加, 但是并沒(méi)有馬上執(zhí)行回調(diào)

這里我們重點(diǎn)關(guān)注if (!registered)這個(gè)條件判斷, 其實(shí)在注冊(cè)完成, registered會(huì)變成true, 但是走到這一步的時(shí)候NioServerSockeChannel并沒(méi)有完成注冊(cè)(可以回顧第一章看注冊(cè)在哪一步), 所以會(huì)進(jìn)到if里并返回自身

DefualtChannelPipeline.callHandlerCallbackLater(AbstractChannelHandlerContext ctx, boolean added)

private void callHandlerCallbackLater(AbstractChannelHandlerContext ctx, boolean added) {
    assert !registered;
    //判斷是否已添加, 未添加, 進(jìn)行添加, 已添加進(jìn)行刪除
    PendingHandlerCallback task = added ? new PendingHandlerAddedTask(ctx) : new PendingHandlerRemovedTask(ctx);
    //獲取第一個(gè)Callback任務(wù)
    PendingHandlerCallback pending = pendingHandlerCallbackHead;
    //如果第一個(gè)Callback任務(wù)為空
    if (pending == null) {
        //將第一個(gè)任務(wù)設(shè)置為剛創(chuàng)建的任務(wù)
        pendingHandlerCallbackHead = task;
    } else {
        while (pending.next != null) {
            pending = pending.next;
        }
        pending.next = task;
    }
}

因我們調(diào)用這個(gè)方法的時(shí)候added傳的true, 所以PendingHandlerCallback task賦值為new PendingHandlerAddedTask(ctx)

PendingHandlerAddedTask這個(gè)類(lèi), 我們從名字可以看出, 這是一個(gè)handler添加的延遲任務(wù), 用于執(zhí)行handler延遲添加的操作, 同樣也對(duì)應(yīng)一個(gè)名字為PendingHandlerRemovedTask的類(lèi), 用于執(zhí)行延遲刪除handler的操作, 這兩個(gè)類(lèi)都繼承抽象類(lèi)PendingHandlerCallback

PendingHandlerAddedTask構(gòu)造方法

PendingHandlerAddedTask(AbstractChannelHandlerContext ctx) {
    super(ctx);
}

進(jìn)入super(ctx)

PendingHandlerCallback構(gòu)造方法

PendingHandlerCallback(AbstractChannelHandlerContext ctx) {
    this.ctx = ctx;
}

在父類(lèi)中, 保存了要添加的context, 也就是ChannelInitializer類(lèi)型的包裝類(lèi)

回到callHandlerCallbackLater方法

PendingHandlerCallback pending = pendingHandlerCallbackHead;

這表示獲取第一個(gè)PendingHandlerCallback的任務(wù), 其實(shí)PendingHandlerCallback是一個(gè)單向鏈表, 自身維護(hù)一個(gè)PendingHandlerCallback類(lèi)型的next, 指向下一個(gè)任務(wù), 在DefaultChannelPipeline這個(gè)類(lèi)中, 定義了個(gè)PendingHandlerCallback類(lèi)型的引用pendingHandlerCallbackHead, 用來(lái)指向延遲回調(diào)任務(wù)的中的第一個(gè)任務(wù)。

之后判斷這個(gè)任務(wù)是為空, 如果是第一次添加handler, 那么這里就是空, 所以將第一個(gè)任務(wù)賦值為我們剛創(chuàng)建的添加任務(wù)。

如果不是第一次添加handler, 則將我們新創(chuàng)建的任務(wù)添加到鏈表的尾部, 因?yàn)檫@里我們是第一次添加, 所以第一個(gè)回調(diào)任務(wù)就指向了我們創(chuàng)建的添加handler的任務(wù)。

完成這一系列操作之后, addLast方法返歸, 此時(shí)并沒(méi)有完成添加操作。

而什么時(shí)候完成添加操作的呢?

回到在服務(wù)端channel注冊(cè)時(shí)候的會(huì)走到AbstractChannel.register0方法回顧下AbstractChannel.register0的調(diào)用鏈路:

ServerBootstrap.bind(8899) ---> AbstractBootstrap.doBind(final SocketAddress localAddress) ---> AbstractBootstrap.initAndRegister() ---> config().group().register(channel) ---> SingleThreadEventLoop.register(final ChannelPromise promise) ---> AbstractChannel.register(EventLoop eventLoop, final ChannelPromise promise) ---> AbstractChannel.register0(ChannelPromise promise)

AbstractChannel.register0(ChannelPromise promise)

private void register0(ChannelPromise promise) {
    try {
        //做實(shí)際的注冊(cè)(1)
        doRegister();
        neverRegistered = false;
        registered = true;
        //觸發(fā)事件(2)
        pipeline.invokeHandlerAddedIfNeeded();
        safeSetSuccess(promise);
        //觸發(fā)注冊(cè)成功事件(3)
        pipeline.fireChannelRegistered();
        if (isActive()) {
            if (firstRegistration) {
                //傳播active事件(4)
                pipeline.fireChannelActive();
            } else if (config().isAutoRead()) {
                beginRead();
            }
        }
    } catch (Throwable t) {
        //省略代碼
    }
}

重點(diǎn)關(guān)注第二步pipeline.invokeHandlerAddedIfNeeded(), 這里已經(jīng)通過(guò)doRegister()方法完成了實(shí)際的注冊(cè), 我們跟到該方法中

pipeline.invokeHandlerAddedIfNeeded()

final void invokeHandlerAddedIfNeeded() {
    assert channel.eventLoop().inEventLoop();
    if (firstRegistration) {
        firstRegistration = false;
        callHandlerAddedForAllHandlers();
    }
}

這里會(huì)判斷是否第一次注冊(cè), 這里返回true, 然后會(huì)執(zhí)行callHandlerAddedForAllHandlers()方法, 我們跟進(jìn)去

DefaultChannelPipeline.callHandlerAddedForAllHandlers

private void callHandlerAddedForAllHandlers() {
    final PendingHandlerCallback pendingHandlerCallbackHead;
    synchronized (this) {
        assert !registered;
        registered = true;
        pendingHandlerCallbackHead = this.pendingHandlerCallbackHead;
        this.pendingHandlerCallbackHead = null;
    }
    //獲取task
    PendingHandlerCallback task = pendingHandlerCallbackHead;
    while (task != null) {
        //執(zhí)行添加handler方法
        task.execute();
        task = task.next;
    }
}

這里拿到第一個(gè)延遲執(zhí)行handler添加的task其實(shí)就是我們之前剖析過(guò)的, 延遲執(zhí)行handler添加的task, 就是PendingHandlerAddedTask對(duì)象

在while循環(huán)中, 通過(guò)執(zhí)行execute()方法將handler添加

進(jìn)入PendingHandlerAddedTask.execute()

void execute() {
    //獲取當(dāng)前eventLoop線程
    EventExecutor executor = ctx.executor();
    //是當(dāng)前執(zhí)行的線程
    if (executor.inEventLoop()) {
        callHandlerAdded0(ctx);
    } else {
        try {
            //添加到隊(duì)列
            executor.execute(this);
        } catch (RejectedExecutionException e) {
            //代碼省略
        }
    }
}

再進(jìn)入callHandlerAdded0方法

callHandlerAdded0(final AbstractChannelHandlerContext ctx)

private void callHandlerAdded0(final AbstractChannelHandlerContext ctx) {
    try {
        ctx.handler().handlerAdded(ctx);
        ctx.setAddComplete();
    } catch (Throwable t) {
        //省略...
    }
}

終于在這里, 我們看到了執(zhí)行回調(diào)的方法

再回到ServerBootstrap.init(Channel channel)

void init(Channel channel) throws Exception {
    //獲取用戶(hù)定義的選項(xiàng)(1)
    final Map&lt;ChannelOption&lt;?&gt;, Object&gt; options = options0();
    synchronized (options) {
        channel.config().setOptions(options);
    }
    //獲取用戶(hù)定義的屬性(2)
    final Map&lt;AttributeKey&lt;?&gt;, Object&gt; attrs = attrs0();
    synchronized (attrs) {
        for (Entry&lt;AttributeKey&lt;?&gt;, Object&gt; e: attrs.entrySet()) {
            @SuppressWarnings("unchecked")
            AttributeKey&lt;Object&gt; key = (AttributeKey&lt;Object&gt;) e.getKey();
            channel.attr(key).set(e.getValue());
        }
    }
    //獲取channel的pipline(3)
    ChannelPipeline p = channel.pipeline();
    //work線程組(4)
    final EventLoopGroup currentChildGroup = childGroup;
    //用戶(hù)設(shè)置的Handler(5)
    final ChannelHandler currentChildHandler = childHandler;
    final Entry&lt;ChannelOption&lt;?&gt;, Object&gt;[] currentChildOptions;
    final Entry&lt;AttributeKey&lt;?&gt;, Object&gt;[] currentChildAttrs;
    //選項(xiàng)轉(zhuǎn)化為Entry對(duì)象(6)
    synchronized (childOptions) { 
        currentChildOptions = childOptions.entrySet().toArray(newOptionArray(childOptions.size()));
    }
    //屬性轉(zhuǎn)化為Entry對(duì)象(7)
    synchronized (childAttrs) { 
        currentChildAttrs = childAttrs.entrySet().toArray(newAttrArray(childAttrs.size()));
    }
    //添加服務(wù)端handler(8)
    p.addLast(new ChannelInitializer&lt;Channel&gt;() {
        //初始化channel
        @Override
        public void initChannel(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(
                            currentChildGroup, currentChildHandler, currentChildOptions, currentChildAttrs));
                }
            });
        }
    });
}

我們繼續(xù)看第8步添加服務(wù)端handler

因?yàn)檫@里的handler是ChannelInitializer, 所以完成添加之后會(huì)調(diào)用ChannelInitializer的handlerAdded方法

跟到handlerAdded方法

ChannelInitializer.handlerAdded(ChannelHandlerContext ctx)

public void handlerAdded(ChannelHandlerContext ctx) throws Exception {
    //默認(rèn)情況下, 會(huì)返回true
    if (ctx.channel().isRegistered()) {
        initChannel(ctx);
    }
}

因?yàn)閳?zhí)行到這步服務(wù)端channel已經(jīng)完成注冊(cè), 所以會(huì)執(zhí)行到initChannel方法

ChannelInitializer.initChannel(ChannelHandlerContext ctx)

private boolean initChannel(ChannelHandlerContext ctx) throws Exception {
    //這段代碼是否被執(zhí)行過(guò)
    if (initMap.putIfAbsent(ctx, Boolean.TRUE) == null) {
        try {
            initChannel((C) ctx.channel());
        } catch (Throwable cause) {
            exceptionCaught(ctx, cause);
        } finally {
            //調(diào)用之后會(huì)刪除當(dāng)前節(jié)點(diǎn)
            remove(ctx);
        }
        return true;
    }
    return false;
}

我們關(guān)注initChannel這個(gè)方法, 這個(gè)方法是在ChannelInitializer的匿名內(nèi)部來(lái)實(shí)現(xiàn)的, 這里我們注意, 在initChannel方法執(zhí)行完畢之后會(huì)調(diào)用remove(ctx)刪除當(dāng)前節(jié)點(diǎn)

繼續(xù)跟進(jìn)initChannel方法

public void initChannel(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(
                    currentChildGroup, currentChildHandler, currentChildOptions, currentChildAttrs));
        }
    });
}

這里首先添加用戶(hù)自定義的handler, 這里如果用戶(hù)沒(méi)有定義, 則添加不成功, 然后, 會(huì)調(diào)用addLast將ServerBootstrapAcceptor這個(gè)handler添加了進(jìn)去, 同樣這個(gè)handler也繼承了ChannelInboundHandlerAdapter, 在這個(gè)handler中, 重寫(xiě)了channelRead方法, 所以, 這就是第一個(gè)問(wèn)題的答案

緊接著我們看第二個(gè)問(wèn)題:客戶(hù)端handler是什么時(shí)候被添加的?

看ServerBootstrapAcceptor的channelRead方法

public void channelRead(ChannelHandlerContext ctx, Object msg) {
    final Channel child = (Channel) msg;
    //添加channelHadler, 這個(gè)channelHandler, 就是用戶(hù)代碼添加的ChannelInitializer
    child.pipeline().addLast(childHandler);
    //代碼省略
    try {
        //work線程注冊(cè)channel
        childGroup.register(child).addListener(new ChannelFutureListener() {
            //代碼省略
        });
    } catch (Throwable t) {
        forceClose(child, t);
    }
}

這里真相可以大白了, 服務(wù)端再創(chuàng)建完客戶(hù)端channel之后, 將新創(chuàng)建的NioSocketChannel作為參數(shù)觸發(fā)channelRead事件(可以回顧NioMessageUnsafe.read方法, 代碼這里就不貼了), 所以這里的參數(shù)msg就是NioSocketChannel

拿到channel時(shí)候再將客戶(hù)端的handler添加進(jìn)去, 我們回顧客戶(hù)端handler的添加過(guò)程:

.childHandler(new ChannelInitializer&lt;SocketChannel&gt;() {
    @Override
    public void initChannel(SocketChannel ch) {
        ch.pipeline().addLast(new StringDecoder());
        ch.pipeline().addLast(new StringEncoder());
        ch.pipeline().addLast(new ServerHandler());
    }
});

和服務(wù)端channel的邏輯一樣, 首先會(huì)添加ChannelInitializer這個(gè)handler但是沒(méi)有注冊(cè)所以沒(méi)有執(zhí)行添加handler的回調(diào), 將任務(wù)保存到一個(gè)延遲回調(diào)的task中

等客戶(hù)端channel注冊(cè)完畢, 會(huì)將執(zhí)行添加handler的回調(diào), 也就是handlerAdded方法, 在回調(diào)中執(zhí)行initChannel方法將客戶(hù)端handler添加進(jìn)去, 然后刪除ChannelInitializer這個(gè)handler

因?yàn)樵诜?wù)端channel中這塊邏輯已經(jīng)進(jìn)行了詳細(xì)的剖析, 所以這邊就不在贅述, 同學(xué)們可以自己跟進(jìn)去走一遍流程

這里注意, 因?yàn)槊縿?chuàng)建一個(gè)NioSoeketChannel都會(huì)調(diào)用服務(wù)端ServerBootstrapAcceptor的channelRead方法, 所以這里會(huì)將每一個(gè)NioSocketChannel的handler進(jìn)行添加

總結(jié)

本文章分析了事件傳輸?shù)南嚓P(guān)邏輯, 包括handler的添加, 刪除, inbound和outbound以及異常事件的傳輸, 最后結(jié)合第一章和第三章, 剖析了服務(wù)端channel和客戶(hù)端channel的添加過(guò)程，更多關(guān)于netty服務(wù)端請(qǐng)求聯(lián)合pipeline的資料請(qǐng)關(guān)注腳本之家其它相關(guān)文章！

您可能感興趣的文章: