Netty 5使用者指南

阿新 • • 發佈：2018-12-22

前言

問題

現如今我們使用通用的應用程式或者類庫來實現系統之間地互相訪問，比如我們經常使用一個HTTP客戶端來從web伺服器上獲取資訊，或者通過web service來執行一個遠端的呼叫。

然而，有時候一個通用的協議和他的實現並沒有覆蓋一些場景。比如我們無法使用一個通用的HTTP伺服器來處理大檔案、電子郵件、近實時訊息比如財務資訊和多人遊戲資料。我們需要一個合適的協議來處理一些特殊的場景。例如你可以實現一個優化的Ajax的聊天應用、媒體流傳輸或者是大檔案傳輸的HTTP伺服器，你甚至可以自己設計和實現一個新的協議來準確地實現你的需求。

另外不可避免的事情是你不得不處理這些私有協議來確保和原有系統的互通。這個例子將會展示如何快速實現一個不影響應用程式穩定性和效能的協議。

解決方案

Netty是一個提供非同步事件驅動的網路應用框架，用以快速開發高效能、高可靠性的網路伺服器和客戶端程式。

換句話說，Netty是一個NIO框架，使用它可以簡單快速地開發網路應用程式，比如客戶端和服務端的協議。Netty大大簡化了網路程式的開發過程比如TCP和UDP的 Socket的開發。

“快速和簡單”並不意味著應用程式會有難維護和效能低的問題，Netty是一個精心設計的框架，它從許多協議的實現中吸收了很多的經驗比如FTP、SMTP、HTTP、許多二進位制和基於文字的傳統協議，Netty在不降低開發效率、效能、穩定性、靈活性情況下，成功地找到了解決方案。

有一些使用者可能已經發現其他的一些網路框架也聲稱自己有同樣的優勢，所以你可能會問是Netty和它們的不同之處。答案就是Netty的哲學設計理念。Netty從第一天開始就為使用者提供了使用者體驗最好的API以及實現設計。正是因為Netty的設計理念，才讓我們得以輕鬆地閱讀本指南並使用Netty。

入門指南

這個章節會介紹Netty核心的結構，並通過一些簡單的例子來幫助你快速入門。當你讀完本章節你馬上就可以用Netty寫出一個客戶端和服務端。

如果你在學習的時候喜歡“自頂向下（top-down）”的方法，那你可能需要要從第二章《架構概述》開始，然後再回到這裡。

開始之前

執行本章節中的兩個例子最低要求是：Netty的最新版本(Netty5)和JDK1.6及以上。最新的Netty版本在專案下載頁面可以找到。為了下載到正確的JDK版本，請到你喜歡的網站下載。

閱讀本章節過程中，你可能會對相關類有疑惑，關於這些類的詳細的資訊請請參考API說明文件。為了方便，所有文件中涉及到的類名字都會被關聯到一個線上的API說明。當然如果有任何錯誤資訊、語法錯誤或者你有任何好的建議來改進文件說明，那麼請聯絡N

etty社群。

DISCARD服務(丟棄服務，指的是會忽略所有接收的資料的一種協議)

世界上最簡單的協議不是”Hello,World!”，是DISCARD，他是一種丟棄了所有接受到的資料，並不做有任何的響應的協議。

為了實現DISCARD協議，你唯一需要做的就是忽略所有收到的資料。讓我們從處理器的實現開始，處理器是由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)

    @Override
    public void channelRead(ChannelHandlerContext ctx, Object msg) { // (2)
        // Discard the received data silently.
        ((ByteBuf) msg).release(); // (3)
    }

    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) { // (4)
        // Close the connection when an exception is raised.
        cause.printStackTrace();
        ctx.close();
    }
}

這裡我們覆蓋了chanelRead()事件處理方法。每當從客戶端收到新的資料時，這個方法會在收到訊息時被呼叫，這個例子中，收到的訊息的型別是ByteBuf
為了實現DISCARD協議，處理器不得不忽略所有接受到的訊息。ByteBuf是一個引用計數物件，這個物件必須顯示地呼叫release()方法來釋放。請記住處理器的職責是釋放所有傳遞到處理器的引用計數物件。通常，channelRead()方法的實現就像下面的這段程式碼：
```
 @Override
public void channelRead(ChannelHandlerContext ctx, Object msg) {
    try {
        // Do something with msg
    } finally {
        ReferenceCountUtil.release(msg);
    }
}
```
exceptionCaught()事件處理方法是當出現Throwable物件才會被呼叫，即當Netty由於IO錯誤或者處理器在處理事件時丟擲的異常時。在大部分情況下，捕獲的異常應該被記錄下來並且把關聯的channel給關閉掉。然而這個方法的處理方式會在遇到不同異常的情況下有不同的實現，比如你可能想在關閉連線之前傳送一個錯誤碼的響應訊息。

到目前為止一切都還比較順利，我們已經實現了DISCARD服務的一半功能，剩下的需要編寫一個main()方法來啟動服務端的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&lt;SocketChannel&gt;() { // (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)

            // 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 &gt; 0) {
            port = Integer.parseInt(args[0]);
        } else {
            port = 8080;
        }
        new DiscardServer(port).run();
    }
}

NioEventLoopGroup 是用來處理I/O操作的多執行緒事件迴圈器，Netty提供了許多不同的EventLoopGroup的實現用來處理不同傳輸協議。在這個例子中我們實現了一個服務端的應用，因此會有2個NioEventLoopGroup會被使用。第一個經常被叫做‘boss’，用來接收進來的連線。第二個經常被叫做‘worker’，用來處理已經被接收的連線，一旦‘boss’接收到連線，就會把連線資訊註冊到‘worker’上。如何知道多少個執行緒已經被使用，如何對映到已經建立的Channels上都需要依賴於EventLoopGroup的實現，並且可以通過建構函式來配置他們的關係。
ServerBootstrap 是一個啟動NIO服務的輔助啟動類。你可以在這個服務中直接使用Channel，但是這會是一個複雜的處理過程，在很多情況下你並不需要這樣做。
這裡的事件處理類經常會被用來處理一個最近的已經接收的Channel。ChannelInitializer是一個特殊的處理類，他的目的是幫助使用者配置一個新的Channel。也許你想通過增加一些處理類比如DiscardServerHandle來配置一個新的Channel或者其對應的ChannelPipeline來實現你的網路程式。當你的程式變的複雜時，可能你會增加更多的處理類到pipline上，然後提取這些匿名類到最頂層的類上。
你可以設定這裡指定的通道實現的配置引數。我們正在寫一個TCP/IP的服務端，因此我們被允許設定socket的引數選項比如tcpNoDelay和keepAlive。請參考ChannelOption和詳細的ChannelConfig實現的介面文件以此可以對ChannelOptions的有一個大概的認識。
你關注過option()和childOption()嗎？option()是提供給NioServerSocketChannel用來接收進來的連線。childOption()是提供給由父管道ServerChannel接收到的連線，在這個例子中也是NioServerSocketChannel。
我們繼續，剩下的就是繫結埠然後啟動服務。這裡我們在機器上綁定了機器所有網絡卡上的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)
    }
}

這個低效的迴圈事實上可以簡化為:System.out.println(in.toString(io.netty.util.CharsetUtil.US_ASCII))
或者，你可以在這裡呼叫in.release()。

如果你再次執行telnet命令，你將會看到服務端打印出了他所接收到的訊息。
完整的discard server程式碼放在了io.netty.example.discard包下面。

ECHO服務（響應式協議）

到目前為止，我們雖然接收到了資料，但沒有做任何的響應。然而一個服務端通常會對一個請求作出響應。讓我們學習怎樣在ECHO協議的實現下編寫一個響應訊息給客戶端，這個協議針對任何接收的資料都會返回一個響應。

和discard server唯一不同的是把在此之前我們實現的channelRead()方法，返回所有的資料替代列印接收資料到控制檯上的邏輯。因此，需要把channelRead()方法修改如下：

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

1. ChannelHandlerContext物件提供了許多操作，使你能夠觸發各種各樣的I/O事件和操作。這裡我們呼叫了write(Object)方法來逐字地把接受到的訊息寫入。請注意不同於DISCARD的例子我們並沒有釋放接受到的訊息，這是因為當寫入的時候Netty已經幫我們釋放了。
2. ctx.write(Object)方法不會使訊息寫入到通道上，他被緩衝在了內部，你需要呼叫ctx.flush()方法來把緩衝區中資料強行輸出。或者你可以用更簡潔的cxt.writeAndFlush(msg)以達到同樣的目的。

如果你再一次執行telnet命令，你會看到服務端會發回一個你已經發送的訊息。
完整的echo服務的程式碼放在了io.netty.example.echo包下面。

TIME服務(時間協議的服務)

在這個部分被實現的協議是TIME協議。和之前的例子不同的是在不接受任何請求時他會發送一個含32位的整數的訊息，並且一旦訊息傳送就會立即關閉連線。在這個例子中，你會學習到如何構建和傳送一個訊息，然後在完成時主動關閉連線。

因為我們將會忽略任何接收到的資料，而只是在連線被建立傳送一個訊息，所以這次我們不能使用channelRead()方法了，代替他的是，我們需要覆蓋channelActive()方法，下面的就是實現的內容：

package io.netty.example.time;

public class TimeServerHandler extends ChannelHandlerAdapter {

    @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();
    }
}

channelActive()方法將會在連線被建立並且準備進行通訊時被呼叫。因此讓我們在這個方法裡完成一個代表當前時間的32位整數訊息的構建工作。
為了傳送一個新的訊息，我們需要分配一個包含這個訊息的新的緩衝。因為我們需要寫入一個32位的整數，因此我們需要一個至少有4個位元組的ByteBuf。通過ChannelHandlerContext.alloc()得到一個當前的ByteBufAllocator，然後分配一個新的緩衝。
和往常一樣我們需要編寫一個構建好的訊息。但是等一等，flip在哪？難道我們使用NIO傳送訊息時不是呼叫java.nio.ByteBuffer.flip()嗎？ByteBuf之所以沒有這個方法因為有兩個指標，一個對應讀操作一個對應寫操作。當你向ByteBuf裡寫入資料的時候寫指標的索引就會增加，同時讀指標的索引沒有變化。讀指標索引和寫指標索引分別代表了訊息的開始和結束。比較起來，NIO緩衝並沒有提供一種簡潔的方式來計算出訊息內容的開始和結尾，除非你呼叫flip方法。當你忘記呼叫flip方法而引起沒有資料或者錯誤資料被髮送時，你會陷入困境。這樣的一個錯誤不會發生在Netty上，因為我們對於不同的操作型別有不同的指標。你會發現這樣的使用方法會讓你過程變得更加的容易，因為你已經習慣一種沒有使用flip的方式。另外一個點需要注意的是ChannelHandlerContext.write()(和writeAndFlush())方法會返回一個ChannelFuture物件，一個ChannelFuture代表了一個還沒有發生的I/O操作。這意味著任何一個請求操作都不會馬上被執行，因為在Netty裡所有的操作都是非同步的。舉個例子下面的程式碼中在訊息被髮送之前可能會先關閉連線。
```
Channel ch = ...;
ch.writeAndFlush(message);
ch.close();
```
因此你需要在write()方法返回的ChannelFuture完成後呼叫close()方法，然後當他的寫操作已經完成他會通知他的監聽者。請注意,close()方法也可能不會立馬關閉，他也會返回一個ChannelFuture。
當一個寫請求已經完成是如何通知到我們？這個只需要簡單地在返回的ChannelFuture上增加一個ChannelFutureListener。這裡我們構建了一個匿名的ChannelFutureListener類用來在操作完成時關閉Channel。或者，你可以使用簡單的預定義監聽器程式碼:
```
f.addListener(ChannelFutureListener.CLOSE);
```

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

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

Port是你在main()函式中指定的埠，host使用locahost就可以了。

Time客戶端

不像DISCARD和ECHO的服務端，對於TIME協議我們需要一個客戶端因為人們不能把一個32位的二進位制資料翻譯成一個日期或者日曆。在這一部分，我們將會討論如何確保服務端是正常工作的，並且學習怎樣用Netty編寫一個客戶端。

在Netty中,編寫服務端和客戶端最大的並且唯一不同的使用了不同的BootStrap和Channel的實現。請看一下下面的程式碼：

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;() {
                @Override
                public 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類似,不過他是對非服務端的channel而言，比如客戶端或者無連線傳輸模式的channel。
如果你只指定了一個EventLoopGroup，那他就會即作為一個‘boss’執行緒，也會作為一個‘workder’執行緒，儘管客戶端不需要使用到‘boss’執行緒。
不像在使用ServerBootstrap時需要用childOption()方法，因為客戶端的SocketChannel沒有父channel的概念。
我們用connect()方法代替了bind()方法。

正如你看到的，他和服務端的程式碼是不一樣的。ChannelHandler是如何實現的?他應該從服務端接受一個32位的整數訊息，把他翻譯成人們能讀懂的格式，並列印翻譯好的時間，最後關閉連線:

package io.netty.example.time;

import java.util.Date;

public class TimeClientHandler extends ChannelHandlerAdapter {
    @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();
    }
}

在TCP/IP中，NETTY會把讀到的資料放到ByteBuf的資料結構中。

這樣看起來非常簡單，並且和服務端的那個例子的程式碼也相差不多。然而，處理器有時候會因為丟擲IndexOutOfBoundsException而拒絕工作。在下個部分我們會討論為什麼會發生這種情況。

流資料的傳輸處理

一個小的Socket Buffer問題

在基於流的傳輸裡比如TCP/IP，接收到的資料會先被儲存到一個socket接收緩衝裡。不幸的是，基於流的傳輸並不是一個數據包佇列，而是一個位元組佇列。即使你傳送了2個獨立的資料包，作業系統也不會作為2個訊息處理而僅僅是作為一連串的位元組而言。因此這是不能保證你遠端寫入的資料就會準確地讀取。舉個例子，讓我們假設作業系統的TCP/TP協議棧已經接收了3個數據包：

由於基於流傳輸的協議的這種普通的性質，在你的應用程式裡讀取資料的時候會有很高的可能性被分成下面的片段。

因此，一個接收方不管他是客戶端還是服務端，都應該把接收到的資料整理成一個或者多個更有意思並且能夠讓程式的業務邏輯更好理解的資料。在上面的例子中，接收到的資料應該被構造成下面的格式：

第一個解決方案

現在讓我們回到TIME客戶端的例子上。這裡我們遇到了同樣的問題，一個32位元組資料是非常小的資料量，他並不見得會被經常拆分到到不同的資料段內。然而，問題是他確實可能會被拆分到不同的資料段內，並且拆分的可能性會隨著通訊量的增加而增加。

最簡單的方案是構造一個內部的可積累的緩衝，直到4個位元組全部接收到了內部緩衝。下面的程式碼修改了TimeClientHandler的實現類修復了這個問題

package io.netty.example.time;

import java.util.Date;

public class TimeClientHandler extends ChannelHandlerAdapter {
    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() &gt;= 4) { // (3)
            long currentTimeMillis = (buf.readInt() - 2208988800L) * 1000L;
            System.out.println(new Date(currentTimeMillis));
            ctx.close();
        }
    }

    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) {
        cause.printStackTrace();
        ctx.close();
    }
}

ChannelHandler有2個生命週期的監聽方法：handlerAdded()和handlerRemoved()。你可以完成任意初始化任務只要他不會被阻塞很長的時間。
首先，所有接收的資料都應該被累積在buf變數裡。
然後，處理器必須檢查buf變數是否有足夠的資料，在這個例子中是4個位元組，然後處理實際的業務邏輯。否則，Netty會重複呼叫channelRead()當有更多資料到達直到4個位元組的資料被積累。

第二個解決方案

儘管第一個解決方案已經解決了Time客戶端的問題了，但是修改後的處理器看起來不那麼的簡潔，想象一下如果由多個欄位比如可變長度的欄位組成的更為複雜的協議時，你的ChannelHandler的實現將很快地變得難以維護。

正如你所知的，你可以增加多個ChannelHandler到ChannelPipeline ,因此你可以把一整個ChannelHandler拆分成多個模組以減少應用的複雜程度，比如你可以把TimeClientHandler拆分成2個處理器：

TimeDecoder處理資料拆分的問題
TimeClientHandler原始版本的實現

幸運地是，Netty提供了一個可擴充套件的類，幫你完成TimeDecoder的開發。

package io.netty.example.time;

public class TimeDecoder extends ByteToMessageDecoder { // (1)
    @Override
    protected 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都會呼叫decode()方法來處理內部的那個累積緩衝。
Decode()方法可以決定當累積緩衝裡沒有足夠資料時可以往out物件裡放任意資料。當有更多的資料被接收了ByteToMessageDecoder會再一次呼叫decode()方法。
如果在decode()方法裡增加了一個物件到out物件裡，這意味著解碼器解碼訊息成功。ByteToMessageDecoder將會丟棄在累積緩衝裡已經被讀過的資料。請記得你不需要對多條訊息呼叫decode()，ByteToMessageDecoder會持續呼叫decode()直到不放任何資料到out裡。

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

如果你是一個大膽的人，你可能會嘗試使用更簡單的解碼類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還提供了更多可以直接拿來用的解碼器使你可以更簡單地實現更多的協議，幫助你避免開發一個難以維護的處理器實現。請參考下面的包以獲取更多更詳細的例子：

用POJO代替ByteBuf

我們已經討論了所有的例子，到目前為止一個訊息的訊息都是使用ByteBuf作為一個基本的資料結構。在這一部分，我們會改進TIME協議的客戶端和服務端的例子，用POJO替代ByteBuf。在你的ChannelHandlerS中使用POJO優勢是比較明顯的。通過從ChannelHandler中提取出ByteBuf的程式碼，將會使ChannelHandler的實現變得更加可維護和可重用。在TIME客戶端和服務端的例子中，我們讀取的僅僅是一個32位的整形資料，直接使用ByteBuf不會是一個主要的問題。然後，你會發現當你需要實現一個真實的協議，分離程式碼變得非常的必要。首先，讓我們定義一個新的型別叫做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;
    }

    @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.readInt()));
}

下面是修改後的解碼器，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);
}

現在，僅僅需要修改的是ChannelHandler的實現，這裡需要把UnixTime物件重新轉化為一個ByteBuf。不過這已經是非常簡單了，因為當你對一個訊息編碼的時候，你不需要再處理拆包和組裝的過程。

package io.netty.example.time;

public class TimeEncoder extends ChannelHandlerAdapter {
    @Override
    public 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)
    }
}

在這幾行程式碼裡還有幾個重要的事情。第一，通過ChannelPromise，當編碼後的資料被寫到了通道上Netty可以通過這個物件標記是成功還是失敗。第二，我們不需要呼叫cxt.flush()。因為處理器已經單獨分離出了一個方法void flush(ChannelHandlerContext cxt),如果像自己實現flush方法內容可以自行覆蓋這個方法。

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

最後的任務就是在TimeServerHandler之前把TimeEncoder插入到ChannelPipeline。但這是不那麼重要的工作。

關閉你的應用

關閉一個Netty應用往往只需要簡單地通過shutdownGracefully()方法來關閉你構建的所有的NioEventLoopGroupS.當EventLoopGroup被完全地終止,並且對應的所有channels都已經被關閉時，Netty會返回一個Future物件。

概述

在這一章節中，我們會快速地回顧下如果在熟練掌握Netty的情況下編寫出一個健壯能執行的網路應用程式。在Netty接下去的章節中還會有更多更相信的資訊。我們也鼓勵你去重新複習下在io.netty.example包下的例子。請注意社群一直在等待你的問題和想法以幫助Netty的持續改進，Netty的文件也是基於你們的快速反饋上。

Netty 5使用者指南

前言

問題

解決方案

入門指南

開始之前

DISCARD服務(丟棄服務，指的是會忽略所有接收的資料的一種協議)

觀察接收到的資料

ECHO服務（響應式協議）

TIME服務(時間協議的服務)

Time客戶端

流資料的傳輸處理

一個小的Socket Buffer問題

第一個解決方案

第二個解決方案

用POJO代替ByteBuf

關閉你的應用

概述

Netty 5使用者指南

Hibernate 5 入門指南-基於類註解

Hibernate 5 入門指南-基於Envers

1002 PowerBuilder 12.5 安裝指南

005-Sencha Cmd 5升級指南

Netty(5)：核心部件：Codec 編解碼器

netty 5 (2)解碼器

5月29日 Java性能調優指南讀後感

《Netty權威指南》（一）簡單的時間服務器P69

Redhat6.5基礎配置指南

《互聯網企業安全高級指南》讀書筆記1-5

Windows下安裝Redmine 2.5.2不全然指南

【執行力決定命】第5集《負能量者，請你滾出辦公室》

CentOS 6 ElasticSearch 5.5.x 安裝指南

2D拾荒者開發所學U3D的基礎知識5

hadoop權威指南(第四版)要點翻譯(5)——Chapter 3. The HDFS(5)

JDK1.5-JDK10 安裝指南(windows)

HTML 5與CSS 3權威指南(第2版·上冊) 中文pdf掃描版?

Netty核心概念(5)之Channel

angular 1.5.X 防入坑指南

Netty 5使用者指南

前言

問題

解決方案

入門指南

開始之前

DISCARD服務(丟棄服務，指的是會忽略所有接收的資料的一種協議)

觀察接收到的資料

ECHO服務（響應式協議）

TIME服務(時間協議的服務)

Time客戶端

流資料的傳輸處理

一個小的Socket Buffer問題

第一個解決方案

第二個解決方案

用POJO代替ByteBuf

關閉你的應用

概述

相關推薦