Netty中的Channel之數據沖刷與線程安全（writeAndFlush）

本文首發我的博客： 貓叔的博客 | MySelf

GitHub項目地址

InChat

一個輕量級、高效率的支持多端（應用與硬件Iot）的異步網絡應用通信框架

前言

本文預設讀者已經瞭解了必定的Netty基礎知識，並可以本身構建一個Netty的通訊服務（包括客戶端與服務端）。那麼你必定使用到了Channel，這是Netty對傳統JavaIO、NIO的連接封裝實例。

那麼接下來讓咱們來了解一下關於Channel的數據沖刷與線程安全吧。

數據沖刷的步驟

一、獲取一個連接實例

@Override
public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
    //獲取連接實例
    Channel channel = ctx.channel();
}

我將案例放在初學者最熟悉的channelRead方法中，這是一個數據接收的方法，咱們自實現Netty的消息處理接口時須要重寫的方法。即客戶端發送消息後，這個方法會被觸發調用，因此咱們在這個方法中進行本次內容的講解。

由上一段代碼，其實目前仍是很簡單，咱們藉助ChannelHandlerContext（這是一個ChannelHandler與ChannelPipeline相交互並對接的一個對象。以下是源碼的解釋）來獲取目前的連接實例Channel。

/* Enables a {@link ChannelHandler} to interact with its {@link ChannelPipeline}
 * and other handlers. Among other things a handler can notify the next {@link ChannelHandler} in the
 * {@link ChannelPipeline} as well as modify the {@link ChannelPipeline} it belongs to dynamically.
 */
 public interface ChannelHandlerContext extends AttributeMap, ChannelInboundInvoker, ChannelOutboundInvoker {
     //......
 }

二、建立一個持有數據的ByteBuf

@Override
public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
    //獲取連接實例
    Channel channel = ctx.channel();
    //建立一個持有數據的ByteBuf
    ByteBuf buf = Unpooled.copiedBuffer("data", CharsetUtil.UTF_8);
}

ByteBuf又是什麼呢？

它是Netty框架本身封裝的 一個字符底層對象，是一個對 byte[] 和 ByteBuffer NIO 的抽象類，更官網的說就是「零個或多個字節的隨機和順序可訪問的序列。」，以下是源碼的解釋

/**
 * A random and sequential accessible sequence of zero or more bytes (octets).
 * This interface provides an abstract view for one or more primitive byte
 * arrays ({@code byte[]}) and {@linkplain ByteBuffer NIO buffers}.
 */
 public abstract class ByteBuf implements ReferenceCounted, Comparable<ByteBuf> {
     //......
 }

由上一段源碼能夠看出，ByteBuf是一個抽象類，因此咱們不能經過 new 的形式來建立一個新的ByteBuf對象。那麼咱們能夠經過Netty提供的一個 final 的工具類 Unpooled（你將其看做是一個建立ByteBuf的工具類就行了）。

/**
 * Creates a new {@link ByteBuf} by allocating new space or by wrapping
 * or copying existing byte arrays, byte buffers and a string.
 */
 public final class Unpooled {
     //......
 }

這真是一個有趣的過程，那麼接下來咱們僅須要再看看 copiedBuffer 這個方法了。這個方法相對簡單，就是咱們將建立一個新的緩衝區，其內容是咱們指定的 UTF-8字符集 編碼指定的 「data」 ，同時這個新的緩衝區的讀索引和寫索引分別是0和字符串的長度。

三、沖刷數據

@Override
public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
    //獲取連接實例
    Channel channel = ctx.channel();
    //建立一個持有數據的ByteBuf
    ByteBuf buf = Unpooled.copiedBuffer("data", CharsetUtil.UTF_8);
    //數據沖刷
    channel.writeAndFlush(buf);
}

我相信大部分人都是直接這麼寫的，由於咱們常常理所固然的啓動測試，並在客戶端接受到了這個 「data」 消息。那麼咱們是否應該注意一下，這個數據沖刷會返回一個什麼值，咱們要如何才能在服務端知道，此次數據沖刷是成功仍是失敗呢？

那麼其實Netty框架已經考慮到了這個點，本次數據沖刷咱們將獲得一個 ChannelFuture 。

@Override
public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
    //獲取連接實例
    Channel channel = ctx.channel();
    //建立一個持有數據的ByteBuf
    ByteBuf buf = Unpooled.copiedBuffer("data", CharsetUtil.UTF_8);
    //數據沖刷
    ChannelFuture cf = channel.writeAndFlush(buf);
}

是的，他就是 Channel 異步IO操做的結果，它是一個接口，並繼承了Future<V>。（以下爲源碼的解釋）

/**
 * The result of an asynchronous {@link Channel} I/O operation.
 */
 public interface ChannelFuture extends Future<Void> {
     //......
 }

既然如此，那麼咱們能夠明顯的知道咱們能夠對其添加對應的監聽。

四、異步回調結果監聽

@Override
public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
    //獲取連接實例
    Channel channel = ctx.channel();
    //建立一個持有數據的ByteBuf
    ByteBuf buf = Unpooled.copiedBuffer("data", CharsetUtil.UTF_8);
    //數據沖刷
    ChannelFuture cf = channel.writeAndFlush(buf);
    //添加ChannelFutureListener以便在寫操做完成後接收通知
    cf.addListener(new ChannelFutureListener() {
        @Override
        public void operationComplete(ChannelFuture future) throws Exception {
            //寫操做完成，並無錯誤發生
            if (future.isSuccess()){
                System.out.println("successful");
            }else{
                //記錄錯誤
                System.out.println("error");
                future.cause().printStackTrace();
            }
        }
    });
}

好的，咱們能夠簡單的從代碼理解到，咱們將經過對異步IO的結果監聽，獲得本次運行的結果。我想這纔是一個相對完整的 數據沖刷（writeAndFlush）。

測試線程安全的流程

對於線程安全的測試，咱們將模擬多個線程去執行數據沖刷操做，咱們能夠用到 Executor 。

咱們能夠這樣理解 Executor ，是一種省略了線程啓用與調度的方式，你只須要傳遞一個 Runnable 給它便可，你再也不須要去 start 一個線程。（以下是源碼的解釋）

/**
 * An object that executes submitted {@link Runnable} tasks. This
 * interface provides a way of decoupling task submission from the
 * mechanics of how each task will be run, including details of thread
 * use, scheduling, etc.  An {@code Executor} is normally used
 * instead of explicitly creating threads. For example, rather than
 * invoking {@code new Thread(new(RunnableTask())).start()} for each
 * of a set of tasks, you might use:...
 */
 public interface Executor {
     //......
 }

那麼咱們的測試代碼，大體是這樣的。

final Channel channel = ctx.channel();
//建立要寫數據的ByteBuf
final ByteBuf buf = Unpooled.copiedBuffer("data",CharsetUtil.UTF_8).retain();
//建立將數據寫到Channel的Runnable
Runnable writer = new Runnable() {
    @Override
    public void run() {
        ChannelFuture cf = channel.writeAndFlush(buf.duplicate());
        cf.addListener(new ChannelFutureListener() {
            @Override
            public void operationComplete(ChannelFuture future) throws Exception {
                //寫操做完成，並無錯誤發生
                if (future.isSuccess()){
                    System.out.println("successful");
                }else{
                    //記錄錯誤
                    System.out.println("error");
                    future.cause().printStackTrace();
                }
            }
        });
    }
};

//獲取到線程池的Executor的引用
Executor executor = Executors.newCachedThreadPool();

//提交到某個線程中執行
executor.execute(writer);

//提交到另外一個線程中執行
executor.execute(writer);

這裏，咱們須要注意的是：

建立 ByteBuf 的時候，咱們使用了 retain 這個方法，他是將咱們生成的這個 ByteBuf 進行保留操做。

在 ByteBuf 中有這樣的一種區域： 非保留和保留派生緩衝區。

這裏有點複雜，咱們能夠簡單的理解，若是調用了 retain 那麼數據就存在派生緩衝區中，若是沒有調用，則會在調用後，移除這一個字符數據。（以下是 ByteBuf 源碼的解釋）

/*<h4>Non-retained and retained derived buffers</h4>
 *
 * Note that the {@link #duplicate()}, {@link #slice()}, {@link #slice(int, int)} and {@link #readSlice(int)} does NOT
 * call {@link #retain()} on the returned derived buffer, and thus its reference count will NOT be increased. If you
 * need to create a derived buffer with increased reference count, consider using {@link #retainedDuplicate()},
 * {@link #retainedSlice()}, {@link #retainedSlice(int, int)} and {@link #readRetainedSlice(int)} which may return
 * a buffer implementation that produces less garbage.
 */

好的，我想你能夠本身動手去測試一下，最好再看看源碼，加深一下實現的原理印象。

這裏的線程池並非現實線程安全，而是用來作測試多線程的，Netty的Channel實現是線程安全的，因此咱們能夠存儲一個到Channel的引用，而且每當咱們須要向遠程節點寫數據時，均可以使用它，即便當時許多線程都在使用它，消息也會被保證按順序發送的。

結語

最後，介紹一下，我的的一個基於Netty的開源項目：InChat

一個輕量級、高效率的支持多端（應用與硬件Iot）的異步網絡應用通信框架

---

參考資料： 《Netty實戰》