Netty中的Channel之資料沖刷與執行緒安全（writeAndFlush）
本文首發公眾號與個人部落格：Java貓說 & 貓叔的部落格 | MySelf，轉載請申明出處。
GitHub專案地址
一個輕量級、高效率的支援多端（應用與硬體Iot）的非同步網路應用通訊框架
前言
本文預設讀者已經瞭解了一定的Netty基礎知識，並能夠自己構建一個Netty的通訊服務（包括客戶端與服務端）。那麼你一定使用到了Channel，這是Netty對傳統JavaIO、NIO的連結封裝例項。
那麼接下來讓我們來了解一下關於Channel的資料沖刷與執行緒安全吧。
資料沖刷的步驟
1、獲取一個連結例項
@Override 我將案例放在初學者最熟悉的channelRead方法中，這是一個數據接收的方法，我們自實現Netty的訊息處理介面時需要重寫的方法。即客戶端傳送訊息後，這個方法會被觸發呼叫，所以我們在這個方法中進行本次內容的講解。
由上一段程式碼，其實目前還是很簡單，我們藉助ChannelHandlerContext（這是一個ChannelHandler與ChannelPipeline相互動並對接的一個物件
/* 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.
*/ 2、建立一個持有資料的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和字串的長度。
3、沖刷資料
@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。（如下為原始碼的解釋）
/**
* The result of an asynchronous {@link Channel} I/O operation.
*/
public interface ChannelFuture extends Future<Void> {
//......
}
複製程式碼既然如此，那麼我們可以明顯的知道我們可以對其新增對應的監聽。
4、非同步回撥結果監聽
@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實戰》
公眾號：Java貓說
現架構設計（碼農）兼創業技術顧問，不羈平庸，熱愛開源，雜談程式人生與不定期乾貨。