Java網路編程--Netty入門

2019 年 10 月 7 日
筆記

Netty簡介

Netty是一個高性能，高可擴展性的非同步事件驅動的網路應用程式框架，它極大的簡化了TCP和UDP客戶端和伺服器端網路開發。它是一個NIO框架，對Java NIO進行了良好的封裝。作為一個非同步NIO框架，Netty的所有IO操作都是非同步非阻塞的，通過Future-Listener機制，用戶可以方便的主動獲取或者通過通知機制獲得IO操作結果。

Netty的特性

統一的API，適用於不同的協議
基於靈活、可擴展的事件驅動模型
高度可訂製的執行緒模型
更好的吞吐量，低延遲
更省資源，盡量減少不必要的記憶體拷貝
完整的SSL/TLS和STARTTLS的支援
能在Applet與Android的限制環境運行良好
不再因過快、過慢或超負載連接導致OutOfMemoryError
不再有在高速網路環境下NIO讀寫頻率不一致的問題

Netty核心內容

Netty中最核心的內容主要有以下四個方面：

Reactor執行緒模型：一種高性能的多執行緒程式設計思路
Netty中自己定義的Channel概念：增強版的通道概念
ChannelPipeline職責鏈設計模式：事件處理機制
記憶體管理：增強的ByteBuf緩衝區

Netty整體結構圖

Netty核心組件

EventLoop：EventLoop維護了一個執行緒和任務隊列，支援非同步提交執行任務。EventLoop自身實現了Executor介面，當調用executor方法提交任務時，則判斷是否啟動，未啟動則調用內置的executor創建新執行緒來觸發run方法執行，其大致流程參考Netty源碼SingleThreadEventExecutor如下：

EventLoopGroup：EventLoopGroup主要是管理eventLoop的生命周期，可以將其看作是一個執行緒池，其內部維護了一組EventLoop，每個eventLoop對應處理多個Channel，而一個Channel只能對應一個EventLoop

Bootstrap：BootStrap 是客戶端的引導類，主要用於客戶端連接遠程主機，有1個EventLoopGroup。Bootstrap 在調用 bind()（連接UDP）和 connect()（連接TCP）方法時，會新創建一個單獨的、沒有父 Channel 的 Channel 來實現所有的網路交換。

ServerBootstrap：ServerBootstrap 是服務端的引導類，主要用戶服務端綁定本地埠，有2個EventLoopGroup。ServerBootstarp 在調用 bind() 方法時會創建一個 ServerChannel 來接受來自客戶端的連接，並且該 ServerChannel 管理了多個子 Channel 用於同客戶端之間的通訊。

Channel：Netty中的Channel是一個抽象的概念，可以理解為對Java NIO Channel的增強和擴展，增加了許多新的屬性和方法，如bing方法等。

ChannelFuture：ChannelFuture能夠註冊一個或者多個ChannelFutureListener 實例，當操作完成時，不管成功還是失敗，均會被通知。ChannelFuture存儲了之後執行的操作的結果並且無法預測操作何時被執行，提交至Channel的操作按照被喚醒的順序被執行。

ChannelHandler：ChannelHandler用來處理業務邏輯，分別有入站和出站的實現。

ChannelPipeline：ChannelPipeline 提供了 ChannelHandler鏈的容器，並定義了用於在該鏈上傳播入站和出站事件流的API。

Netty執行緒模型

Netty的執行緒模型是基於Reactor模式的執行緒實現。關於Reactor模式可以參考Reactor模式，Netty中依據用戶的配置可以支援單執行緒的Reactor模型，多執行緒的Reactor模型以及主從多Reactor的模型。在Netty中其大致流程如下如下：

Netty入門程式碼示例

服務端程式碼示例：

import io.netty.bootstrap.ServerBootstrap;  import io.netty.buffer.ByteBuf;  import io.netty.buffer.Unpooled;  import io.netty.channel.*;  import io.netty.channel.nio.NioEventLoopGroup;  import io.netty.channel.socket.SocketChannel;  import io.netty.channel.socket.nio.NioServerSocketChannel;  import io.netty.handler.logging.LogLevel;  import io.netty.handler.logging.LoggingHandler;  import java.nio.charset.Charset;  public class EchoServer {    public static void main(String[] args) {        // accept執行緒組，用來接受連接      EventLoopGroup bossGroup = new NioEventLoopGroup(1);        // I/O執行緒組， 用於處理業務邏輯      EventLoopGroup workerGroup = new NioEventLoopGroup(1);        try {        // 服務端啟動引導        ServerBootstrap b = new ServerBootstrap();          b.group(bossGroup, workerGroup) // 綁定兩個執行緒組            .channel(NioServerSocketChannel.class) // 指定通道類型            .option(ChannelOption.SO_BACKLOG, 100) // 設置TCP連接的緩衝區            .handler(new LoggingHandler(LogLevel.INFO)) // 設置日誌級別            .childHandler(                new ChannelInitializer<SocketChannel>() {                  @Override                  protected void initChannel(SocketChannel socketChannel) throws Exception {                    ChannelPipeline pipeline = socketChannel.pipeline(); // 獲取處理器鏈                    pipeline.addLast(new EchoServerHandler()); // 添加新的件處理器                    }                  });          // 通過bind啟動服務        ChannelFuture f = b.bind(8080).sync();          // 阻塞主執行緒，知道網路服務被關閉        f.channel().closeFuture().sync();        } catch (Exception e) {        e.printStackTrace();      } finally {        workerGroup.shutdownGracefully();        bossGroup.shutdownGracefully();      }    }    }    class EchoServerHandler extends ChannelInboundHandlerAdapter {      // 每當從客戶端收到新的數據時，這個方法會在收到消息時被調用    @Override    public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {      System.out.println("收到數據：" + ((ByteBuf) msg).toString(Charset.defaultCharset()));      ctx.write(Unpooled.wrappedBuffer("Server message".getBytes()));      ctx.fireChannelRead(msg);    }    // 數據讀取完後被調用  @Override    public void channelReadComplete(ChannelHandlerContext ctx) throws Exception {      ctx.flush();    }    // 當Netty由於IO錯誤或者處理器在處理事件時拋出的異常時被調用  @Override    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {      cause.printStackTrace();      ctx.close();    }  }

客戶端程式碼示例：

import io.netty.bootstrap.Bootstrap;  import io.netty.buffer.ByteBuf;  import io.netty.buffer.Unpooled;  import io.netty.channel.*;  import io.netty.channel.nio.NioEventLoopGroup;  import io.netty.channel.socket.SocketChannel;  import io.netty.channel.socket.nio.NioSocketChannel;  import java.nio.charset.Charset;  public class EchoClient {    public static void main(String[] args) {      EventLoopGroup group = new NioEventLoopGroup();      try {        Bootstrap b = new Bootstrap();        b.group(group)            .channel(NioSocketChannel.class)            .option(ChannelOption.TCP_NODELAY, true)            .handler(                new ChannelInitializer<SocketChannel>() {                  @Override                  public void initChannel(SocketChannel ch) throws Exception {                    ChannelPipeline p = ch.pipeline();                    p.addLast(new EchoClientHandler());                  }                });        ChannelFuture f = b.connect("127.0.0.1", 8080).sync();        f.channel().closeFuture().sync();      } catch (Exception e) {        e.printStackTrace();      } finally {        group.shutdownGracefully();      }    }    }    class EchoClientHandler extends ChannelInboundHandlerAdapter {    private final ByteBuf firstMessage;    public EchoClientHandler() {      firstMessage = Unpooled.buffer(256);      for (int i = 0; i < firstMessage.capacity(); i++) {        firstMessage.writeByte((byte) i);      }    }    @Override    public void channelActive(ChannelHandlerContext ctx) {      ctx.writeAndFlush(firstMessage);    }    @Override    public void channelRead(ChannelHandlerContext ctx, Object msg) {      System.out.println("收到數據：" + ((ByteBuf) msg).toString(Charset.defaultCharset()));      ctx.write(Unpooled.wrappedBuffer("Client message".getBytes()));    }    @Override    public void channelReadComplete(ChannelHandlerContext ctx) {      ctx.flush();    }    @Override    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) {      cause.printStackTrace();      ctx.close();    }  }

Tags: