# 開始
[TOC=2,2]
本章圍繞 Netty 的核心架構，通過簡單的示例帶你快速入門。當你讀完本章節，你馬上就可以用 Netty 寫出一個客戶端和服務器。
如果你在學習的時候喜歡“top-down（自頂向下）”，那你可能需要要從第二章《[Architectural Overview （架構總覽）](#)》開始，然后再回到這里。
## 開始之前
在運行本章示例之前，需要準備：最新版的 Netty 以及 JDK 1.6 或以上版本。最新版的 Netty 在這[下載](http://netty.io/downloads.html)。自行下載 JDK。
閱讀本章節過程中，你可能會對相關類有疑惑，關于這些類的詳細的信息請請參考 API 說明文檔。為了方便，所有文檔中涉及到的類名字都會被關聯到一個在線的 API 說明。當然，如果有任何錯誤信息、語法錯誤或者你有任何好的建議來改進文檔說明，那么請[聯系Netty社區](http://netty.io/community.html)。
*譯者注：對本翻譯有任何疑問，在[https://github.com/waylau/netty-4-user-guide/issues](https://github.com/waylau/netty-4-user-guide/issues)提問*
## 寫個丟棄服務器
世上最簡單的協議不是'Hello, World!' 而是 [DISCARD(丟棄)](http://tools.ietf.org/html/rfc863)。這個協議將會丟掉任何收到的數據，而不響應。
為了實現 DISCARD 協議，你只需忽略所有收到的數據。讓我們從 handler （處理器）的實現開始，handler 是由 Netty 生成用來處理 I/O 事件的。
~~~
import io.netty.buffer.ByteBuf;
?
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.ChannelInboundHandlerAdapter;
?
/**
* 處理服務端 channel.
*/
public class DiscardServerHandler extends ChannelInboundHandlerAdapter { // (1)
?
@Override
public void channelRead(ChannelHandlerContext ctx, Object msg) { // (2)
// 默默地丟棄收到的數據
((ByteBuf) msg).release(); // (3)
}
?
@Override
public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) { // (4)
// 當出現異常就關閉連接
cause.printStackTrace();
ctx.close();
}
}
~~~
1.DiscardServerHandler 繼承自 [ChannelInboundHandlerAdapter](http://netty.io/4.0/api/io/netty/channel/ChannelInboundHandlerAdapter.html)，這個類實現了 [ChannelInboundHandler](http://netty.io/4.0/api/io/netty/channel/ChannelInboundHandler.html)接口，ChannelInboundHandler 提供了許多事件處理的接口方法，然后你可以覆蓋這些方法。現在僅僅只需要繼承 ChannelInboundHandlerAdapter 類而不是你自己去實現接口方法。
2.這里我們覆蓋了 chanelRead() 事件處理方法。每當從客戶端收到新的數據時，這個方法會在收到消息時被調用，這個例子中，收到的消息的類型是 [ByteBuf](http://netty.io/4.0/api/io/netty/buffer/ByteBuf.html)
3.為了實現 DISCARD 協議，處理器不得不忽略所有接受到的消息。ByteBuf 是一個引用計數對象，這個對象必須顯示地調用 release() 方法來釋放。請記住處理器的職責是釋放所有傳遞到處理器的引用計數對象。通常，channelRead() 方法的實現就像下面的這段代碼：
~~~
?
@Override
public void channelRead(ChannelHandlerContext ctx, Object msg) {
try {
// Do something with msg
} finally {
ReferenceCountUtil.release(msg);
}
}
~~~
4.exceptionCaught() 事件處理方法是當出現 Throwable 對象才會被調用，即當 Netty 由于 IO 錯誤或者處理器在處理事件時拋出的異常時。在大部分情況下，捕獲的異常應該被記錄下來并且把關聯的 channel 給關閉掉。然而這個方法的處理方式會在遇到不同異常的情況下有不同的實現，比如你可能想在關閉連接之前發送一個錯誤碼的響應消息。
目前為止一切都還不錯，我們已經實現了 DISCARD 服務器的一半功能，剩下的需要編寫一個 main() 方法來啟動服務端的 DiscardServerHandler。
~~~
import io.netty.bootstrap.ServerBootstrap;
?
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.ChannelOption;
import io.netty.channel.EventLoopGroup;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioServerSocketChannel;
?
/**
* 丟棄任何進入的數據
*/
public class DiscardServer {
?
private int port;
?
public DiscardServer(int port) {
this.port = port;
}
?
public void run() throws Exception {
EventLoopGroup bossGroup = new NioEventLoopGroup(); // (1)
EventLoopGroup workerGroup = new NioEventLoopGroup();
try {
ServerBootstrap b = new ServerBootstrap(); // (2)
b.group(bossGroup, workerGroup)
.channel(NioServerSocketChannel.class) // (3)
.childHandler(new ChannelInitializer<SocketChannel>() { // (4)
@Override
public void initChannel(SocketChannel ch) throws Exception {
ch.pipeline().addLast(new DiscardServerHandler());
}
})
.option(ChannelOption.SO_BACKLOG, 128) // (5)
.childOption(ChannelOption.SO_KEEPALIVE, true); // (6)
?
// 綁定端口，開始接收進來的連接
ChannelFuture f = b.bind(port).sync(); // (7)
?
// 等待服務器 socket 關閉 。
// 在這個例子中，這不會發生，但你可以優雅地關閉你的服務器。
f.channel().closeFuture().sync();
} finally {
workerGroup.shutdownGracefully();
bossGroup.shutdownGracefully();
}
}
?
public static void main(String[] args) throws Exception {
int port;
if (args.length > 0) {
port = Integer.parseInt(args[0]);
} else {
port = 8080;
}
new DiscardServer(port).run();
}
}
~~~
1.[NioEventLoopGroup](http://netty.io/4.0/api/io/netty/channel/nio/NioEventLoopGroup.html) 是用來處理I/O操作的多線程事件循環器，Netty 提供了許多不同的 [EventLoopGroup](http://netty.io/4.0/api/io/netty/channel/EventLoopGroup.html) 的實現用來處理不同的傳輸。在這個例子中我們實現了一個服務端的應用，因此會有2個 NioEventLoopGroup 會被使用。第一個經常被叫做‘boss’，用來接收進來的連接。第二個經常被叫做‘worker’，用來處理已經被接收的連接，一旦‘boss’接收到連接，就會把連接信息注冊到‘worker’上。如何知道多少個線程已經被使用，如何映射到已經創建的 [Channel](http://netty.io/4.0/api/io/netty/channel/Channel.html)上都需要依賴于 EventLoopGroup 的實現，并且可以通過構造函數來配置他們的關系。
2.[ServerBootstrap](http://netty.io/4.0/api/io/netty/bootstrap/ServerBootstrap.html) 是一個啟動 NIO 服務的輔助啟動類。你可以在這個服務中直接使用 Channel，但是這會是一個復雜的處理過程，在很多情況下你并不需要這樣做。
3.這里我們指定使用 [NioServerSocketChannel](http://netty.io/4.0/api/io/netty/channel/socket/nio/NioServerSocketChannel.html) 類來舉例說明一個新的 Channel 如何接收進來的連接。
4.這里的事件處理類經常會被用來處理一個最近的已經接收的 Channel。[ChannelInitializer](http://netty.io/4.0/api/io/netty/channel/ChannelInitializer.html) 是一個特殊的處理類，他的目的是幫助使用者配置一個新的 Channel。也許你想通過增加一些處理類比如DiscardServerHandler 來配置一個新的 Channel 或者其對應的[ChannelPipeline](http://netty.io/4.0/api/io/netty/channel/ChannelPipeline.html) 來實現你的網絡程序。當你的程序變的復雜時，可能你會增加更多的處理類到 pipline 上，然后提取這些匿名類到最頂層的類上。
5.你可以設置這里指定的 Channel 實現的配置參數。我們正在寫一個TCP/IP 的服務端，因此我們被允許設置 socket 的參數選項比如tcpNoDelay 和 keepAlive。請參考 [ChannelOption](http://netty.io/4.0/api/io/netty/channel/ChannelOption.html) 和詳細的 [ChannelConfig](http://netty.io/4.0/api/io/netty/channel/ChannelConfig.html) 實現的接口文檔以此可以對ChannelOption 的有一個大概的認識。
6.你關注過 option() 和 childOption() 嗎？option() 是提供給[NioServerSocketChannel](http://netty.io/4.0/api/io/netty/channel/socket/nio/NioServerSocketChannel.html) 用來接收進來的連接。childOption() 是提供給由父管道 [ServerChannel](http://netty.io/4.0/api/io/netty/channel/ServerChannel.html) 接收到的連接，在這個例子中也是 NioServerSocketChannel。
7.我們繼續，剩下的就是綁定端口然后啟動服務。這里我們在機器上綁定了機器所有網卡上的 8080 端口。當然現在你可以多次調用 bind() 方法(基于不同綁定地址)。
恭喜！你已經熟練地完成了第一個基于 Netty 的服務端程序。
## 查看收到的數據
現在我們已經編寫出我們第一個服務端，我們需要測試一下他是否真的可以運行。最簡單的測試方法是用 telnet 命令。例如，你可以在命令行上輸入`telnet localhost 8080`或者其他類型參數。
然而我們能說這個服務端是正常運行了嗎？事實上我們也不知道，因為他是一個 discard 服務，你根本不可能得到任何的響應。為了證明他仍然是在正常工作的，讓我們修改服務端的程序來打印出他到底接收到了什么。
我們已經知道 channelRead() 方法是在數據被接收的時候調用。讓我們放一些代碼到 DiscardServerHandler 類的 channelRead() 方法。
~~~
@Override
public void channelRead(ChannelHandlerContext ctx, Object msg) {
ByteBuf in = (ByteBuf) msg;
try {
while (in.isReadable()) { // (1)
System.out.print((char) in.readByte());
System.out.flush();
}
} finally {
ReferenceCountUtil.release(msg); // (2)
}
}
~~~
1.這個低效的循環事實上可以簡化為:System.out.println(in.toString(io.netty.util.CharsetUtil.US_ASCII))
2.或者，你可以在這里調用 in.release()。
如果你再次運行 telnet 命令，你將會看到服務端打印出了他所接收到的消息。
完整的discard server代碼放在了[io.netty.example.discard](http://netty.io/4.0/xref/io/netty/example/discard/package-summary.html)包下面。
*譯者注：翻譯版本的項目源碼見 [https://github.com/waylau/netty-4-user-guide-demos](https://github.com/waylau/netty-4-user-guide-demos) 中的`com.waylau.netty.demo.discard` 包下*
## 寫個應答服務器
到目前為止，我們雖然接收到了數據，但沒有做任何的響應。然而一個服務端通常會對一個請求作出響應。讓我們學習怎樣在 [ECHO](http://tools.ietf.org/html/rfc862) 協議的實現下編寫一個響應消息給客戶端，這個協議針對任何接收的數據都會返回一個響應。
和 discard server 唯一不同的是把在此之前我們實現的 channelRead() 方法，返回所有的數據替代打印接收數據到控制臺上的邏輯。因此，需要把 channelRead() 方法修改如下：
~~~
?
@Override
public void channelRead(ChannelHandlerContext ctx, Object msg) {
ctx.write(msg); // (1)
ctx.flush(); // (2)
}
~~~
1. [ChannelHandlerContext](http://netty.io/4.0/api/io/netty/channel/ChannelHandlerContext.html) 對象提供了許多操作，使你能夠觸發各種各樣的 I/O 事件和操作。這里我們調用了 write(Object) 方法來逐字地把接受到的消息寫入。請注意不同于 DISCARD 的例子我們并沒有釋放接受到的消息，這是因為當寫入的時候 Netty 已經幫我們釋放了。
1. ctx.write(Object) 方法不會使消息寫入到通道上，他被緩沖在了內部，你需要調用 ctx.flush() 方法來把緩沖區中數據強行輸出。或者你可以用更簡潔的 cxt.writeAndFlush(msg) 以達到同樣的目的。
如果你再一次運行 telnet 命令，你會看到服務端會發回一個你已經發送的消息。
完整的echo服務的代碼放在了 [io.netty.example.echo](http://netty.io/4.0/xref/io/netty/example/echo/package-summary.html)包下面。
*譯者注：翻譯版本的項目源碼見 [https://github.com/waylau/netty-4-user-guide-demos](https://github.com/waylau/netty-4-user-guide-demos) 中的`com.waylau.netty.demo.echo` 包下*
## 寫個時間服務器
在這個部分被實現的協議是 [TIME](http://tools.ietf.org/html/rfc868) 協議。和之前的例子不同的是在不接受任何請求時他會發送一個含32位的整數的消息，并且一旦消息發送就會立即關閉連接。在這個例子中，你會學習到如何構建和發送一個消息，然后在完成時關閉連接。
因為我們將會忽略任何接收到的數據，而只是在連接被創建發送一個消息，所以這次我們不能使用 channelRead() 方法了，代替他的是，我們需要覆蓋 channelActive() 方法，下面的就是實現的內容：
~~~
public class TimeServerHandler extends ChannelInboundHandlerAdapter {
?
@Override
public void channelActive(final ChannelHandlerContext ctx) { // (1)
final ByteBuf time = ctx.alloc().buffer(4); // (2)
time.writeInt((int) (System.currentTimeMillis() / 1000L + 2208988800L));
?
final ChannelFuture f = ctx.writeAndFlush(time); // (3)
f.addListener(new ChannelFutureListener() {
@Override
public void operationComplete(ChannelFuture future) {
assert f == future;
ctx.close();
}
}); // (4)
}
?
@Override
public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) {
cause.printStackTrace();
ctx.close();
}
}
~~~
1.channelActive() 方法將會在連接被建立并且準備進行通信時被調用。因此讓我們在這個方法里完成一個代表當前時間的32位整數消息的構建工作。
2.為了發送一個新的消息，我們需要分配一個包含這個消息的新的緩沖。因為我們需要寫入一個32位的整數，因此我們需要一個至少有4個字節的 [ByteBuf](http://netty.io/4.0/api/io/netty/buffer/ByteBuf.html)。通過 ChannelHandlerContext.alloc() 得到一個當前的[ByteBufAllocator](http://netty.io/4.0/api/io/netty/buffer/ByteBufAllocator.html)，然后分配一個新的緩沖。
3.和往常一樣我們需要編寫一個構建好的消息。但是等一等，flip 在哪？難道我們使用 NIO 發送消息時不是調用 java.nio.ByteBuffer.flip() 嗎？ByteBuf 之所以沒有這個方法因為有兩個指針，一個對應讀操作一個對應寫操作。當你向 ByteBuf 里寫入數據的時候寫指針的索引就會增加，同時讀指針的索引沒有變化。讀指針索引和寫指針索引分別代表了消息的開始和結束。
比較起來，NIO 緩沖并沒有提供一種簡潔的方式來計算出消息內容的開始和結尾，除非你調用 flip 方法。當你忘記調用 flip 方法而引起沒有數據或者錯誤數據被發送時，你會陷入困境。這樣的一個錯誤不會發生在 Netty 上，因為我們對于不同的操作類型有不同的指針。你會發現這樣的使用方法會讓你過程變得更加的容易，因為你已經習慣一種沒有使用 flip 的方式。
另外一個點需要注意的是 ChannelHandlerContext.write() (和 writeAndFlush() )方法會返回一個 [ChannelFuture](http://netty.io/4.0/api/io/netty/channel/ChannelFuture.html) 對象，一個 ChannelFuture 代表了一個還沒有發生的 I/O 操作。這意味著任何一個請求操作都不會馬上被執行，因為在 Netty 里所有的操作都是異步的。舉個例子下面的代碼中在消息被發送之前可能會先關閉連接。
~~~
?
Channel ch = ...;
ch.writeAndFlush(message);
ch.close();
~~~
因此你需要在 write() 方法返回的 ChannelFuture 完成后調用 close() 方法，然后當他的寫操作已經完成他會通知他的監聽者。請注意,close() 方法也可能不會立馬關閉，他也會返回一個ChannelFuture。
4.當一個寫請求已經完成是如何通知到我們？這個只需要簡單地在返回的 ChannelFuture 上增加一個[ChannelFutureListener](http://netty.io/4.0/api/io/netty/channel/ChannelFutureListener.html)。這里我們構建了一個匿名的 ChannelFutureListener 類用來在操作完成時關閉 Channel。
或者，你可以使用簡單的預定義監聽器代碼:
~~~
f.addListener(ChannelFutureListener.CLOSE);
~~~
為了測試我們的time服務如我們期望的一樣工作，你可以使用 UNIX 的 rdate 命令
~~~
$ rdate -o <port> -p <host>
~~~
Port 是你在main()函數中指定的端口，host 使用 locahost 就可以了。
## 寫個時間客戶端
不像 DISCARD 和 ECHO 的服務端，對于 TIME 協議我們需要一個客戶端,因為人們不能把一個32位的二進制數據翻譯成一個日期或者日歷。在這一部分，我們將會討論如何確保服務端是正常工作的，并且學習怎樣用Netty 編寫一個客戶端。
在 Netty 中,編寫服務端和客戶端最大的并且唯一不同的使用了不同的[BootStrap](http://netty.io/4.0/api/io/netty/bootstrap/Bootstrap.html) 和 [Channel](http://netty.io/4.0/api/io/netty/channel/Channel.html)的實現。請看一下下面的代碼：
~~~
public class TimeClient {
?
public static void main(String[] args) throws Exception {
?
String host = args[0];
int port = Integer.parseInt(args[1]);
EventLoopGroup workerGroup = new NioEventLoopGroup();
?
try {
Bootstrap b = new Bootstrap(); // (1)
b.group(workerGroup); // (2)
b.channel(NioSocketChannel.class); // (3)
b.option(ChannelOption.SO_KEEPALIVE, true); // (4)
b.handler(new ChannelInitializer<SocketChannel>() {
@Override
public void initChannel(SocketChannel ch) throws Exception {
ch.pipeline().addLast(new TimeClientHandler());
}
});
?
// 啟動客戶端
ChannelFuture f = b.connect(host, port).sync(); // (5)
?
// 等待連接關閉
f.channel().closeFuture().sync();
} finally {
workerGroup.shutdownGracefully();
}
}
}
~~~
1.BootStrap 和 [ServerBootstrap](http://netty.io/4.0/api/io/netty/bootstrap/ServerBootstrap.html) 類似,不過他是對非服務端的 channel 而言，比如客戶端或者無連接傳輸模式的 channel。
2.如果你只指定了一個 [EventLoopGroup](http://netty.io/4.0/api/io/netty/channel/EventLoopGroup.html)，那他就會即作為一個 boss group ，也會作為一個 workder group，盡管客戶端不需要使用到 boss worker 。
3.代替[NioServerSocketChannel](http://netty.io/4.0/api/io/netty/channel/socket/nio/NioServerSocketChannel.html)的是[NioSocketChannel](http://netty.io/4.0/api/io/netty/channel/socket/nio/NioSocketChannel.html),這個類在客戶端channel 被創建時使用。
4.不像在使用 ServerBootstrap 時需要用 childOption() 方法，因為客戶端的 [SocketChannel](http://netty.io/4.0/api/io/netty/channel/socket/SocketChannel.html) 沒有父親。
5.我們用 connect() 方法代替了 bind() 方法。
正如你看到的，他和服務端的代碼是不一樣的。[ChannelHandler](http://netty.io/4.0/api/io/netty/channel/ChannelHandler.html) 是如何實現的?他應該從服務端接受一個32位的整數消息，把他翻譯成人們能讀懂的格式，并打印翻譯好的時間，最后關閉連接:
~~~
import java.util.Date;
?
public class TimeClientHandler extends ChannelInboundHandlerAdapter {
@Override
public void channelRead(ChannelHandlerContext ctx, Object msg) {
ByteBuf m = (ByteBuf) msg; // (1)
try {
long currentTimeMillis = (m.readUnsignedInt() - 2208988800L) * 1000L;
System.out.println(new Date(currentTimeMillis));
ctx.close();
} finally {
m.release();
}
}
?
@Override
public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) {
cause.printStackTrace();
ctx.close();
}
}
~~~
1.在TCP/IP中，Netty 會把讀到的數據放到 ByteBuf 的數據結構中。
這樣看起來非常簡單，并且和服務端的那個例子的代碼也相差不多。然而，處理器有時候會因為拋出 IndexOutOfBoundsException 而拒絕工作。在下個部分我們會討論為什么會發生這種情況。
## 處理一個基于流的傳輸
### 關于 Socket Buffer的一個小警告
基于流的傳輸比如 TCP/IP, 接收到數據是存在 socket 接收的 buffer 中。不幸的是，基于流的傳輸并不是一個數據包隊列，而是一個字節隊列。意味著，即使你發送了2個獨立的數據包，操作系統也不會作為2個消息處理而僅僅是作為一連串的字節而言。因此這是不能保證你遠程寫入的數據就會準確地讀取。舉個例子，讓我們假設操作系統的 TCP/TP 協議棧已經接收了3個數據包：
由于基于流傳輸的協議的這種普通的性質，在你的應用程序里讀取數據的時候會有很高的可能性被分成下面的片段
因此，一個接收方不管他是客戶端還是服務端，都應該把接收到的數據整理成一個或者多個更有意思并且能夠讓程序的業務邏輯更好理解的數據。在上面的例子中，接收到的數據應該被構造成下面的格式：
### The First Solution 辦法一
回到 TIME 客戶端例子。同樣也有類似的問題。一個32位整型是非常小的數據，他并不見得會被經常拆分到到不同的數據段內。然而，問題是他確實可能會被拆分到不同的數據段內，并且拆分的可能性會隨著通信量的增加而增加。
最簡單的方案是構造一個內部的可積累的緩沖，直到4個字節全部接收到了內部緩沖。下面的代碼修改了 TimeClientHandler 的實現類修復了這個問題
~~~
public class TimeClientHandler extends ChannelInboundHandlerAdapter {
private ByteBuf buf;
?
@Override
public void handlerAdded(ChannelHandlerContext ctx) {
buf = ctx.alloc().buffer(4); // (1)
}
?
@Override
public void handlerRemoved(ChannelHandlerContext ctx) {
buf.release(); // (1)
buf = null;
}
?
@Override
public void channelRead(ChannelHandlerContext ctx, Object msg) {
ByteBuf m = (ByteBuf) msg;
buf.writeBytes(m); // (2)
m.release();
?
if (buf.readableBytes() >= 4) { // (3)
long currentTimeMillis = (buf.readUnsignedInt() - 2208988800L) * 1000L;
System.out.println(new Date(currentTimeMillis));
ctx.close();
}
}
?
@Override
public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) {
cause.printStackTrace();
ctx.close();
}
}
~~~
1.[ChannelHandler](http://netty.io/4.0/api/io/netty/channel/ChannelHandler.html) 有2個生命周期的監聽方法：handlerAdded()和 handlerRemoved()。你可以完成任意初始化任務只要他不會被阻塞很長的時間。
2.首先，所有接收的數據都應該被累積在 buf 變量里。
3.然后，處理器必須檢查 buf 變量是否有足夠的數據，在這個例子中是4個字節，然后處理實際的業務邏輯。否則，Netty 會重復調用channelRead() 當有更多數據到達直到4個字節的數據被積累。
### The Second Solution 方法二
盡管第一個解決方案已經解決了 TIME 客戶端的問題了，但是修改后的處理器看起來不那么的簡潔，想象一下如果由多個字段比如可變長度的字段組成的更為復雜的協議時，你的 [ChannelInboundHandler](http://netty.io/4.0/api/io/netty/channel/ChannelInboundHandler.html) 的實現將很快地變得難以維護。
正如你所知的，你可以增加多個 [ChannelHandler](http://netty.io/4.0/api/io/netty/channel/ChannelHandler.html) 到[ChannelPipeline](http://netty.io/4.0/api/io/netty/channel/ChannelPipeline.html) ,因此你可以把一整個ChannelHandler 拆分成多個模塊以減少應用的復雜程度，比如你可以把TimeClientHandler 拆分成2個處理器：
- TimeDecoder 處理數據拆分的問題
- TimeClientHandler 原始版本的實現
幸運地是，Netty 提供了一個可擴展的類，幫你完成 TimeDecoder 的開發。
~~~
public class TimeDecoder extends ByteToMessageDecoder { // (1)
@Override
protected void decode(ChannelHandlerContext ctx, ByteBuf in, List<Object> out) { // (2)
if (in.readableBytes() < 4) {
return; // (3)
}
?
out.add(in.readBytes(4)); // (4)
}
}
~~~
1.[ByteToMessageDecoder](http://netty.io/4.0/api/io/netty/handler/codec/ByteToMessageDecoder.html) 是 [ChannelInboundHandler](http://netty.io/4.0/api/io/netty/channel/ChannelInboundHandler.html) 的一個實現類，他可以在處理數據拆分的問題上變得很簡單。
2.每當有新數據接收的時候，ByteToMessageDecoder 都會調用 decode() 方法來處理內部的那個累積緩沖。
3.Decode() 方法可以決定當累積緩沖里沒有足夠數據時可以往 out 對象里放任意數據。當有更多的數據被接收了 ByteToMessageDecoder 會再一次調用 decode() 方法。
4.如果在 decode() 方法里增加了一個對象到 out 對象里，這意味著解碼器解碼消息成功。ByteToMessageDecoder 將會丟棄在累積緩沖里已經被讀過的數據。請記得你不需要對多條消息調用 decode()，ByteToMessageDecoder 會持續調用 decode() 直到不放任何數據到 out 里。
現在我們有另外一個處理器插入到 [ChannelPipeline](http://netty.io/4.0/api/io/netty/channel/ChannelPipeline.html) 里，我們應該在 TimeClient 里修改 ChannelInitializer 的實現：
~~~
b.handler(new ChannelInitializer<SocketChannel>() {
@Override
public void initChannel(SocketChannel ch) throws Exception {
ch.pipeline().addLast(new TimeDecoder(), new TimeClientHandler());
}
});
~~~
如果你是一個大膽的人，你可能會嘗試使用更簡單的解碼類[ReplayingDecoder](http://netty.io/4.0/api/io/netty/handler/codec/ReplayingDecoder.html)。不過你還是需要參考一下 API 文檔來獲取更多的信息。
~~~
public class TimeDecoder extends ReplayingDecoder<Void> {
@Override
protected void decode(
ChannelHandlerContext ctx, ByteBuf in, List<Object> out) {
out.add(in.readBytes(4));
}
}
~~~
此外，Netty還提供了更多開箱即用的解碼器使你可以更簡單地實現更多的協議，幫助你避免開發一個難以維護的處理器實現。請參考下面的包以獲取更多更詳細的例子：
- 對于二進制協議請看 [io.netty.example.factorial](http://netty.io/4.0/xref/io/netty/example/factorial/package-summary.html)
- 對于基于文本協議請看 [io.netty.example.telnet](http://netty.io/4.0/xref/io/netty/example/telnet/package-summary.html)
*譯者注：翻譯版本的項目源碼見 [https://github.com/waylau/netty-4-user-guide-demos](https://github.com/waylau/netty-4-user-guide-demos) 中的`com.waylau.netty.demo.factorial` 和 `com.waylau.netty.demo.telnet` 包下*
## 用POJO代替ByteBuf
我們回顧了迄今為止的所有例子使用 [ByteBuf](http://netty.io/4.0/api/io/netty/buffer/ByteBuf.html) 作為協議消息的主要數據結構。在本節中,我們將改善的 TIME 協議客戶端和服務器例子，使用 POJO 代替 ByteBuf。
在 [ChannelHandler](http://netty.io/4.0/api/io/netty/channel/ChannelHandler.html) 使用 POIO 的好處很明顯：通過從ChannelHandler 中提取出 ByteBuf 的代碼，將會使 ChannelHandler的實現變得更加可維護和可重用。在 TIME 客戶端和服務器的例子中，我們讀取的僅僅是一個32位的整形數據，直接使用 ByteBuf 不會是一個主要的問題。然而，你會發現當你需要實現一個真實的協議，分離代碼變得非常的必要。
首先，讓我們定義一個新的類型叫做 UnixTime。
~~~
public class UnixTime {
?
private final long value;
?
public UnixTime() {
this(System.currentTimeMillis() / 1000L + 2208988800L);
}
?
public UnixTime(long value) {
this.value = value;
}
?
public long value() {
return value;
}
?
@Override
public String toString() {
return new Date((value() - 2208988800L) * 1000L).toString();
}
}
~~~
現在我們可以修改下 TimeDecoder 類，返回一個 UnixTime，以替代ByteBuf
~~~
@Override
protected void decode(ChannelHandlerContext ctx, ByteBuf in, List<Object> out) {
if (in.readableBytes() < 4) {
return;
}
?
out.add(new UnixTime(in.readUnsignedInt()));
}
~~~
下面是修改后的解碼器，TimeClientHandler 不再任何的 ByteBuf 代碼了。
~~~
@Override
public void channelRead(ChannelHandlerContext ctx, Object msg) {
UnixTime m = (UnixTime) msg;
System.out.println(m);
ctx.close();
}
~~~
是不是變得更加簡單和優雅了？相同的技術可以被運用到服務端。讓我們修改一下 TimeServerHandler 的代碼。
~~~
@Override
public void channelActive(ChannelHandlerContext ctx) {
ChannelFuture f = ctx.writeAndFlush(new UnixTime());
f.addListener(ChannelFutureListener.CLOSE);
}
~~~
現在,唯一缺少的功能是一個編碼器,是[ChannelOutboundHandler](http://netty.io/4.0/api/io/netty/channel/ChannelOutboundHandler.html)的實現，用來將 UnixTime 對象重新轉化為一個 ByteBuf。這是比編寫一個解碼器簡單得多,因為沒有需要處理的數據包編碼消息時拆分和組裝。
~~~
public class TimeEncoder extends ChannelOutboundHandlerAdapter {
@Override
public void write(ChannelHandlerContext ctx, Object msg, ChannelPromise promise) {
UnixTime m = (UnixTime) msg;
ByteBuf encoded = ctx.alloc().buffer(4);
encoded.writeInt((int)m.value());
ctx.write(encoded, promise); // (1)
}
}
~~~
1.在這幾行代碼里還有幾個重要的事情。第一，通過 [ChannelPromise](http://netty.io/4.0/api/io/netty/channel/ChannelPromise.html)，當編碼后的數據被寫到了通道上 Netty 可以通過這個對象標記是成功還是失敗。第二， 我們不需要調用 cxt.flush()。因為處理器已經單獨分離出了一個方法 void flush(ChannelHandlerContext cxt),如果像自己實現 flush() 方法內容可以自行覆蓋這個方法。
進一步簡化操作，你可以使用 [MessageToByteEncode](http://netty.io/4.0/api/io/netty/handler/codec/MessageToByteEncoder.html):
~~~
public class TimeEncoder extends MessageToByteEncoder<UnixTime> {
@Override
protected void encode(ChannelHandlerContext ctx, UnixTime msg, ByteBuf out) {
out.writeInt((int)msg.value());
}
}
~~~
最后的任務就是在 TimeServerHandler 之前把 TimeEncoder 插入到ChannelPipeline。 但這是不那么重要的工作。
## 關閉你的應用
關閉一個 Netty 應用往往只需要簡單地通過 shutdownGracefully() 方法來關閉你構建的所有的 [EventLoopGroup](http://netty.io/4.0/api/io/netty/channel/EventLoopGroup.html)。當EventLoopGroup 被完全地終止,并且對應的所有 [channel](http://netty.io/4.0/api/io/netty/channel/Channel.html) 都已經被關閉時，Netty 會返回一個[Future](http://netty.io/4.0/api/io/netty/util/concurrent/Future.html)對象來通知你。
## 總結
在這一章節中，我們快速地回顧下如果在熟練掌握 Netty 的情況下編寫出一個健壯能運行的網絡應用程序。在 Netty 接下去的章節中還會有更多更相信的信息。我們也鼓勵你去重新復習下在 [io.netty.example](https://github.com/netty/netty/tree/4.0/example/src/main/java/io/netty/example) 包下的例子。請注意[社區](http://netty.io/community.html)一直在等待你的問題和想法以幫助 Netty 的持續改進，Netty 的文檔也是基于你們的快速反饋上。
*譯者注：翻譯版本的項目源碼見 [https://github.com/waylau/netty-4-user-guide-demos](https://github.com/waylau/netty-4-user-guide-demos)。如對本翻譯有任何建議，可以在[https://github.com/waylau/netty-4-user-guide/issues](https://github.com/waylau/netty-4-user-guide/issues)留言*
