問題

現(xiàn)如今我們使用通用的應(yīng)用程序或者類庫來實(shí)現(xiàn)系統(tǒng)之間地互相訪問，比如我們經(jīng)常使用一個(gè)HTTP客戶端來從web服務(wù)器上獲取信息，或者通過web service來執(zhí)行一個(gè)遠(yuǎn)程的調(diào)用。

然而，有時(shí)候一個(gè)通用的協(xié)議和他的實(shí)現(xiàn)并沒有覆蓋一些場景。比如我們無法使用一個(gè)通用的HTTP服務(wù)器來處理大文件、電子郵件、近實(shí)時(shí)消息比如財(cái)務(wù)信息和多人游戲數(shù)據(jù)。我們需要一個(gè)合適的協(xié)議來處理一些特殊的場景。例如你可以實(shí)現(xiàn)一個(gè)優(yōu)化的Ajax的聊天應(yīng)用、媒體流傳輸或者是大文件傳輸?shù)腍TTP服務(wù)器，你甚至可以自己設(shè)計(jì)和實(shí)現(xiàn)一個(gè)新的協(xié)議來準(zhǔn)確地實(shí)現(xiàn)你的需求。

?

另外不可避免的事情是你不得不處理這些私有協(xié)議來確保和原有系統(tǒng)的互通。這個(gè)例子將會(huì)展示如何快速實(shí)現(xiàn)一個(gè)不影響應(yīng)用程序穩(wěn)定性和性能的協(xié)議。

解決方案

Netty是一個(gè)提供異步事件驅(qū)動(dòng)的網(wǎng)絡(luò)應(yīng)用框架，用以快速開發(fā)高性能、高可靠性的網(wǎng)絡(luò)服務(wù)器和客戶端程序。

換句話說，Netty是一個(gè)NIO框架，使用它可以簡單快速地開發(fā)網(wǎng)絡(luò)應(yīng)用程序，比如客戶端和服務(wù)端的協(xié)議。Netty大大簡化了網(wǎng)絡(luò)程序的開發(fā)過程比如TCP和UDP的 Socket的開發(fā)。

“快速和簡單”并不意味著應(yīng)用程序會(huì)有難維護(hù)和性能低的問題，Netty是一個(gè)精心設(shè)計(jì)的框架，它從許多協(xié)議的實(shí)現(xiàn)中吸收了很多的經(jīng)驗(yàn)比如FTP、SMTP、HTTP、許多二進(jìn)制和基于文本的傳統(tǒng)協(xié)議，Netty在不降低開發(fā)效率、性能、穩(wěn)定性、靈活性情況下，成功地找到了解決方案。

有一些用戶可能已經(jīng)發(fā)現(xiàn)其他的一些網(wǎng)絡(luò)框架也聲稱自己有同樣的優(yōu)勢，所以你可能會(huì)問是Netty和它們的不同之處。答案就是Netty的哲學(xué)設(shè)計(jì)理念。Netty從第一天開始就為用戶提供了用戶體驗(yàn)最好的API以及實(shí)現(xiàn)設(shè)計(jì)。正是因?yàn)?strong>Netty的設(shè)計(jì)理念，才讓我們得以輕松地閱讀本指南并使用Netty。

入門指南

這個(gè)章節(jié)會(huì)介紹Netty核心的結(jié)構(gòu)，并通過一些簡單的例子來幫助你快速入門。當(dāng)你讀完本章節(jié)你馬上就可以用Netty寫出一個(gè)客戶端和服務(wù)端。

如果你在學(xué)習(xí)的時(shí)候喜歡“自頂向下（top-down）”的方法，那你可能需要要從第二章《架構(gòu)概述》開始，然后再回到這里。

開始之前

運(yùn)行本章節(jié)中的兩個(gè)例子最低要求是：Netty的最新版本(Netty5)和JDK1.6及以上。最新的Netty版本在項(xiàng)目下載頁面可以找到。為了下載到正確的JDK版本，請(qǐng)到你喜歡的網(wǎng)站下載。

閱讀本章節(jié)過程中，你可能會(huì)對(duì)相關(guān)類有疑惑，關(guān)于這些類的詳細(xì)的信息請(qǐng)請(qǐng)參考API說明文檔。為了方便，所有文檔中涉及到的類名字都會(huì)被關(guān)聯(lián)到一個(gè)在線的API說明。當(dāng)然如果有任何錯(cuò)誤信息、語法錯(cuò)誤或者你有任何好的建議來改進(jìn)文檔說明，那么請(qǐng)聯(lián)系Netty社區(qū)。

DISCARD服務(wù)(丟棄服務(wù)，指的是會(huì)忽略所有接收的數(shù)據(jù)的一種協(xié)議)

世界上最簡單的協(xié)議不是”Hello,World!”，是DISCARD，他是一種丟棄了所有接受到的數(shù)據(jù)，并不做有任何的響應(yīng)的協(xié)議。

為了實(shí)現(xiàn)DISCARD協(xié)議，你唯一需要做的就是忽略所有收到的數(shù)據(jù)。讓我們從處理器的實(shí)現(xiàn)開始，處理器是由Netty生成用來處理I/O事件的。

package io.netty.example.discard;import io.netty.buffer.ByteBuf;import io.netty.channel.ChannelHandlerContext; import io.netty.channel.ChannelHandlerAdapter;/*** Handles a server-side channel.*/ public class DiscardServerHandler extends ChannelHandlerAdapter { // (1)@Overridepublic void channelRead(ChannelHandlerContext ctx, Object msg) { // (2)// Discard the received data silently.((ByteBuf) msg).release(); // (3)}@Overridepublic void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) { // (4)// Close the connection when an exception is raised.cause.printStackTrace();ctx.close();} }

DisCardServerHandler 繼承自?ChannelHandlerAdapter，這個(gè)類實(shí)現(xiàn)了ChannelHandler接口，ChannelHandler提供了許多事件處理的接口方法，然后你可以覆蓋這些方法。現(xiàn)在僅僅只需要繼承ChannelHandlerAdapter類而不是你自己去實(shí)現(xiàn)接口方法。

這里我們覆蓋了chanelRead()事件處理方法。每當(dāng)從客戶端收到新的數(shù)據(jù)時(shí)，這個(gè)方法會(huì)在收到消息時(shí)被調(diào)用，這個(gè)例子中，收到的消息的類型是ByteBuf

為了實(shí)現(xiàn)DISCARD協(xié)議，處理器不得不忽略所有接受到的消息。ByteBuf是一個(gè)引用計(jì)數(shù)對(duì)象，這個(gè)對(duì)象必須顯示地調(diào)用release()方法來釋放。請(qǐng)記住處理器的職責(zé)是釋放所有傳遞到處理器的引用計(jì)數(shù)對(duì)象。通常，channelRead()方法的實(shí)現(xiàn)就像下面的這段代碼： @Override public void channelRead(ChannelHandlerContext ctx, Object msg) {try {// Do something with msg} finally {ReferenceCountUtil.release(msg);} }

exceptionCaught()事件處理方法是當(dāng)出現(xiàn)Throwable對(duì)象才會(huì)被調(diào)用，即當(dāng)Netty由于IO錯(cuò)誤或者處理器在處理事件時(shí)拋出的異常時(shí)。在大部分情況下，捕獲的異常應(yīng)該被記錄下來并且把關(guān)聯(lián)的channel給關(guān)閉掉。然而這個(gè)方法的處理方式會(huì)在遇到不同異常的情況下有不同的實(shí)現(xiàn)，比如你可能想在關(guān)閉連接之前發(fā)送一個(gè)錯(cuò)誤碼的響應(yīng)消息。

到目前為止一切都還比較順利，我們已經(jīng)實(shí)現(xiàn)了DISCARD服務(wù)的一半功能，剩下的需要編寫一個(gè)main()方法來啟動(dòng)服務(wù)端的DiscardServerHandler。

package io.netty.example.discard;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;/*** Discards any incoming data.*/ 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)@Overridepublic void initChannel(SocketChannel ch) throws Exception {ch.pipeline().addLast(new DiscardServerHandler());}}).option(ChannelOption.SO_BACKLOG, 128) // (5).childOption(ChannelOption.SO_KEEPALIVE, true); // (6)// Bind and start to accept incoming connections.ChannelFuture f = b.bind(port).sync(); // (7)// Wait until the server socket is closed.// In this example, this does not happen, but you can do that to gracefully// shut down your server.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();} }

NioEventLoopGroup?是用來處理I/O操作的多線程事件循環(huán)器，Netty提供了許多不同的EventLoopGroup的實(shí)現(xiàn)用來處理不同傳輸協(xié)議。在這個(gè)例子中我們實(shí)現(xiàn)了一個(gè)服務(wù)端的應(yīng)用，因此會(huì)有2個(gè)NioEventLoopGroup會(huì)被使用。第一個(gè)經(jīng)常被叫做‘boss’，用來接收進(jìn)來的連接。第二個(gè)經(jīng)常被叫做‘worker’，用來處理已經(jīng)被接收的連接，一旦‘boss’接收到連接，就會(huì)把連接信息注冊到‘worker’上。如何知道多少個(gè)線程已經(jīng)被使用，如何映射到已經(jīng)創(chuàng)建的Channels上都需要依賴于EventLoopGroup的實(shí)現(xiàn)，并且可以通過構(gòu)造函數(shù)來配置他們的關(guān)系。

ServerBootstrap?是一個(gè)啟動(dòng)NIO服務(wù)的輔助啟動(dòng)類。你可以在這個(gè)服務(wù)中直接使用Channel，但是這會(huì)是一個(gè)復(fù)雜的處理過程，在很多情況下你并不需要這樣做。

這里我們指定使用NioServerSocketChannel類來舉例說明一個(gè)新的Channel如何接收進(jìn)來的連接。

這里的事件處理類經(jīng)常會(huì)被用來處理一個(gè)最近的已經(jīng)接收的Channel。ChannelInitializer是一個(gè)特殊的處理類，他的目的是幫助使用者配置一個(gè)新的Channel。也許你想通過增加一些處理類比如DiscardServerHandle來配置一個(gè)新的Channel或者其對(duì)應(yīng)的ChannelPipeline來實(shí)現(xiàn)你的網(wǎng)絡(luò)程序。當(dāng)你的程序變的復(fù)雜時(shí)，可能你會(huì)增加更多的處理類到pipline上，然后提取這些匿名類到最頂層的類上。

你可以設(shè)置這里指定的通道實(shí)現(xiàn)的配置參數(shù)。我們正在寫一個(gè)TCP/IP的服務(wù)端，因此我們被允許設(shè)置socket的參數(shù)選項(xiàng)比如tcpNoDelay和keepAlive。請(qǐng)參考ChannelOption和詳細(xì)的ChannelConfig實(shí)現(xiàn)的接口文檔以此可以對(duì)ChannelOptions的有一個(gè)大概的認(rèn)識(shí)。

你關(guān)注過option()和childOption()嗎？option()是提供給NioServerSocketChannel用來接收進(jìn)來的連接。childOption()是提供給由父管道ServerChannel接收到的連接，在這個(gè)例子中也是NioServerSocketChannel。

我們繼續(xù)，剩下的就是綁定端口然后啟動(dòng)服務(wù)。這里我們在機(jī)器上綁定了機(jī)器所有網(wǎng)卡上的8080端口。當(dāng)然現(xiàn)在你可以多次調(diào)用bind()方法(基于不同綁定地址)。

恭喜！你已經(jīng)完成熟練地完成了第一個(gè)基于Netty的服務(wù)端程序。

觀察接收到的數(shù)據(jù)

現(xiàn)在我們已經(jīng)編寫出我們第一個(gè)服務(wù)端，我們需要測試一下他是否真的可以運(yùn)行。最簡單的測試方法是用telnet 命令。例如，你可以在命令行上輸入telnet localhost 8080或者其他類型參數(shù)。

然而我們能說這個(gè)服務(wù)端是正常運(yùn)行了嗎？事實(shí)上我們也不知道因?yàn)樗且粋€(gè)discard服務(wù)，你根本不可能得到任何的響應(yīng)。為了證明他仍然是在工作的，讓我們修改服務(wù)端的程序來打印出他到底接收到了什么。

我們已經(jīng)知道channelRead()方法是在數(shù)據(jù)被接收的時(shí)候調(diào)用。讓我們放一些代碼到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)} }

這個(gè)低效的循環(huán)事實(shí)上可以簡化為:System.out.println(in.toString(io.netty.util.CharsetUtil.US_ASCII))

或者，你可以在這里調(diào)用in.release()。

如果你再次運(yùn)行telnet命令，你將會(huì)看到服務(wù)端打印出了他所接收到的消息。
完整的discard server代碼放在了io.netty.example.discard包下面。

ECHO服務(wù)（響應(yīng)式協(xié)議）

到目前為止，我們雖然接收到了數(shù)據(jù)，但沒有做任何的響應(yīng)。然而一個(gè)服務(wù)端通常會(huì)對(duì)一個(gè)請(qǐng)求作出響應(yīng)。讓我們學(xué)習(xí)怎樣在ECHO協(xié)議的實(shí)現(xiàn)下編寫一個(gè)響應(yīng)消息給客戶端，這個(gè)協(xié)議針對(duì)任何接收的數(shù)據(jù)都會(huì)返回一個(gè)響應(yīng)。

和discard server唯一不同的是把在此之前我們實(shí)現(xiàn)的channelRead()方法，返回所有的數(shù)據(jù)替代打印接收數(shù)據(jù)到控制臺(tái)上的邏輯。因此，需要把channelRead()方法修改如下：

@Override public void channelRead(ChannelHandlerContext ctx, Object msg) {ctx.write(msg); // (1)ctx.flush(); // (2) }

1.?ChannelHandlerContext對(duì)象提供了許多操作，使你能夠觸發(fā)各種各樣的I/O事件和操作。這里我們調(diào)用了write(Object)方法來逐字地把接受到的消息寫入。請(qǐng)注意不同于DISCARD的例子我們并沒有釋放接受到的消息，這是因?yàn)楫?dāng)寫入的時(shí)候Netty已經(jīng)幫我們釋放了。
2. ctx.write(Object)方法不會(huì)使消息寫入到通道上，他被緩沖在了內(nèi)部，你需要調(diào)用ctx.flush()方法來把緩沖區(qū)中數(shù)據(jù)強(qiáng)行輸出。或者你可以用更簡潔的cxt.writeAndFlush(msg)以達(dá)到同樣的目的。

如果你再一次運(yùn)行telnet命令，你會(huì)看到服務(wù)端會(huì)發(fā)回一個(gè)你已經(jīng)發(fā)送的消息。
完整的echo服務(wù)的代碼放在了io.netty.example.echo包下面。

TIME服務(wù)(時(shí)間協(xié)議的服務(wù))

在這個(gè)部分被實(shí)現(xiàn)的協(xié)議是TIME協(xié)議。和之前的例子不同的是在不接受任何請(qǐng)求時(shí)他會(huì)發(fā)送一個(gè)含32位的整數(shù)的消息，并且一旦消息發(fā)送就會(huì)立即關(guān)閉連接。在這個(gè)例子中，你會(huì)學(xué)習(xí)到如何構(gòu)建和發(fā)送一個(gè)消息，然后在完成時(shí)主動(dòng)關(guān)閉連接。

因?yàn)槲覀儗?huì)忽略任何接收到的數(shù)據(jù)，而只是在連接被創(chuàng)建發(fā)送一個(gè)消息，所以這次我們不能使用channelRead()方法了，代替他的是，我們需要覆蓋channelActive()方法，下面的就是實(shí)現(xiàn)的內(nèi)容：

package io.netty.example.time;public class TimeServerHandler extends ChannelHandlerAdapter {@Overridepublic 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() {@Overridepublic void operationComplete(ChannelFuture future) {assert f == future;ctx.close();}}); // (4)}@Overridepublic void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) {cause.printStackTrace();ctx.close();} }

channelActive()方法將會(huì)在連接被建立并且準(zhǔn)備進(jìn)行通信時(shí)被調(diào)用。因此讓我們在這個(gè)方法里完成一個(gè)代表當(dāng)前時(shí)間的32位整數(shù)消息的構(gòu)建工作。

為了發(fā)送一個(gè)新的消息，我們需要分配一個(gè)包含這個(gè)消息的新的緩沖。因?yàn)槲覀冃枰獙懭胍粋€(gè)32位的整數(shù)，因此我們需要一個(gè)至少有4個(gè)字節(jié)的ByteBuf。通過ChannelHandlerContext.alloc()得到一個(gè)當(dāng)前的ByteBufAllocator，然后分配一個(gè)新的緩沖。

和往常一樣我們需要編寫一個(gè)構(gòu)建好的消息。但是等一等，flip在哪？難道我們使用NIO發(fā)送消息時(shí)不是調(diào)用java.nio.ByteBuffer.flip()嗎？ByteBuf之所以沒有這個(gè)方法因?yàn)橛袃蓚€(gè)指針，一個(gè)對(duì)應(yīng)讀操作一個(gè)對(duì)應(yīng)寫操作。當(dāng)你向ByteBuf里寫入數(shù)據(jù)的時(shí)候?qū)懼羔樀乃饕蜁?huì)增加，同時(shí)讀指針的索引沒有變化。讀指針?biāo)饕蛯懼羔標(biāo)饕謩e代表了消息的開始和結(jié)束。比較起來，NIO緩沖并沒有提供一種簡潔的方式來計(jì)算出消息內(nèi)容的開始和結(jié)尾，除非你調(diào)用flip方法。當(dāng)你忘記調(diào)用flip方法而引起沒有數(shù)據(jù)或者錯(cuò)誤數(shù)據(jù)被發(fā)送時(shí)，你會(huì)陷入困境。這樣的一個(gè)錯(cuò)誤不會(huì)發(fā)生在Netty上，因?yàn)槲覀儗?duì)于不同的操作類型有不同的指針。你會(huì)發(fā)現(xiàn)這樣的使用方法會(huì)讓你過程變得更加的容易，因?yàn)槟阋呀?jīng)習(xí)慣一種沒有使用flip的方式。另外一個(gè)點(diǎn)需要注意的是ChannelHandlerContext.write()(和writeAndFlush())方法會(huì)返回一個(gè)ChannelFuture對(duì)象，一個(gè)ChannelFuture代表了一個(gè)還沒有發(fā)生的I/O操作。這意味著任何一個(gè)請(qǐng)求操作都不會(huì)馬上被執(zhí)行，因?yàn)樵?strong>Netty里所有的操作都是異步的。舉個(gè)例子下面的代碼中在消息被發(fā)送之前可能會(huì)先關(guān)閉連接。 Channel ch = ...; ch.writeAndFlush(message); ch.close();

因此你需要在write()方法返回的ChannelFuture完成后調(diào)用close()方法，然后當(dāng)他的寫操作已經(jīng)完成他會(huì)通知他的監(jiān)聽者。請(qǐng)注意,close()方法也可能不會(huì)立馬關(guān)閉，他也會(huì)返回一個(gè)ChannelFuture。

當(dāng)一個(gè)寫請(qǐng)求已經(jīng)完成是如何通知到我們？這個(gè)只需要簡單地在返回的ChannelFuture上增加一個(gè)ChannelFutureListener。這里我們構(gòu)建了一個(gè)匿名的ChannelFutureListener類用來在操作完成時(shí)關(guān)閉Channel。或者，你可以使用簡單的預(yù)定義監(jiān)聽器代碼: f.addListener(ChannelFutureListener.CLOSE);

為了測試我們的time服務(wù)如我們期望的一樣工作，你可以使用UNIX的rdate命令

$ rdate -o <port> -p <host>

Port是你在main()函數(shù)中指定的端口，host使用locahost就可以了。

Time客戶端

不像DISCARD和ECHO的服務(wù)端，對(duì)于TIME協(xié)議我們需要一個(gè)客戶端因?yàn)槿藗儾荒馨岩粋€(gè)32位的二進(jìn)制數(shù)據(jù)翻譯成一個(gè)日期或者日歷。在這一部分，我們將會(huì)討論如何確保服務(wù)端是正常工作的，并且學(xué)習(xí)怎樣用Netty編寫一個(gè)客戶端。

在Netty中,編寫服務(wù)端和客戶端最大的并且唯一不同的使用了不同的BootStrap和Channel的實(shí)現(xiàn)。請(qǐng)看一下下面的代碼：

package io.netty.example.time;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&lt;SocketChannel&gt;() {@Overridepublic void initChannel(SocketChannel ch) throws Exception {ch.pipeline().addLast(new TimeClientHandler());}});// Start the client.ChannelFuture f = b.connect(host, port).sync(); // (5)// Wait until the connection is closed.f.channel().closeFuture().sync();} finally {workerGroup.shutdownGracefully();}} }

BootStrap和ServerBootstrap類似,不過他是對(duì)非服務(wù)端的channel而言，比如客戶端或者無連接傳輸模式的channel。

如果你只指定了一個(gè)EventLoopGroup，那他就會(huì)即作為一個(gè)‘boss’線程，也會(huì)作為一個(gè)‘workder’線程，盡管客戶端不需要使用到‘boss’線程。

代替NioServerSocketChannel的是NioSocketChannel,這個(gè)類在客戶端channel被創(chuàng)建時(shí)使用。

不像在使用ServerBootstrap時(shí)需要用childOption()方法，因?yàn)榭蛻舳说腟ocketChannel沒有父channel的概念。

我們用connect()方法代替了bind()方法。

正如你看到的，他和服務(wù)端的代碼是不一樣的。ChannelHandler是如何實(shí)現(xiàn)的?他應(yīng)該從服務(wù)端接受一個(gè)32位的整數(shù)消息，把他翻譯成人們能讀懂的格式，并打印翻譯好的時(shí)間，最后關(guān)閉連接:

package io.netty.example.time;import java.util.Date;public class TimeClientHandler extends ChannelHandlerAdapter {@Overridepublic 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();}}@Overridepublic void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) {cause.printStackTrace();ctx.close();} }

在TCP/IP中，NETTY會(huì)把讀到的數(shù)據(jù)放到ByteBuf的數(shù)據(jù)結(jié)構(gòu)中。

這樣看起來非常簡單，并且和服務(wù)端的那個(gè)例子的代碼也相差不多。然而，處理器有時(shí)候會(huì)因?yàn)閽伋鯥ndexOutOfBoundsException而拒絕工作。在下個(gè)部分我們會(huì)討論為什么會(huì)發(fā)生這種情況。

流數(shù)據(jù)的傳輸處理

一個(gè)小的Socket Buffer問題

在基于流的傳輸里比如TCP/IP，接收到的數(shù)據(jù)會(huì)先被存儲(chǔ)到一個(gè)socket接收緩沖里。不幸的是，基于流的傳輸并不是一個(gè)數(shù)據(jù)包隊(duì)列，而是一個(gè)字節(jié)隊(duì)列。即使你發(fā)送了2個(gè)獨(dú)立的數(shù)據(jù)包，操作系統(tǒng)也不會(huì)作為2個(gè)消息處理而僅僅是作為一連串的字節(jié)而言。因此這是不能保證你遠(yuǎn)程寫入的數(shù)據(jù)就會(huì)準(zhǔn)確地讀取。舉個(gè)例子，讓我們假設(shè)操作系統(tǒng)的TCP/TP協(xié)議棧已經(jīng)接收了3個(gè)數(shù)據(jù)包：

由于基于流傳輸?shù)膮f(xié)議的這種普通的性質(zhì)，在你的應(yīng)用程序里讀取數(shù)據(jù)的時(shí)候會(huì)有很高的可能性被分成下面的片段。

因此，一個(gè)接收方不管他是客戶端還是服務(wù)端，都應(yīng)該把接收到的數(shù)據(jù)整理成一個(gè)或者多個(gè)更有意思并且能夠讓程序的業(yè)務(wù)邏輯更好理解的數(shù)據(jù)。在上面的例子中，接收到的數(shù)據(jù)應(yīng)該被構(gòu)造成下面的格式：

第一個(gè)解決方案

現(xiàn)在讓我們回到TIME客戶端的例子上。這里我們遇到了同樣的問題，一個(gè)32字節(jié)數(shù)據(jù)是非常小的數(shù)據(jù)量，他并不見得會(huì)被經(jīng)常拆分到到不同的數(shù)據(jù)段內(nèi)。然而，問題是他確實(shí)可能會(huì)被拆分到不同的數(shù)據(jù)段內(nèi)，并且拆分的可能性會(huì)隨著通信量的增加而增加。

最簡單的方案是構(gòu)造一個(gè)內(nèi)部的可積累的緩沖，直到4個(gè)字節(jié)全部接收到了內(nèi)部緩沖。下面的代碼修改了TimeClientHandler的實(shí)現(xiàn)類修復(fù)了這個(gè)問題

package io.netty.example.time;import java.util.Date;public class TimeClientHandler extends ChannelHandlerAdapter {private ByteBuf buf;@Overridepublic void handlerAdded(ChannelHandlerContext ctx) {buf = ctx.alloc().buffer(4); // (1)}@Overridepublic void handlerRemoved(ChannelHandlerContext ctx) {buf.release(); // (1)buf = null;}@Overridepublic void channelRead(ChannelHandlerContext ctx, Object msg) {ByteBuf m = (ByteBuf) msg;buf.writeBytes(m); // (2)m.release();if (buf.readableBytes() &gt;= 4) { // (3)long currentTimeMillis = (buf.readInt() - 2208988800L) * 1000L;System.out.println(new Date(currentTimeMillis));ctx.close();}}@Overridepublic void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) {cause.printStackTrace();ctx.close();} }

ChannelHandler有2個(gè)生命周期的監(jiān)聽方法：handlerAdded()和handlerRemoved()。你可以完成任意初始化任務(wù)只要他不會(huì)被阻塞很長的時(shí)間。

首先，所有接收的數(shù)據(jù)都應(yīng)該被累積在buf變量里。

然后，處理器必須檢查buf變量是否有足夠的數(shù)據(jù)，在這個(gè)例子中是4個(gè)字節(jié)，然后處理實(shí)際的業(yè)務(wù)邏輯。否則，Netty會(huì)重復(fù)調(diào)用channelRead()當(dāng)有更多數(shù)據(jù)到達(dá)直到4個(gè)字節(jié)的數(shù)據(jù)被積累。

第二個(gè)解決方案

盡管第一個(gè)解決方案已經(jīng)解決了Time客戶端的問題了，但是修改后的處理器看起來不那么的簡潔，想象一下如果由多個(gè)字段比如可變長度的字段組成的更為復(fù)雜的協(xié)議時(shí)，你的ChannelHandler的實(shí)現(xiàn)將很快地變得難以維護(hù)。

正如你所知的，你可以增加多個(gè)ChannelHandler到ChannelPipeline?,因此你可以把一整個(gè)ChannelHandler拆分成多個(gè)模塊以減少應(yīng)用的復(fù)雜程度，比如你可以把TimeClientHandler拆分成2個(gè)處理器：

• TimeDecoder處理數(shù)據(jù)拆分的問題
• TimeClientHandler原始版本的實(shí)現(xiàn)

幸運(yùn)地是，Netty提供了一個(gè)可擴(kuò)展的類，幫你完成TimeDecoder的開發(fā)。

package io.netty.example.time;public class TimeDecoder extends ByteToMessageDecoder { // (1)@Overrideprotected void decode(ChannelHandlerContext ctx, ByteBuf in, List&lt;Object&gt; out) { // (2)if (in.readableBytes() &lt; 4) {return; // (3)}out.add(in.readBytes(4)); // (4)} }

ByteToMessageDecoder是ChannelHandler的一個(gè)實(shí)現(xiàn)類，他可以在處理數(shù)據(jù)拆分的問題上變得很簡單。

每當(dāng)有新數(shù)據(jù)接收的時(shí)候，ByteToMessageDecoder都會(huì)調(diào)用decode()方法來處理內(nèi)部的那個(gè)累積緩沖。

Decode()方法可以決定當(dāng)累積緩沖里沒有足夠數(shù)據(jù)時(shí)可以往out對(duì)象里放任意數(shù)據(jù)。當(dāng)有更多的數(shù)據(jù)被接收了ByteToMessageDecoder會(huì)再一次調(diào)用decode()方法。

如果在decode()方法里增加了一個(gè)對(duì)象到out對(duì)象里，這意味著解碼器解碼消息成功。ByteToMessageDecoder將會(huì)丟棄在累積緩沖里已經(jīng)被讀過的數(shù)據(jù)。請(qǐng)記得你不需要對(duì)多條消息調(diào)用decode()，ByteToMessageDecoder會(huì)持續(xù)調(diào)用decode()直到不放任何數(shù)據(jù)到out里。

現(xiàn)在我們有另外一個(gè)處理器插入到ChannelPipeline里，我們應(yīng)該在TimeClient里修改ChannelInitializer?的實(shí)現(xiàn)：

b.handler(new ChannelInitializer&lt;SocketChannel&gt;() {@Overridepublic void initChannel(SocketChannel ch) throws Exception {ch.pipeline().addLast(new TimeDecoder(), new TimeClientHandler());} });

如果你是一個(gè)大膽的人，你可能會(huì)嘗試使用更簡單的解碼類ReplayingDecoder。不過你還是需要參考一下API文檔來獲取更多的信息。

public class TimeDecoder extends ReplayingDecoder { @Override protected void decode( ChannelHandlerContext ctx, ByteBuf in, List<object width="300" height="150">out) {out.add(in.readBytes(4));}}

此外，Netty還提供了更多可以直接拿來用的解碼器使你可以更簡單地實(shí)現(xiàn)更多的協(xié)議，幫助你避免開發(fā)一個(gè)難以維護(hù)的處理器實(shí)現(xiàn)。請(qǐng)參考下面的包以獲取更多更詳細(xì)的例子：

• 對(duì)于二進(jìn)制協(xié)議請(qǐng)看io.netty.example.factorial
• 對(duì)于基于文本協(xié)議請(qǐng)看io.netty.example.telnet

用POJO代替ByteBuf

我們已經(jīng)討論了所有的例子，到目前為止一個(gè)消息的消息都是使用ByteBuf作為一個(gè)基本的數(shù)據(jù)結(jié)構(gòu)。在這一部分，我們會(huì)改進(jìn)TIME協(xié)議的客戶端和服務(wù)端的例子，用POJO替代ByteBuf。在你的ChannelHandlerS中使用POJO優(yōu)勢是比較明顯的。通過從ChannelHandler中提取出ByteBuf的代碼，將會(huì)使ChannelHandler的實(shí)現(xiàn)變得更加可維護(hù)和可重用。在TIME客戶端和服務(wù)端的例子中，我們讀取的僅僅是一個(gè)32位的整形數(shù)據(jù)，直接使用ByteBuf不會(huì)是一個(gè)主要的問題。然后，你會(huì)發(fā)現(xiàn)當(dāng)你需要實(shí)現(xiàn)一個(gè)真實(shí)的協(xié)議，分離代碼變得非常的必要。首先，讓我們定義一個(gè)新的類型叫做UnixTime。

package io.netty.example.time;import java.util.Date;public class UnixTime {private final int value;public UnixTime() {this((int) (System.currentTimeMillis() / 1000L + 2208988800L));}public UnixTime(int value) {this.value = value;}public int value() {return value;}@Overridepublic String toString() {return new Date((value() - 2208988800L) * 1000L).toString();} }

現(xiàn)在我們可以修改下TimeDecoder類，返回一個(gè)UnixTime，以替代ByteBuf

@Override protected void decode(ChannelHandlerContext ctx, ByteBuf in, List<Object> out) {if (in.readableBytes() < 4) {return;}out.add(new UnixTime(in.readInt())); }

下面是修改后的解碼器，TimeClientHandler不再有任何的ByteBuf代碼了。

@Override public void channelRead(ChannelHandlerContext ctx, Object msg) {UnixTime m = (UnixTime) msg;System.out.println(m);ctx.close(); }

是不是變得更加簡單和優(yōu)雅了？相同的技術(shù)可以被運(yùn)用到服務(wù)端。讓我們修改一下TimeServerHandler的代碼。

@Override public void channelActive(ChannelHandlerContext ctx) {ChannelFuture f = ctx.writeAndFlush(new UnixTime());f.addListener(ChannelFutureListener.CLOSE); }

現(xiàn)在，僅僅需要修改的是ChannelHandler的實(shí)現(xiàn)，這里需要把UnixTime對(duì)象重新轉(zhuǎn)化為一個(gè)ByteBuf。不過這已經(jīng)是非常簡單了，因?yàn)楫?dāng)你對(duì)一個(gè)消息編碼的時(shí)候，你不需要再處理拆包和組裝的過程。

package io.netty.example.time;public class TimeEncoder extends ChannelHandlerAdapter {@Overridepublic void write(ChannelHandlerContext ctx, Object msg, ChannelPromise promise) {UnixTime m = (UnixTime) msg;ByteBuf encoded = ctx.alloc().buffer(4);encoded.writeInt(m.value());ctx.write(encoded, promise); // (1)} }

在這幾行代碼里還有幾個(gè)重要的事情。第一， 通過ChannelPromise，當(dāng)編碼后的數(shù)據(jù)被寫到了通道上Netty可以通過這個(gè)對(duì)象標(biāo)記是成功還是失敗。第二， 我們不需要調(diào)用cxt.flush()。因?yàn)樘幚砥饕呀?jīng)單獨(dú)分離出了一個(gè)方法void flush(ChannelHandlerContext cxt),如果像自己實(shí)現(xiàn)flush方法內(nèi)容可以自行覆蓋這個(gè)方法。

進(jìn)一步簡化操作，你可以使用MessageToByteEncode:

public class TimeEncoder extends MessageToByteEncoder<UnixTime> {@Overrideprotected void encode(ChannelHandlerContext ctx, UnixTime msg, ByteBuf out) {out.writeInt(msg.value());} }

最后的任務(wù)就是在TimeServerHandler之前把TimeEncoder插入到ChannelPipeline。但這是不那么重要的工作。

關(guān)閉你的應(yīng)用

關(guān)閉一個(gè)Netty應(yīng)用往往只需要簡單地通過shutdownGracefully()方法來關(guān)閉你構(gòu)建的所有的NioEventLoopGroupS.當(dāng)EventLoopGroup被完全地終止,并且對(duì)應(yīng)的所有channels都已經(jīng)被關(guān)閉時(shí)，Netty會(huì)返回一個(gè)Future對(duì)象。

概述

在這一章節(jié)中，我們會(huì)快速地回顧下如果在熟練掌握Netty的情況下編寫出一個(gè)健壯能運(yùn)行的網(wǎng)絡(luò)應(yīng)用程序。在Netty接下去的章節(jié)中還會(huì)有更多更相信的信息。我們也鼓勵(lì)你去重新復(fù)習(xí)下在io.netty.example包下的例子。請(qǐng)注意社區(qū)一直在等待你的問題和想法以幫助Netty的持續(xù)改進(jìn)，Netty的文檔也是基于你們的快速反饋上。

總結(jié)

以上是生活随笔為你收集整理的Netty 5用户指南的全部內(nèi)容，希望文章能夠幫你解決所遇到的問題。

如果覺得生活随笔網(wǎng)站內(nèi)容還不錯(cuò)，歡迎將生活随笔推薦給好友。