摘要：但是它不是自己創(chuàng)建線程，而是從調(diào)用構(gòu)造方法時(shí)指定的線程池中獲取線程。這就意味著，即使發(fā)送兩個(gè)獨(dú)立的消息，操作系統(tǒng)會(huì)把他們視為一個(gè)字節(jié)串。釋放過程很簡單，調(diào)用它的方法，所有相關(guān)的和線程池將會(huì)自動(dòng)關(guān)閉。

簡單找了下發(fā)現(xiàn)網(wǎng)上沒有關(guān)于Netty3比較完整的源碼解析的文章，于是我就去讀官方文檔，為了加強(qiáng)記憶，翻譯成了中文，有適當(dāng)?shù)暮喕?/p>

原文檔地址：Netty3文檔

運(yùn)行demo的前提有兩個(gè)：最新版本的Netty3和JDK1.5以上

最簡單的協(xié)議就是Discard協(xié)議——忽略所有接收到的數(shù)據(jù)并且不作任何響應(yīng)。我們從Netty處理I/O事件的handler實(shí)現(xiàn)開始：

public class DiscardServerHandler extends SimpleChannelHandler {

    @Override
    public void messageReceived(ChannelHandlerContext ctx, MessageEvent e) {
        
    }

    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, ExceptionEvent e) {

        e.getCause().printStackTrace();

        Channel ch = e.getChannel();
        ch.close();
    }
}

DiscardServerHandler 繼承SimpleChannelHandler——ChannelHandler的一個(gè)實(shí)現(xiàn)；

messageReceived方法接收MessageEvent類型的參數(shù)，它包含接收的客戶端數(shù)據(jù)；

exceptionCaught方法在出現(xiàn)I/O錯(cuò)誤或者處理事件時(shí)拋出錯(cuò)誤時(shí)被調(diào)用，通常包含記錄錯(cuò)誤信息和關(guān)閉通道的動(dòng)作；

接下來寫一個(gè)main方法來開啟使用DiscardServerHandler的服務(wù)：

public class DiscardServer {
    
    public static void main(String[] args) throws Exception {
        ChannelFactory factory =
                new NioServerSocketChannelFactory(
                        Executors.newCachedThreadPool(),
                        Executors.newCachedThreadPool());

        ServerBootstrap bootstrap = new ServerBootstrap(factory);

        bootstrap.setPipelineFactory(new ChannelPipelineFactory() {
            public ChannelPipeline getPipeline() {
                return Channels.pipeline(new DiscardServerHandler());
            }
        });
        
        bootstrap.setOption("child.tcpNoDelay", true);
        bootstrap.setOption("child.keepAlive", true);

        bootstrap.bind(new InetSocketAddress(8080));
    }
}

ChannelFactory是創(chuàng)建和管理Channel及其關(guān)聯(lián)資源的工廠，它負(fù)責(zé)處理所有I/O請求并且執(zhí)行I/O生成ChannelEvent。但是它不是自己創(chuàng)建I/O線程，而是從調(diào)用構(gòu)造方法時(shí)指定的線程池中獲取線程。服務(wù)端應(yīng)用使用NioServerSocketChannelFactory；

ServerBootstrap是一個(gè)設(shè)置服務(wù)端的幫助類；

當(dāng)服務(wù)端接收到一個(gè)新的連接，指定的ChannelPipelineFactory就會(huì)創(chuàng)建一個(gè)新的ChannelPipeline，這個(gè)新的Pipeline包含一個(gè)DiscardServerHandler對象；

你可以給Channel實(shí)現(xiàn)設(shè)置具體的參數(shù)，選項(xiàng)帶"child."前綴代表應(yīng)用在接收到的Channel上而不是服務(wù)端本身的ServerSocketChannel；

剩下的就是綁定端口啟動(dòng)服務(wù)，可以綁定多個(gè)不同的端口。

我們可以通過"telnet localhost 8080"命令去測試服務(wù)，但因?yàn)槭荄iscard服務(wù)，我們都不知道服務(wù)是否正常工作。所以我們修改下服務(wù)，讓它打印出接收到的數(shù)據(jù)。

@Override
public void messageReceived(ChannelHandlerContext ctx, MessageEvent e) {
    ChannelBuffer buf = (ChannelBuffer) e.getMessage();
    while(buf.readable()) {
        System.out.println((char) buf.readByte());
        System.out.flush();
    }
}

ChannelBuffer是Netty基本的存儲(chǔ)字節(jié)的數(shù)據(jù)結(jié)構(gòu)，跟NIO的ByteBuffer類似，但是更容易使用更靈活。比如Netty允許你在盡量少的內(nèi)存復(fù)制次數(shù)的情況下把多個(gè)ChannelBuffer組合成一個(gè)。

一個(gè)服務(wù)通常對請求是有響應(yīng)的。接下來我們嘗試寫一個(gè)實(shí)現(xiàn)Echo協(xié)議——將接收的數(shù)據(jù)原路返回給客戶端的服務(wù)：

@Override
public void messageReceived(ChannelHandlerContext ctx, MessageEvent e) {
    Channel ch = e.getChannel();
    ch.write(e.getMessage());
}

MessageEvent繼承了ChannnelEvent，一個(gè)ChannnelEvent持有它相關(guān)的Channel的引用。我們可以獲取這個(gè)Channel然后調(diào)用寫方法寫入數(shù)據(jù)返回給客戶端。

這次我們實(shí)現(xiàn)一個(gè)時(shí)間協(xié)議——在不需要任何請求數(shù)據(jù)的情況下返回一個(gè)32位整型數(shù)字并且在發(fā)送之后關(guān)閉連接。因?yàn)槲覀兒雎哉埱髷?shù)據(jù)，只需要在連接建立的發(fā)送消息，所以這次不能使用messageReceived方法而是重寫channelConnected方法：

@Override
public void channelConnected(ChannelHandlerContext ctx, ChannelStateEvent e) {
    Channel ch = e.getChannel();

    ChannelBuffer time = ChannelBuffers.buffer(4);
    time.writeInt((int) (System.currentTimeMillis() / 1000L + 2208988800L));

    ChannelFuture f = ch.write(time);

    f.addListener(new ChannelFutureListener() {
        public void operationComplete(ChannelFuture future) {
            Channel ch = future.getChannel();
            ch.close();
        }
    });
}

channelConnected方法在連接建立的時(shí)間被調(diào)用，然后我們寫入一個(gè)32位整型數(shù)字代表以秒為單位的當(dāng)前時(shí)間；

我們使用ChannelBuffers工具類分配了一個(gè)容量為4字節(jié)的ChannelBuffer來存放這個(gè)32位整型數(shù)字；

然后我們把ChannelBuffer寫入Channel...等一下，flip方法哪里去了？在NIO中我們不是要在寫入通道前調(diào)用ByteBuffer的flip方法的嗎？ChannelBuffer沒有這個(gè)方法，因?yàn)樗袃蓚€(gè)指針，一個(gè)用于讀操作一個(gè)用于寫操作。當(dāng)數(shù)據(jù)寫入ChannelBuffer時(shí)寫索引增加而讀索引不變。讀索引和寫索引相互獨(dú)立。對比之下，Netty的ChannelBuffer比NIO的buffer更容易使用。

另外需要注意的一點(diǎn)是ChannelBuffer的write方法返回的是一個(gè)ChannelFuture對象。它表示一個(gè)還未發(fā)生的I/O操作，因?yàn)镹etty中所有操作都是異步的。所以我們必須在ChannelFuture收到操作完成的通知之后才能關(guān)閉Channel。哦，對了，close方法也是返回ChannelFuture...

那么問題來了，我們?nèi)绾蔚玫讲僮魍瓿傻耐ㄖ?？只需要簡單得向返回的ChannelFuture對象中添加一個(gè)ChannelFutureListener，這里我們創(chuàng)建了一個(gè)ChannelFutureListener的匿名內(nèi)部類，它在操作完成的時(shí)候會(huì)關(guān)閉Channel。

我們還需要一個(gè)遵守時(shí)間協(xié)議，即能把整型數(shù)字翻譯成日期的客戶端。Netty服務(wù)端和客戶端唯一的區(qū)別就是要求不同的Bootstrap和ChannelFactory：

public static void main(String[] args) throws Exception {
    String host = args[0];
    int port = Integer.parseInt(args[1]);

    ChannelFactory factory =
            new NioClientSocketChannelFactory(
                    Executors.newCachedThreadPool(),
                    Executors.newCachedThreadPool());

    ClientBootstrap bootstrap = new ClientBootstrap(factory);

    bootstrap.setPipelineFactory(new ChannelPipelineFactory() {
        public ChannelPipeline getPipeline() {
            return Channels.pipeline(new TimeClientHandler());
        }
    });

    bootstrap.setOption("tcpNoDelay", true);
    bootstrap.setOption("keepAlive", true);

    bootstrap.connect(new InetSocketAddress(host, port));
}

NioClientSocketChannelFactory，用來創(chuàng)建一個(gè)客戶端Channel；

ClientBootstrap是ServerBootStrap在客戶端的對應(yīng)部分；

需要注意的是設(shè)置參數(shù)時(shí)不需要"child."前綴，客戶端SocketChannel沒有父Channel；

對應(yīng)服務(wù)端的bind方法，這里我們需要調(diào)用connect方法。

另外我們需要一個(gè)ChannelHandler實(shí)現(xiàn)，負(fù)責(zé)把接收到服務(wù)端返回的32位整型數(shù)字翻譯成日期并打印出來，然后斷開連接：

public class TimeClientHandler extends SimpleChannelHandler {

    @Override
    public void messageReceived(ChannelHandlerContext ctx, MessageEvent e) {
        ChannelBuffer buf = (ChannelBuffer) e.getMessage();
        long currentTimeMillis = buf.readInt() * 1000L;
        System.out.println(new Date(currentTimeMillis));
        e.getChannel().close();
    }

    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, ExceptionEvent e) {
        e.getCause().printStackTrace();
        e.getChannel().close();
    }
}

看上去很簡單是吧？但是實(shí)際運(yùn)行過程中這個(gè)handler有時(shí)會(huì)拋出一個(gè)IndexOutOfBoundsException。下一節(jié)我們會(huì)討論為什么會(huì)這樣。

在像TCP/IP那樣基于流的傳輸中，接收數(shù)據(jù)保存在一個(gè)socket接收緩存中。但是這個(gè)緩存不是一個(gè)以包為單位的隊(duì)列，而是一個(gè)以字節(jié)為單位的隊(duì)列。這就意味著，即使發(fā)送兩個(gè)獨(dú)立的消息，操作系統(tǒng)會(huì)把他們視為一個(gè)字節(jié)串。因此，不能保證你讀到的和另一端寫入的一樣。所以，不管是客戶端還是服務(wù)端，對于接收到的數(shù)據(jù)都需要整理成符合應(yīng)用程序邏輯的結(jié)構(gòu)。

回到前面的時(shí)間客戶端的問題，32位整型數(shù)字很小，但是它也是可以拆分的，特別是當(dāng)流量上升的時(shí)候，被拆分的可能性也隨之上升。
一個(gè)簡單的處理方式就是內(nèi)部創(chuàng)建一個(gè)累計(jì)的緩存，直到接收滿4個(gè)字節(jié)才進(jìn)行處理。

private final ChannelBuffer buf = dynamicBuffer();

@Override
public void messageReceived(ChannelHandlerContext ctx, MessageEvent e) {
    ChannelBuffer m = (ChannelBuffer) e.getMessage();
    buf.writeBytes(m);
    if (buf.readableBytes() >= 4) {
        long currentTimeMillis = buf.readInt() * 1000L;
        System.out.println(new Date(currentTimeMillis));
        e.getChannel().close();
    }
}

ChannelBuffers.dynamicBuffer()返回一個(gè)自動(dòng)擴(kuò)容的ChannelBuffer；

所有接收的數(shù)據(jù)都累積到這個(gè)動(dòng)態(tài)緩存中；

handler需要檢查緩存是否滿4個(gè)字節(jié)，是的話才能繼續(xù)業(yè)務(wù)邏輯；否則，Netty會(huì)在數(shù)據(jù)繼續(xù)到達(dá)之后持續(xù)調(diào)用messageReceive。

第一種方案有很多問題，比如一個(gè)復(fù)雜的協(xié)議，由多個(gè)可變長度的域組成，這種情況下第一種方案的handler就無法支持了。
你會(huì)發(fā)現(xiàn)你可以添加多個(gè)ChannelHandler到ChannelPipeline中，利用這個(gè)特性，你可以把一個(gè)臃腫的ChannelHandler拆分到多個(gè)模塊化的ChannelHandler中，這樣可以降低應(yīng)用程序的復(fù)雜度。比如，你可以把TimeClientHandler拆分成兩個(gè)handler：

TimeDecoder，負(fù)責(zé)分段問題；

最初那個(gè)簡版的TimeClientHandler.

Netty提供了可擴(kuò)展的類幫助你實(shí)現(xiàn)TimeDecoder：

public class TimeDecoder extends FrameDecoder {

    @Override
    protected Object decode(ChannelHandlerContext ctx, Channel channel, ChannelBuffer buffer) {

        if (buffer.readableBytes() < 4) {
            return null;
        }

        return buffer.readBytes(4);
    }
}

FrameDecoder是ChannelHandler的一種實(shí)現(xiàn)，專門用來處理分段問題；

FrameDecoder在每次接收到新的數(shù)據(jù)時(shí)調(diào)用decode方法，攜帶一個(gè)內(nèi)部維持的累積緩存；

如果返回null，說明目前數(shù)據(jù)接收的還不夠，當(dāng)數(shù)據(jù)量足夠時(shí)FrameDecoder會(huì)再次調(diào)用方法；

如果返回非null對象，代表解碼成功，F(xiàn)rameDecoder會(huì)丟棄累積緩存中剩余的數(shù)據(jù)。你無需提供批量解碼，F(xiàn)rameDecoder會(huì)繼續(xù)調(diào)用decode方法直到返回null。

拆分之后，我們需要修改TimeClient的ChannelPipelineFactory實(shí)現(xiàn)：

bootstrap.setPipelineFactory(new ChannelPipelineFactory() {
    public ChannelPipeline getPipeline() {
        return Channels.pipeline(
                new TimeDecoder(),
                new TimeClientHandler());
    }
});

Netty還提供了進(jìn)一步簡化解碼的ReplayingDecoder：

public class TimeDecoder extends ReplayingDecoder {
    @Override
    protected Object decode(
            ChannelHandlerContext ctx, Channel channel,
            ChannelBuffer buffer, VoidEnum state) {
        return buffer.readBytes(4);
    }
 }

此外，Netty提供了一批開箱即用的解碼器，讓你可以簡單得實(shí)現(xiàn)大多數(shù)協(xié)議：

org.jboss.netty.example.factorial 用于二進(jìn)制協(xié)議；

org.jboss.netty.example.telnet 用于基于行的文本協(xié)議.

上面的demo我們都是用ChannelBuffer作為協(xié)議化消息的基本數(shù)據(jù)結(jié)構(gòu)，這一節(jié)我們用POJO替代ChannelBuffer。將從ChannelBuffer提取信息的代碼跟handler分離開，會(huì)使handler變得更加可維護(hù)的和可重用的。從上面的demo里不容易看出這個(gè)優(yōu)勢，但是實(shí)際應(yīng)用中分離很有必要。
首先，我們定義一個(gè)類型UnixTime：

public class UnixTime {

    private final int value;

    public UnixTime(int value) {
        this.value = value;
    }

    public int getValue() {
        return value;
    }

    @Override
    public String toString() {
        return new Date(value * 1000L).toString();
    }
}

現(xiàn)在我們可以修改TimeDecoder讓它返回一個(gè)UnixTime而不是ChannelBuffer：

@Override
protected Object decode(ChannelHandlerContext ctx, Channel channel, ChannelBuffer buffer) {

    if (buffer.readableBytes() < 4) {
        return null;
    }

    return new UnixTime(buffer.readInt());
}

編碼器改了，那么相應(yīng)的TimeClientHandler就不會(huì)繼續(xù)使用ChannelBuffer：

@Override
public void messageReceived(ChannelHandlerContext ctx, MessageEvent e) {
    UnixTime m = (UnixTime) e.getMessage();
    System.out.println(m);
    e.getChannel().close();
}

同樣的技術(shù)也可以應(yīng)用到服務(wù)端的TimeServerHandler上：

@Override
public void channelConnected(ChannelHandlerContext ctx, ChannelStateEvent e) {
    UnixTime time = new UnixTime((int)(System.currentTimeMillis() / 1000));
    ChannelFuture f = e.getChannel().write(time);
    f.addListener(ChannelFutureListener.CLOSE);
}

能這樣運(yùn)用的前提是有一個(gè)編碼器，可以把UnixTime對象翻譯成ChannelBuffer：

public class TimeEncoder extends SimpleChannelHandler {

    public void writeRequested(ChannelHandlerContext ctx, MessageEvent e) {
        UnixTime time = (UnixTime) e.getMessage();
        ChannelBuffer buf = buffer(4);
        buf.writeInt(time.getValue());
        Channels.write(ctx, e.getFuture(), buf);
    }
}

一個(gè)編碼器重寫writeRequested方法攔截一個(gè)寫請求。這里需要注意的一點(diǎn)是，盡管這里的writeRequested方法參數(shù)里也有一個(gè)MessageEvent對象，客戶端TimeClientHandler的messageReceived的參數(shù)里也有一個(gè)，但是它們的解讀是完全不同的。一個(gè)ChannelEvent可以是upstream也可以是downstream事件，這取決于事件的流向。messageReceived方法里的MessageEvent是一個(gè)upstream事件，而writeRequested方法里的是downstream事件。

當(dāng)把POJO類轉(zhuǎn)化為ChannelBuffer后，你需要把ChannelBuffer轉(zhuǎn)發(fā)到之前在ChannelPipeline內(nèi)的ChannelDownstreamHandler，也就是TimeServerHandler。Channels提供了多個(gè)幫助方法創(chuàng)建和發(fā)送ChanenlEvent。

同樣，TimeEncoder也需要加入到服務(wù)端的ChannelPipeline中：

bootstrap.setPipelineFactory(new ChannelPipelineFactory() {
    public ChannelPipeline getPipeline() {
        return Channels.pipeline(
                new TimeServerHandler(),
                new TimeEncoder());
    }
});

為了關(guān)閉I/O線程讓應(yīng)用程序優(yōu)雅得退出，我們需要釋放ChannelFactory分配的資源。
一個(gè)典型網(wǎng)絡(luò)應(yīng)用程序的關(guān)閉過程分為三步：

關(guān)閉所有服務(wù)端socket連接；

關(guān)閉所有非服務(wù)端socket連接（包括客戶端socket和服務(wù)端接收到的socket）；

釋放ChannelFactory使用的所有資源。

應(yīng)用到TimeClient上：

ChannelFuture future = bootstrap.connect(...);
future.awaitUninterruptibly();
if (!future.isSuccess()) {
    future.getCause().printStackTrace();
}
future.getChannel().getCloseFuture().awaitUninterruptibly();
factory.releaseExternalResources();

CilentBootStrap的connect方法返回一個(gè)ChannelFuture，當(dāng)連接嘗試成功或者失敗時(shí)會(huì)通知到ChannelFuture。它還持有連接嘗試關(guān)聯(lián)的Channel的引用；

ChannelFuture.awaitUninterruptibly()等待ChannelFuture確定連接是否嘗試成功；

如果連接失敗，我們打印出失敗的原因。ChannelFuture.getCause()會(huì)在連接即沒有成功也沒有取消的情況下返回失敗的原因；

連接嘗試的情況處理之后，我們還需要等待連接關(guān)閉。每個(gè)Channel有它自己的closeFuture，用來通知你連接關(guān)閉然后你可以針對關(guān)閉做一些動(dòng)作。即使連接嘗試失敗了，closeFuture仍然會(huì)被通知，因?yàn)镃hannel會(huì)在連接失敗后自動(dòng)關(guān)閉；

所有連接關(guān)閉之后，剩下的就是釋放ChannelFactory使用的資源了。釋放過程很簡單，調(diào)用它的releaseExternalResources方法，所有相關(guān)的NIO Selector和線程池將會(huì)自動(dòng)關(guān)閉。

關(guān)閉一個(gè)客戶端很簡單，那服務(wù)端呢？你需要從端口解綁并關(guān)閉所有接收到的連接。前提是你需要一個(gè)保持跟蹤活躍連接的數(shù)據(jù)結(jié)構(gòu)，Netty提供了ChannelGroup。

ChannelGroup是Java集合API一個(gè)特殊的的擴(kuò)展，它代表一組打開的Channel。如果一個(gè)Channel被添加到ChannelGroup，然后這個(gè)Channel被關(guān)閉了，它會(huì)從ChannelGroup中自動(dòng)移除。你可以對同一ChannelGroup中的Channel做批量操作，比如在關(guān)閉服務(wù)的時(shí)候關(guān)閉所有Channel。

要跟蹤打開的socket，你需要修改TimeServerHandler，把新打開的Channel添加到全局的ChannelGroup變量中。ChannelGroup是線程安全的。

@Override
public void channelOpen(ChannelHandlerContext ctx, ChannelStateEvent e) {
    TimeServer.allChannels.add(e.getChannel());
}

現(xiàn)在我們自動(dòng)維持了一個(gè)包含所有活躍Channel的列表，關(guān)閉服務(wù)端就像關(guān)閉客戶端一樣容易了。

public class TimeServer {
    static final ChannelGroup allChannels = new DefaultChannelGroup("time-server");

    public static void main(String[] args) throws Exception {
        ...
        ChannelFactory factory = ...;
        ...
        ServerBootstrap bootstrap = ...;
        ...
        Channel channel = bootstrap.bind(new InetSocketAddress(8080));
        allChannels.add(channel);
        waitForShutdownCommand();
        ChannelGroupFuture future = allChannels.close();
        future.awaitUninterruptibly();
        factory.releaseExternalResources();
    }
}

DefaultChannelGroup構(gòu)造方法接收組名為參數(shù)，組名是它的唯一標(biāo)識(shí)；

ServerBootstrap的bind方法返回一個(gè)服務(wù)端的綁定指定本地地址的Channel，調(diào)用Channel的close方法將會(huì)使它與本地地址解綁；

所有Channel類型都可以被添加到ChannelGroup中，不管是客戶端、服務(wù)端或是服務(wù)端接收的。因?yàn)槟憧梢栽诜?wù)器關(guān)閉時(shí)同時(shí)關(guān)閉綁定的Channel和接收到的Channel；

waitForShutdownCommand()是一個(gè)等待關(guān)閉信號的虛構(gòu)方法。

我們可以對ChannelGroup中的Channel進(jìn)行統(tǒng)一操作，這里我們調(diào)用close方法，相當(dāng)于解綁服務(wù)端Channel并且異步關(guān)閉所有接收到的Channel。close方法返回一個(gè)功能和ChannelFuture相近的ChannelGroupFuture，在所有連接都成功關(guān)閉通知我們。

這一節(jié)我們快速瀏覽了Netty，示范了如何用Netty寫一個(gè)能正常工作的網(wǎng)絡(luò)應(yīng)用。
下一節(jié)將介紹Netty的更多細(xì)節(jié)。