Netty源碼分析 （八）----- write過程 源碼分析

上一篇文章主要講了netty的read過程，本文主要分析一下write和writeAndFlush。

主要內容

本文分以下幾個部分闡述一個java對象最后是如何轉變成字節流，寫到socket緩沖區中去的

1. pipeline中的標准鏈表結構
2. java對象編碼過程
3. write：寫隊列
4. flush：刷新寫隊列
5. writeAndFlush: 寫隊列並刷新

pipeline中的標准鏈表結構

一個標准的pipeline鏈式結構如下

數據從head節點流入，先拆包，然后解碼成業務對象，最后經過業務Handler處理，調用write，將結果對象寫出去。而寫的過程先通過tail節點，然后通過encoder節點將對象編碼成ByteBuf，最后將該ByteBuf對象傳遞到head節點，調用底層的Unsafe寫到jdk底層管道

java對象編碼過程

為什么我們在pipeline中添加了encoder節點，java對象就轉換成netty可以處理的ByteBuf，寫到管道里？

我們先看下調用write的code

BusinessHandler

protected void channelRead0(ChannelHandlerContext ctx, Request request) throws Exception {
    Response response = doBusiness(request);

    if (response != null) {
        ctx.channel().write(response);
    }
}

業務處理器接受到請求之后，做一些業務處理，返回一個Response，然后，response在pipeline中傳遞，落到 Encoder節點，我們來跟蹤一下 ctx.channel().write(response);

public ChannelFuture write(Object msg) {
    return this.pipeline.write(msg);
}

調用了Channel中的pipeline中的write方法，我們接着看

public final ChannelFuture write(Object msg) {
    return this.tail.write(msg);
}

pipeline中有屬性tail，調用tail中的write，由此我們知道write消息的時候，從tail開始，接着往下看

private void write(Object msg, boolean flush, ChannelPromise promise) {
    AbstractChannelHandlerContext next = this.findContextOutbound();
    Object m = this.pipeline.touch(msg, next);
    EventExecutor executor = next.executor();
    if (executor.inEventLoop()) {
        if (flush) {
            next.invokeWriteAndFlush(m, promise);
        } else {
           next.invokeWrite(m, promise);
        }
    } else {
        Object task;
        if (flush) {
            task = AbstractChannelHandlerContext.WriteAndFlushTask.newInstance(next, m, promise);
        } else {
            task = AbstractChannelHandlerContext.WriteTask.newInstance(next, m, promise);
        }

        safeExecute(executor, (Runnable)task, promise, m);
    }

}

中間我省略了幾個重載的方法，我們來看看第一行代碼，next = this.findContextOutbound();

private AbstractChannelHandlerContext findContextOutbound() {
    AbstractChannelHandlerContext ctx = this;

    do {
        ctx = ctx.prev;
    } while(!ctx.outbound);

    return ctx;
}

通過 ctx = ctx.prev; 我們知道從tail開始找到pipeline中的第一個outbound的handler，然后調用 invokeWrite(m, promise)，此時找到的第一個outbound的handler就是我們自定義的編碼器Encoder

我們接着看 next.invokeWrite(m, promise);

private void invokeWrite(Object msg, ChannelPromise promise) {
    if (this.invokeHandler()) {
        this.invokeWrite0(msg, promise);
    } else {
        this.write(msg, promise);
    }

}
private void invokeWrite0(Object msg, ChannelPromise promise) {
    try {
        ((ChannelOutboundHandler)this.handler()).write(this, msg, promise);
    } catch (Throwable var4) {
        notifyOutboundHandlerException(var4, promise);
    }

}

一路代碼跟下來，我們可以知道是調用了第一個outBound類型的handler中的write方法，也就是第一個調用的是我們自定義編碼器Encoder的write方法

我們來看看自定義Encoder

public class Encoder extends MessageToByteEncoder<Response> {
    @Override
    protected void encode(ChannelHandlerContext ctx, Response response, ByteBuf out) throws Exception {
        out.writeByte(response.getVersion());
        out.writeInt(4 + response.getData().length);
        out.writeBytes(response.getData());
    }
}

自定義Encoder繼承 MessageToByteEncoder ，並且重寫了 encode方法，這就是編碼器的核心，我們先來看 MessageToByteEncoder

public abstract class MessageToByteEncoder<I> extends ChannelOutboundHandlerAdapter {

我們看到 MessageToByteEncoder 繼承了 ChannelOutboundHandlerAdapter，說明了 Encoder 是一個 Outbound的handler

我們來看看 Encoder 的父類 MessageToByteEncoder中的write方法

MessageToByteEncoder

@Override
public void write(ChannelHandlerContext ctx, Object msg, ChannelPromise promise) throws Exception {
    ByteBuf buf = null;
    try {
        // 判斷當前Handelr是否能處理寫入的消息
        if (acceptOutboundMessage(msg)) {
            @SuppressWarnings("unchecked")
            // 強制換換
            I cast = (I) msg;
            // 分配一段ButeBuf
            buf = allocateBuffer(ctx, cast, preferDirect); try {
            // 調用encode，這里就調回到  `Encoder` 這個Handelr中    
 encode(ctx, cast, buf);
            } finally {
                // 既然自定義java對象轉換成ByteBuf了，那么這個對象就已經無用了，釋放掉
                // (當傳入的msg類型是ByteBuf的時候，就不需要自己手動釋放了)
                ReferenceCountUtil.release(cast);
            }
            // 如果buf中寫入了數據，就把buf傳到下一個節點
            if (buf.isReadable()) { ctx.write(buf, promise); } else { // 否則，釋放buf，將空數據傳到下一個節點 
 buf.release(); ctx.write(Unpooled.EMPTY_BUFFER, promise); }
            buf = null;
        } else {
            // 如果當前節點不能處理傳入的對象，直接扔給下一個節點處理
            ctx.write(msg, promise);
        }
    } catch (EncoderException e) {
        throw e;
    } catch (Throwable e) {
        throw new EncoderException(e);
    } finally {
        // 當buf在pipeline中處理完之后，釋放
        if (buf != null) {
            buf.release();
        }
    }
}

這里，我們詳細闡述一下Encoder是如何處理傳入的java對象的

1.判斷當前Handler是否能處理寫入的消息，如果能處理，進入下面的流程，否則，直接扔給下一個節點處理
2.將對象強制轉換成Encoder可以處理的 Response對象
3.分配一個ByteBuf
4.調用encoder，即進入到 Encoder 的 encode方法，該方法是用戶代碼，用戶將數據寫入ByteBuf
5.既然自定義java對象轉換成ByteBuf了，那么這個對象就已經無用了，釋放掉，(當傳入的msg類型是ByteBuf的時候，就不需要自己手動釋放了)
6.如果buf中寫入了數據，就把buf傳到下一個節點，否則，釋放buf，將空數據傳到下一個節點
7.最后，當buf在pipeline中處理完之后，釋放節點

總結一點就是，Encoder節點分配一個ByteBuf，調用encode方法，將java對象根據自定義協議寫入到ByteBuf，然后再把ByteBuf傳入到下一個節點，在我們的例子中，最終會傳入到head節點，因為head節點是一個OutBount類型的handler

HeadContext

public void write(ChannelHandlerContext ctx, Object msg, ChannelPromise promise) throws Exception {
    unsafe.write(msg, promise);
}

這里的msg就是前面在Encoder節點中，載有java對象數據的自定義ByteBuf對象，進入下一節

write：寫隊列

我們來看看channel中unsafe的write方法，先來看看其中的一個屬性

AbstractUnsafe

protected abstract class AbstractUnsafe implements Unsafe {
    private volatile ChannelOutboundBuffer outboundBuffer = new ChannelOutboundBuffer(AbstractChannel.this);

我們來看看 ChannelOutboundBuffer 這個類

public final class ChannelOutboundBuffer {
    private final Channel channel;
    private ChannelOutboundBuffer.Entry flushedEntry;
    private ChannelOutboundBuffer.Entry unflushedEntry;
    private ChannelOutboundBuffer.Entry tailEntry;

ChannelOutboundBuffer內部維護了一個Entry鏈表，並使用Entry封裝msg。其中的屬性我們下面會詳細講

我們回到正題，接着看 unsafe.write(msg, promise);

AbstractUnsafe

@Override
public final void write(Object msg, ChannelPromise promise) {
    assertEventLoop();

    ChannelOutboundBuffer outboundBuffer = this.outboundBuffer; int size;
    try {
        msg = filterOutboundMessage(msg);
        size = pipeline.estimatorHandle().size(msg);
        if (size < 0) {
            size = 0;
        }
    } catch (Throwable t) {
        safeSetFailure(promise, t);
        ReferenceCountUtil.release(msg);
        return;
    }

  outboundBuffer.addMessage(msg, size, promise);
}

1.調用 filterOutboundMessage() 方法，將待寫入的對象過濾，把非ByteBuf對象和FileRegion過濾，把所有的非直接內存轉換成直接內存DirectBuffer

@Override
protected final Object filterOutboundMessage(Object msg) {
    if (msg instanceof ByteBuf) {
        ByteBuf buf = (ByteBuf) msg;
        if (buf.isDirect()) {
            return msg;
        }

        return newDirectBuffer(buf);
    }

    if (msg instanceof FileRegion) {
        return msg;
    }

    throw new UnsupportedOperationException(
            "unsupported message type: " + StringUtil.simpleClassName(msg) + EXPECTED_TYPES);
}

2.接下來，估算出需要寫入的ByteBuf的size
3.最后，調用 ChannelOutboundBuffer 的addMessage(msg, size, promise) 方法，所以，接下來，我們需要重點看一下這個方法干了什么事情

ChannelOutboundBuffer

public void addMessage(Object msg, int size, ChannelPromise promise) {
    // 創建一個待寫出的消息節點
    Entry entry = Entry.newInstance(msg, size, total(msg), promise); if (tailEntry == null) {
        flushedEntry = null;
        tailEntry = entry;
    } else {
        Entry tail = tailEntry;
        tail.next = entry;
        tailEntry = entry;
    }
    if (unflushedEntry == null) {
        unflushedEntry = entry;
    }

    incrementPendingOutboundBytes(size, false);
}

想要理解上面這段代碼，必須得掌握寫緩存中的幾個消息指針，如下圖

ChannelOutboundBuffer 里面的數據結構是一個單鏈表結構，每個節點是一個 Entry，Entry 里面包含了待寫出ByteBuf 以及消息回調 promise，下面分別是三個指針的作用

1.flushedEntry 指針表示第一個被寫到操作系統Socket緩沖區中的節點
2.unFlushedEntry 指針表示第一個未被寫入到操作系統Socket緩沖區中的節點
3.tailEntry指針表示ChannelOutboundBuffer緩沖區的最后一個節點

初次調用 addMessage 之后，各個指針的情況為

fushedEntry指向空，unFushedEntry和 tailEntry 都指向新加入的節點

第二次調用 addMessage之后，各個指針的情況為

第n次調用 addMessage之后，各個指針的情況為

可以看到，調用n次addMessage，flushedEntry指針一直指向NULL，表示現在還未有節點需要寫出到Socket緩沖區，而unFushedEntry之后有n個節點，表示當前還有n個節點尚未寫出到Socket緩沖區中去

flush：刷新寫隊列

不管調用channel.flush()，還是ctx.flush()，最終都會落地到pipeline中的head節點

HeadContext

@Override
public void flush(ChannelHandlerContext ctx) throws Exception {
    unsafe.flush();
}

之后進入到AbstractUnsafe

AbstractUnsafe

public final void flush() {
   assertEventLoop();

   ChannelOutboundBuffer outboundBuffer = this.outboundBuffer;
   if (outboundBuffer == null) {
       return;
   }

   outboundBuffer.addFlush();
   flush0();
}

flush方法中，先調用 outboundBuffer.addFlush();

ChannelOutboundBuffer

public void addFlush() {
    Entry entry = unflushedEntry;
    if (entry != null) {
        if (flushedEntry == null) {
            flushedEntry = entry;
        }
        do {
            flushed ++;
            if (!entry.promise.setUncancellable()) {
                int pending = entry.cancel();
                decrementPendingOutboundBytes(pending, false, true);
            }
            entry = entry.next;
        } while (entry != null);
        unflushedEntry = null;
    }
}

可以結合前面的圖來看，首先拿到 unflushedEntry 指針，然后將 flushedEntry 指向unflushedEntry所指向的節點，調用完畢之后，三個指針的情況如下所示

 

相當於所有的節點都即將開始推送出去

接下來，調用 flush0();

AbstractUnsafe

protected void flush0() {
    doWrite(outboundBuffer);
}

發現這里的核心代碼就一個 doWrite，繼續跟

AbstractNioByteChannel

protected void doWrite(ChannelOutboundBuffer in) throws Exception {
    int writeSpinCount = -1;

    boolean setOpWrite = false;
    for (;;) {
        // 拿到第一個需要flush的節點的數據
        Object msg = in.current(); if (msg instanceof ByteBuf) {
            // 強轉為ByteBuf，若發現沒有數據可讀，直接刪除該節點
            ByteBuf buf = (ByteBuf) msg;

            boolean done = false;
            long flushedAmount = 0;
            // 拿到自旋鎖迭代次數
            if (writeSpinCount == -1) {
                writeSpinCount = config().getWriteSpinCount();
            }
            // 自旋，將當前節點寫出
            for (int i = writeSpinCount - 1; i >= 0; i --) {
                int localFlushedAmount = doWriteBytes(buf); if (localFlushedAmount == 0) {
                    setOpWrite = true;
                    break;
                }

                flushedAmount += localFlushedAmount;
                if (!buf.isReadable()) {
                    done = true;
                    break;
                }
            }

            in.progress(flushedAmount);

            // 寫完之后，將當前節點刪除
            if (done) {
                in.remove();
            } else {
                break;
            }
        } 
    }
}

這里略微有點復雜，我們分析一下

1.第一步，調用current()先拿到第一個需要flush的節點的數據

 ChannelOutBoundBuffer

public Object current() {
    Entry entry = flushedEntry;
    if (entry == null) {
        return null;
    }

    return entry.msg;
}

2.第二步,拿到自旋鎖的迭代次數

if (writeSpinCount == -1) {
    writeSpinCount = config().getWriteSpinCount();
}

3.自旋的方式將ByteBuf寫出到jdk nio的Channel

for (int i = writeSpinCount - 1; i >= 0; i --) {
    int localFlushedAmount = doWriteBytes(buf); if (localFlushedAmount == 0) {
        setOpWrite = true;
        break;
    }

    flushedAmount += localFlushedAmount;
    if (!buf.isReadable()) {
        done = true;
        break;
    }
}

doWriteBytes 方法跟進去

protected int doWriteBytes(ByteBuf buf) throws Exception {
    final int expectedWrittenBytes = buf.readableBytes();
    return buf.readBytes(javaChannel(), expectedWrittenBytes);
}

我們發現，出現了 javaChannel()，表明已經進入到了jdk nio Channel的領域，我們來看看 buf.readBytes(javaChannel(), expectedWrittenBytes);

public int readBytes(GatheringByteChannel out, int length) throws IOException {
    this.checkReadableBytes(length);
    int readBytes = this.getBytes(this.readerIndex, out, length); this.readerIndex += readBytes;
    return readBytes;
}

我們來看關鍵代碼 this.getBytes(this.readerIndex, out, length)

private int getBytes(int index, GatheringByteChannel out, int length, boolean internal) throws IOException {
    this.checkIndex(index, length);
    if (length == 0) {
        return 0;
    } else {
        ByteBuffer tmpBuf;
        if (internal) {
            tmpBuf = this.internalNioBuffer();
        } else {
            tmpBuf = ((ByteBuffer)this.memory).duplicate();
        }

        index = this.idx(index);
        tmpBuf.clear().position(index).limit(index + length);
        //將tmpBuf中的數據寫到out中
        return out.write(tmpBuf);
    }
}

我們來看看out.write(tmpBuf)

public int write(ByteBuffer src) throws IOException {
    ensureOpen();
    if (!writable)
        throw new NonWritableChannelException();
    synchronized (positionLock) {
        int n = 0;
        int ti = -1;
        try {
            begin();
            ti = threads.add();
            if (!isOpen())
                return 0;
            do {
                n = IOUtil.write(fd, src, -1, nd);
            } while ((n == IOStatus.INTERRUPTED) && isOpen());
            return IOStatus.normalize(n);
        } finally {
            threads.remove(ti);
            end(n > 0);
            assert IOStatus.check(n);
        }
    }
}

和read實現一樣，SocketChannelImpl的write方法通過IOUtil的write實現：關鍵代碼 n = IOUtil.write(fd, src, -1, nd);

static int write(FileDescriptor var0, ByteBuffer var1, long var2, NativeDispatcher var4) throws IOException {
    //如果是DirectBuffer，直接寫，將堆外緩存中的數據拷貝到內核緩存中進行發送
    if (var1 instanceof DirectBuffer) {
        return writeFromNativeBuffer(var0, var1, var2, var4);
    } else {
        //非DirectBuffer
        //獲取已經讀取到的位置
        int var5 = var1.position();
        //獲取可以讀到的位置
        int var6 = var1.limit();

        assert var5 <= var6;
        //申請一個原buffer可讀大小的DirectByteBuffer
        int var7 = var5 <= var6 ? var6 - var5 : 0;
        ByteBuffer var8 = Util.getTemporaryDirectBuffer(var7);

        int var10;
        try {

            var8.put(var1);
            var8.flip();
            var1.position(var5);
            //通過DirectBuffer寫，將堆外緩存的數據拷貝到內核緩存中進行發送
            int var9 = writeFromNativeBuffer(var0, var8, var2, var4);
            if (var9 > 0) {
                var1.position(var5 + var9);
            }

            var10 = var9;
        } finally {
            //回收分配的DirectByteBuffer
            Util.offerFirstTemporaryDirectBuffer(var8);
        }

        return var10;
    }
}

代碼邏輯我們就不再講了，代碼注釋已經很清楚了，這里我們關注一點，我們可以看看我們前面的一個方法 filterOutboundMessage()，將待寫入的對象過濾，把非ByteBuf對象和FileRegion過濾，把所有的非直接內存轉換成直接內存DirectBuffer

說明到了這一步所有的 var1 意境是直接內存DirectBuffer，就不需要走到else，就不需要write兩次了

4.刪除該節點

節點的數據已經寫入完畢，接下來就需要刪除該節點

ChannelOutBoundBuffer

public boolean remove() {
 Entry e = flushedEntry;
    Object msg = e.msg;

    ChannelPromise promise = e.promise;
    int size = e.pendingSize;

    removeEntry(e);

    if (!e.cancelled) {
        ReferenceCountUtil.safeRelease(msg);
        safeSuccess(promise);
    }

    // recycle the entry
    e.recycle();

    return true;
}

首先拿到當前被flush掉的節點(flushedEntry所指)，然后拿到該節點的回調對象 ChannelPromise, 調用 removeEntry()方法移除該節點

private void removeEntry(Entry e) {
    if (-- flushed == 0) {
        flushedEntry = null;
        if (e == tailEntry) {
            tailEntry = null;
            unflushedEntry = null;
        }
    } else {
       flushedEntry = e.next;
    }
}

這里的remove是邏輯移除，只是將flushedEntry指針移到下個節點，調用完畢之后，節點圖示如下

writeAndFlush: 寫隊列並刷新

理解了write和flush這兩個過程，writeAndFlush 也就不難了

public final ChannelFuture writeAndFlush(Object msg) {
    return tail.writeAndFlush(msg);
}

public ChannelFuture writeAndFlush(Object msg) {
    return writeAndFlush(msg, newPromise());
}

public ChannelFuture writeAndFlush(Object msg, ChannelPromise promise) {
    write(msg, true, promise);

    return promise;
}

private void write(Object msg, boolean flush, ChannelPromise promise) {
    AbstractChannelHandlerContext next = findContextOutbound();
    EventExecutor executor = next.executor();
    if (executor.inEventLoop()) {
        if (flush) {
            next.invokeWriteAndFlush(m, promise);
        } else {
            next.invokeWrite(m, promise);
        }
    } 
}

可以看到，最終，通過一個boolean變量，表示是調用 invokeWriteAndFlush，還是 invokeWrite，invokeWrite便是我們上文中的write過程

private void invokeWriteAndFlush(Object msg, ChannelPromise promise) {
    invokeWrite0(msg, promise);
    invokeFlush0();
}

可以看到，最終調用的底層方法和單獨調用 write 和 flush 是一樣的

private void invokeWrite(Object msg, ChannelPromise promise) {
        invokeWrite0(msg, promise);
}

private void invokeFlush(Object msg, ChannelPromise promise) {
        invokeFlush0(msg, promise);
}

由此看來，invokeWriteAndFlush基本等價於write方法之后再來一次flush

總結

1.pipeline中的編碼器原理是創建一個ByteBuf,將java對象轉換為ByteBuf，然后再把ByteBuf繼續向前傳遞
2.調用write方法並沒有將數據寫到Socket緩沖區中，而是寫到了一個單向鏈表的數據結構中，flush才是真正的寫出
3.writeAndFlush等價於先將數據寫到netty的緩沖區，再將netty緩沖區中的數據寫到Socket緩沖區中，寫的過程與並發編程類似，用自旋鎖保證寫成功
4.netty中的緩沖區中的ByteBuf為DirectByteBuf