《netty實戰(zhàn)》閱讀筆記（1）——Netty 的概念及體系結(jié)構(gòu)

solocoder 發(fā)布于2019-08-15 17:50 / 1199人閱讀

摘要：它使用了事件通知以確定在一組非阻塞套接字中有哪些已經(jīng)就緒能夠進行相關(guān)的操作。目前，可以把看作是傳入入站或者傳出出站數(shù)據(jù)的載體。出站事件是未來將會觸發(fā)的某個動作的操作結(jié)果，這些動作包括打開或者關(guān)閉到遠(yuǎn)程節(jié)點的連接將數(shù)據(jù)寫到或者沖刷到套接字。

netty的概念 定義

Netty 是一款異步的事件驅(qū)動的網(wǎng)絡(luò)應(yīng)用程序框架，支持快速地開發(fā)可維護的高性能的面向協(xié)議的服務(wù)器和客戶端。我們可以很簡單的使用Netty 構(gòu)建應(yīng)用程序，你不必是一名網(wǎng)絡(luò)編程專家；而且Netty 比直接使用底層的Java API 容易得多，它推崇良好的設(shè)計實踐，可以將你的應(yīng)用程序邏輯和網(wǎng)絡(luò)層解耦。

Netty的特性總結(jié)

在我們開始首次深入地了解Netty 之前，請仔細(xì)審視表1-1 中所總結(jié)的關(guān)鍵特性。有些是技術(shù)性的，而其他的更多的則是關(guān)于架構(gòu)或設(shè)計哲學(xué)的。在本書的學(xué)習(xí)過程中，我們將不止一次地重新審視它們。

Java NIO

在深入netty之前，我們先來簡單說說NIO；我們都知道，它和以前的普通I/O相比的最大優(yōu)勢在于它是非阻塞的。

阻塞I/O

因為普通I/O的阻塞，以前我們設(shè)計并發(fā)只能像下圖這樣，為每個I/O分配一個線程：

這顯然帶來了一些問題：

在任何時候都可能有大量的線程處于休眠狀態(tài)，只是等待輸入或者輸出數(shù)據(jù)就緒，這可能算是一種資源浪費。

需要為每個線程的調(diào)用棧都分配內(nèi)存，其默認(rèn)值大小區(qū)間為64 KB 到1 MB，具體取決于操作系統(tǒng)。

即使Java 虛擬機（JVM）在物理上可以支持非常大數(shù)量的線程，但是遠(yuǎn)在到達該極限之前，上下文切換所帶來的開銷就會帶來麻煩，例如，在達到10 000 個連接的時候。

非阻塞I/O

Java 對于非阻塞I/O 的支持是在2002 年引入的，位于JDK 1.4 的java.nio 包中。下圖展示了一個非阻塞設(shè)計，其實際上消除了普通I/O的那些弊端。選擇器使得我們能夠通過較少的線程便可監(jiān)視許多連接上的事件。

class java.nio.channels.Selector 是Java 的非阻塞I/O 實現(xiàn)的關(guān)鍵。它使用了事件通知API以確定在一組非阻塞套接字中有哪些已經(jīng)就緒能夠進行I/O 相關(guān)的操作。因為可以在任何的時間檢查任意的讀操作或者寫操作的完成狀態(tài)，所以如上圖所示，一個單一的線程便可以處理多個并發(fā)的連接。

總體來看，與阻塞I/O 模型相比，這種模型提供了更好的資源管理：

使用較少的線程便可以處理許多連接，因此也減少了內(nèi)存管理和上下文切換所帶來開銷；

當(dāng)沒有I/O 操作需要處理的時候，線程也可以被用于其他任務(wù)。盡管已經(jīng)有許多直接使用Java NIO API 的應(yīng)用程序被構(gòu)建了，但是要做到如此正確和安全并不容易。特別是，在高負(fù)載下可靠和高效地處理和調(diào)度I/O 操作是一項繁瑣而且容易出錯的任務(wù)，最好留給高性能的網(wǎng)絡(luò)編程專家——Netty。

netty核心組件

在本節(jié)中我將要討論Netty 的主要構(gòu)件塊：

Channel —— 可以看做是Socket的抽象；

回調(diào)；

ChannelFuture—— 異步通知；

事件和ChannelHandler。

EventLoop —— 控制流、多線程處理、并發(fā)；

ChannelPipeline —— 提供了ChannelHandler 鏈的容器

引導(dǎo) —— Bootstrap和ServerBootstrap

這些構(gòu)建塊代表了不同類型的構(gòu)造：資源、邏輯以及通知。你的應(yīng)用程序?qū)⑹褂盟鼈儊碓L問網(wǎng)絡(luò)以及流經(jīng)網(wǎng)絡(luò)的數(shù)據(jù)。

對于每個組件來說，我們都將提供一個基本的定義，并且在適當(dāng)?shù)那闆r下，還會提供一個簡單的示例代碼來說明它的用法。

Channel

基本的I/O 操作（bind()、connect()、read()和write()）依賴于底層網(wǎng)絡(luò)傳輸所提供的原語。在基于Java 的網(wǎng)絡(luò)編程中，其基本的構(gòu)造是class Socket。Netty 的Channel 接口所提供的API，大大地降低了直接使用Socket 類的復(fù)雜性。

Channel 是Java NIO 的一個基本構(gòu)造。它代表一個到實體（如一個硬件設(shè)備、一個文件、一個網(wǎng)絡(luò)套接字或者一個能夠執(zhí)行一個或者多個不同的I/O操作的程序組件）的開放連接，如讀操作和寫操作。

目前，可以把Channel 看作是傳入（入站）或者傳出（出站）數(shù)據(jù)的載體。因此，它可以被打開或者被關(guān)閉，連接或者斷開連接。

回調(diào)

一個回調(diào)其實就是一個方法，一個指向已經(jīng)被提供給另外一個方法的方法的引用。這使得后者可以在適當(dāng)?shù)臅r候調(diào)用前者?；卣{(diào)在廣泛的編程場景中都有應(yīng)用，而且也是在操作完成后通知相關(guān)方最常見的方式之一。

ChannelFuture

Future 提供了另一種在操作完成時通知應(yīng)用程序的方式。這個對象可以看作是一個異步操作的結(jié)果的占位符；它將在未來的某個時刻完成，并提供對其結(jié)果的訪問。

JDK 預(yù)置了interface java.util.concurrent.Future，但是其所提供的實現(xiàn)，只允許手動檢查對應(yīng)的操作是否已經(jīng)完成，或者一直阻塞直到它完成。這是非常繁瑣的，所以Netty提供了它自己的實現(xiàn)——ChannelFuture，用于在執(zhí)行異步操作的時候使用。

ChannelFuture提供了幾種額外的方法，這些方法使得我們能夠注冊一個或者多個ChannelFutureListener實例。監(jiān)聽器的回調(diào)方法operationComplete()，將會在對應(yīng)的操作完成時被調(diào)用。然后監(jiān)聽器可以判斷該操作是成功地完成了還是出錯了。如果是后者，我們可以檢索產(chǎn)生的Throwable。簡而言之，由ChannelFutureListener提供的通知機制消除了手動檢查對應(yīng)的操作是否完成的必要。

下面展示了一個異步地連接到遠(yuǎn)程節(jié)點,ChannelFuture 作為一個I/O 操作的一部分返回的例子。這里，connect()方法將會直接返回，而不會阻塞。

Channel channel = ...;
ChannelFuture future = channel.connect(
new InetSocketAddress("192.168.0.1", 25));

下面的代碼顯示了如何利用ChannelFutureListener。首先，要連接到遠(yuǎn)程節(jié)點上。然后，要注冊一個新的ChannelFutureListener 到對connect()方法的調(diào)用所返回的ChannelFuture 上。當(dāng)該監(jiān)聽器被通知連接已經(jīng)建立的時候，要檢查對應(yīng)的狀態(tài)。如果該操作是成功的，那么將數(shù)據(jù)寫到該Channel。否則，要從ChannelFuture 中檢索對應(yīng)的Throwable。

Channel channel = ...;
// 連接遠(yuǎn)程節(jié)點
ChannelFuture future = channel.connect(
    new InetSocketAddress("192.168.0.1", 25));
//注冊一個ChannelFutureListener，以便在操作完成時獲得通知
future.addListener(new ChannelFutureListener() {
    @Override
        public void operationComplete(ChannelFuture future) {
        //狀態(tài)判斷
        if (future.isSuccess()){
            //如果操作是成功的，則創(chuàng)建一個ByteBuf 以持有數(shù)據(jù)
            ByteBuf buffer = Unpooled.copiedBuffer(
                "Hello",Charset.defaultCharset());
            //將數(shù)據(jù)異步地發(fā)送到遠(yuǎn)程節(jié)點。返回一個ChannelFuture
            ChannelFuture wf = future.channel()
                .writeAndFlush(buffer);
            ....
        } else {
            //如果發(fā)生錯誤，則訪問描述原因的Throwable
            Throwable cause = future.cause();
            cause.printStackTrace();
        }
    }
});

如果你把ChannelFutureListener 看作是回調(diào)的一個更加精細(xì)的版本，那么你是對的。事實上，回調(diào)和Future 是相互補充的機制；它們相互結(jié)合，構(gòu)成了Netty 本身的關(guān)鍵構(gòu)件塊之一。

事件和ChannelHandler 事件

Netty 使用不同的事件來通知我們狀態(tài)的改變或者是操作的狀態(tài)。這使得我們能夠基于已經(jīng)發(fā)生的事件來觸發(fā)適當(dāng)?shù)膭幼?。這些動作可能是：

記錄日志；

數(shù)據(jù)轉(zhuǎn)換；

流控制；

應(yīng)用程序邏輯。

Netty 是一個網(wǎng)絡(luò)編程框架，所以事件是按照它們與入站或出站數(shù)據(jù)流的相關(guān)性進行分類的?？赡苡扇胝緮?shù)據(jù)或者相關(guān)的狀態(tài)更改而觸發(fā)的事件包括：

連接已被激活或者連接失活；

數(shù)據(jù)讀?。?/p>

用戶事件；

錯誤事件。

出站事件是未來將會觸發(fā)的某個動作的操作結(jié)果，這些動作包括：

打開或者關(guān)閉到遠(yuǎn)程節(jié)點的連接；

將數(shù)據(jù)寫到或者沖刷到套接字。

ChannelHandler

從應(yīng)用程序開發(fā)人員的角度來看，Netty 的主要組件是ChannelHandler，它充當(dāng)了所有處理入站和出站數(shù)據(jù)的應(yīng)用程序邏輯的容器。該組件實現(xiàn)了服務(wù)器對從客戶端接收的數(shù)據(jù)的處理。每個事件都可以被分發(fā)給ChannelHandler 類中的某個用戶實現(xiàn)的方法。這是一個很好的將事件驅(qū)動范式直接轉(zhuǎn)換為應(yīng)用程序構(gòu)件塊的例子。圖1-3 展示了一個事件是如何被一個這樣的ChannelHandler 鏈處理的。

Netty 的ChannelHandler 為處理器提供了基本的抽象，如圖1-3 所示的那些。你可以認(rèn)為每個ChannelHandler 的實例都類似于一種為了響應(yīng)特定事件而被執(zhí)行的回調(diào)。
下列代碼就是一個handler的示例：

@ChannelHandler.Sharable
public class EchoClientHandler extends
        SimpleChannelInboundHandler {
    //重寫了channelActive()方法，其將在一個連接建立時被調(diào)用
    @Override
    public void channelActive(ChannelHandlerContext ctx) {
        ctx.writeAndFlush(Unpooled.copiedBuffer("Netty rocks!",
                CharsetUtil.UTF_8));
    }
    //重寫了channelRead0()方法。每當(dāng)接收數(shù)據(jù)時，都會調(diào)用這個方法。
    //需要注意的是，由服務(wù)器發(fā)送的消息可能會被分塊接收。
    // 也就是說，如果服務(wù)器發(fā)送了5 字節(jié)，那么不能保證這5 字節(jié)會被一次性接收。
    //即使是對于這么少量的數(shù)據(jù)，channelRead0()方法也可能
    // 會被調(diào)用兩次，第一次使用一個持有3 字節(jié)的ByteBuf（Netty 的字節(jié)容器）
    // 第二次使用一個持有2 字節(jié)的ByteBuf。
    @Override
    public void channelRead0(ChannelHandlerContext ctx, ByteBuf in) {
        System.out.println(
                "Client received: " + in.toString(CharsetUtil.UTF_8));
    }
    //發(fā)生異常時被調(diào)用
    @Override
    public void exceptionCaught(ChannelHandlerContext ctx,
                                Throwable cause) {
        cause.printStackTrace();
        ctx.close();
    }
}

channelHandler的主要抽象方法都定義于ChannelHandlerAdapter類中，我們通過重寫適當(dāng)?shù)姆椒?，來控制整個生命周期的重要節(jié)點的邏輯。

Netty 提供了大量預(yù)定義的可以開箱即用的ChannelHandler 實現(xiàn)，包括用于各種協(xié)議（如HTTP 和SSL/TLS）的ChannelHandler。在內(nèi)部，ChannelHandler 自己也使用了事件和Future，使得它們也成為了你的應(yīng)用程序?qū)⑹褂玫南嗤橄蟮南M者。

ChannelPipeline

ChannelPipeline 提供了ChannelHandler 鏈的容器，并定義了用于在該鏈上傳播入站和出站事件流的API。當(dāng)Channel 被創(chuàng)建時，它會被自動地分配到它專屬的ChannelPipeline。

ChannelHandler 安裝到ChannelPipeline 中的過程如下所示：

    ServerBootstrap b = new ServerBootstrap();
    //一個ChannelInitializer的實現(xiàn)被注冊到了ServerBootstrap中①；
    b.group(group)
        .channel(NioServerSocketChannel.class)
        .localAddress(new InetSocketAddress(port))
        .childHandler(new ChannelInitializer() {
            //當(dāng)ChannelInitializer.initChannel()方法被調(diào)用時ChannelInitializer將在
            //ChannelPipeline 中安裝一組自定義的ChannelHandler  serverHandler；
            @Override
            public void initChannel(SocketChannel ch)
                    throws Exception {
                ch.pipeline().addLast(serverHandler);
            }
        });

EventLoop

EventLoop 定義了Netty 的核心抽象，用于處理連接的生命周期中所發(fā)生的事件。圖3-1
下圖在高層次上說明了Channel、EventLoop、Thread 以及EventLoopGroup 之間的關(guān)系。

這些關(guān)系是：

一個EventLoopGroup 包含一個或者多個EventLoop；

一個EventLoop 在它的生命周期內(nèi)只和一個Thread 綁定；所有由EventLoop 處理的I/O 事件都將在它專有的Thread 上被處理；

一個Channel 在它的生命周期內(nèi)只注冊于一個EventLoop；一個EventLoop 可能會被分配給一個或多個Channel。

注意，在這種設(shè)計中，一個給定Channel 的I/O 操作都是由相同的Thread 執(zhí)行的，實際
上消除了不同線程間對于同步的需要。

Bootstrap和ServerBootstrap

Netty 的引導(dǎo)類為應(yīng)用程序的網(wǎng)絡(luò)層配置提供了容器，這涉及將一個進程綁定到某個指定的端口（ServerBootstrap），或者將一個進程連接到另一個運行在某個指定主機的指定端口上的進程（Bootstrap）。Netty提供兩種類型的引導(dǎo)，一種用于客戶端（簡單地稱為Bootstrap），而另一種（ServerBootstrap）用于服務(wù)器。無論你的應(yīng)用程序使用哪種協(xié)議或者處理哪種類型的數(shù)據(jù)，唯一決定它使用哪種引導(dǎo)類的是它是作為一個客戶端還是作為一個服務(wù)器。表3-1 比較了這兩種類型的引導(dǎo)類。

這兩種類型的引導(dǎo)類之間的第一個區(qū)別已經(jīng)討論過了：ServerBootstrap 將綁定到一個端口，因為服務(wù)器必須要監(jiān)聽連接，而Bootstrap 則是由想要連接到遠(yuǎn)程節(jié)點的客戶端應(yīng)用程序所使用的。

第二個區(qū)別可能更加明顯。引導(dǎo)一個客戶端只需要一個EventLoopGroup，但是一個ServerBootstrap 則需要兩個（也可以是同一個實例）。為什么呢？

因為服務(wù)器需要兩組不同的Channel。第一組將只包含一個ServerChannel，代表服務(wù)器自身的已綁定到某個本地端口的正在監(jiān)聽的套接字。而第二組將包含所有已創(chuàng)建的用來處理傳入客戶端連接（對于每個服務(wù)器已經(jīng)接受的連接都有一個）的Channel。圖3-4 說明了這個模型，并且展示了為何需要兩個不同的EventLoopGroup。

與ServerChannel 相關(guān)聯(lián)的EventLoopGroup 將分配一個負(fù)責(zé)為傳入連接請求創(chuàng)建Channel 的EventLoop。一旦連接被接受，第二個EventLoopGroup 就會給它的Channel分配一個EventLoop。

一個簡單的Netty服務(wù)端和客戶端交互demo 客戶端 自定義channelHandler

import io.netty.buffer.ByteBuf;
import io.netty.buffer.Unpooled;
import io.netty.channel.ChannelHandler;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.SimpleChannelInboundHandler;
import io.netty.util.CharsetUtil;

@ChannelHandler.Sharable
public class EchoClientHandler extends
        SimpleChannelInboundHandler {
    //重寫了channelActive()方法，其將在一個連接建立時被調(diào)用
    @Override
    public void channelActive(ChannelHandlerContext ctx) {
        ctx.writeAndFlush(Unpooled.copiedBuffer("Netty rocks!",
                CharsetUtil.UTF_8));
    }
    //重寫了channelRead0()方法。每當(dāng)接收數(shù)據(jù)時，都會調(diào)用這個方法。
    //需要注意的是，由服務(wù)器發(fā)送的消息可能會被分塊接收。
    // 也就是說，如果服務(wù)器發(fā)送了5 字節(jié)，那么不能保證這5 字節(jié)會被一次性接收。
    //即使是對于這么少量的數(shù)據(jù)，channelRead0()方法也可能
    // 會被調(diào)用兩次，第一次使用一個持有3 字節(jié)的ByteBuf（Netty 的字節(jié)容器）
    // 第二次使用一個持有2 字節(jié)的ByteBuf。
    @Override
    public void channelRead0(ChannelHandlerContext ctx, ByteBuf in) {
        System.out.println(
                "Client received: " + in.toString(CharsetUtil.UTF_8));
    }
    //發(fā)生異常時被調(diào)用
    @Override
    public void exceptionCaught(ChannelHandlerContext ctx,
                                Throwable cause) {
        cause.printStackTrace();
        ctx.close();
    }
}

客戶端實例

import io.netty.bootstrap.Bootstrap;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.EventLoopGroup;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioSocketChannel;

import java.net.InetSocketAddress;

public class EchoClient {
    private final String host;
    private final int port;
    public EchoClient(String host, int port) {
        this.host = host;
        this.port = port;
    }
    public void start() throws Exception {
        //定義EventLoop
        EventLoopGroup group = new NioEventLoopGroup();
        try {
            //Bootstrap類包提供包含豐富API的幫助類，能夠非常方便的實現(xiàn)典型的服務(wù)器端和客戶端通道初始化功能。
            Bootstrap b = new Bootstrap();
            //綁定EventLoop
            b.group(group)
                    //使用默認(rèn)的channelFactory創(chuàng)建一個channel
                    .channel(NioSocketChannel.class)
                    //定義遠(yuǎn)程地址
                    .remoteAddress(new InetSocketAddress(host, port))
                    //綁定自定義的EchoClientHandler到ChannelPipeline上
                    .handler(new ChannelInitializer() {
                        @Override
                        public void initChannel(SocketChannel ch)
                                throws Exception {
                            ch.pipeline().addLast(
                                    new EchoClientHandler());
                        }
                    });
            //同步式的鏈接
            ChannelFuture f = b.connect().sync();
            f.channel().closeFuture().sync();
        } finally {
            group.shutdownGracefully().sync();
        }
    }
    public static void main(String[] args) throws Exception {
        new EchoClient("localhost", 8155).start();
    }
}

服務(wù)端 自定義channelHandler

import io.netty.buffer.ByteBuf;
import io.netty.buffer.Unpooled;
import io.netty.channel.*;
import io.netty.util.CharsetUtil;

/**
 * 因為你的Echo 服務(wù)器會響應(yīng)傳入的消息，所以它需要實現(xiàn)ChannelInboundHandler 接口，用
 * 來定義響應(yīng)入站事件的方法。
 */

//標(biāo)示一個ChannelHandler 可以被多個Channel 安全地共享
@ChannelHandler.Sharable
public class EchoServerHandler extends ChannelInboundHandlerAdapter {
    //對于每個傳入的消息都會被調(diào)用；
    @Override
    public void channelRead(ChannelHandlerContext ctx, Object msg) {
        //將消息記錄到控制臺
        ByteBuf in = (ByteBuf) msg;
        System.out.println(
                "Server received: " + in.toString(CharsetUtil.UTF_8));
        ctx.write(in);
    }
    //通知ChannelInboundHandler最后一次對channelRead()
    //的調(diào)用是當(dāng)前批量讀取中的最后一條消息；
    @Override
    public void channelReadComplete(ChannelHandlerContext ctx) {
        ctx.writeAndFlush(Unpooled.EMPTY_BUFFER)
                .addListener(ChannelFutureListener.CLOSE);
    }
    //在讀取操作期間，有異常拋出時會調(diào)用。
    @Override
    public void exceptionCaught(ChannelHandlerContext ctx,
                                Throwable cause) {
        cause.printStackTrace();
        ctx.close();
    }
}

服務(wù)端實例

import io.netty.bootstrap.ServerBootstrap;
import io.netty.channel.*;
import io.netty.channel.nio.*;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioServerSocketChannel;

import java.net.InetSocketAddress;

public class EchoServer {
    private final int port;

    public EchoServer(int port) {
        this.port = port;
    }

    public static void main(String[] args) throws Exception {
        //設(shè)置端口值（如果端口參數(shù)的格式不正確，則拋出一個NumberFormatException）
        int port = 8155;
        new EchoServer(port).start();
    }

    public void start() throws Exception {
        //定義EventLoop
        EventLoopGroup group = new NioEventLoopGroup();
        try {
            //與Bootstrap類包包含豐富的客戶端API一樣，ServerBootstrap能夠非常方便的實現(xiàn)典型的服務(wù)端。
            ServerBootstrap b = new ServerBootstrap();
            b.group(group)
                    //指定所使用的NIO傳輸Channel
                    .channel(NioServerSocketChannel.class)
                    //使用指定的端口設(shè)置套接字地址
                    .localAddress(new InetSocketAddress(port))
                    //添加一個EchoServerHandler 到子Channel的ChannelPipeline
                    .childHandler(new ChannelInitializer() {
                        @Override
                        public void initChannel(SocketChannel ch)
                                throws Exception {
                            ch.pipeline().addLast(new EchoServerHandler());
                        }
                    });
            //新建一個future實例,異步地綁定服務(wù)器；調(diào)用sync()方法阻塞等待直到綁定完成
            ChannelFuture f = b.bind().sync();
            //獲取Channel 的CloseFuture，并且阻塞當(dāng)前線程直到它完成
            //該應(yīng)用程序?qū)枞却钡椒?wù)器的Channel關(guān)閉（因為你在Channel 的CloseFuture 上調(diào)用了sync()方法）。
            f.channel().closeFuture().sync();
        } finally {
            //關(guān)閉EventLoopGroup，釋放所有的資源
            group.shutdownGracefully().sync();
        }
    }
}

小結(jié)

在本章中，我們從技術(shù)和體系結(jié)構(gòu)這兩個角度探討了理解Netty 的重要性。我們也更加詳細(xì)地重新審視了之前引入的一些概念和組件，特別是ChannelHandler、ChannelPipeline和引導(dǎo)。

特別地，我們討論了ChannelHandler 類的層次結(jié)構(gòu)，并介紹了編碼器和解碼器，描述了它們在數(shù)據(jù)和網(wǎng)絡(luò)字節(jié)格式之間來回轉(zhuǎn)換的互補功能。下面的許多章節(jié)都將致力于深入研究這些組件，而這里所呈現(xiàn)的概覽應(yīng)該有助于你對整體
的把控。