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模式](2.1.3 Reactor模式.md) ，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();
  }
}

相關推薦