如何進行Java--Netty的入門

這篇文章將為大家詳細講解有關(guān)如何進行Java--Netty的入門，文章內(nèi)容質(zhì)量較高，因此小編分享給大家做個參考，希望大家閱讀完這篇文章后對相關(guān)知識有一定的了解。

Netty簡介

Netty是一個高性能，高可擴展性的異步事件驅(qū)動的網(wǎng)絡(luò)應(yīng)用程序框架，它極大的簡化了TCP和UDP客戶端和服務(wù)器端網(wǎng)絡(luò)開發(fā)。它是一個NIO框架，對Java NIO進行了良好的封裝。作為一個異步NIO框架，Netty的所有IO操作都是異步非阻塞的，通過Future-Listener機制，用戶可以方便的主動獲取或者通過通知機制獲得IO操作結(jié)果。

Netty的特性

• 統(tǒng)一的API，適用于不同的協(xié)議

• 基于靈活、可擴展的事件驅(qū)動模型

• 高度可定制的線程模型

• 更好的吞吐量，低延遲

• 更省資源，盡量減少不必要的內(nèi)存拷貝

• 完整的SSL/TLS和STARTTLS的支持

• 能在Applet與Android的限制環(huán)境運行良好

• 不再因過快、過慢或超負載連接導(dǎo)致OutOfMemoryError

• 不再有在高速網(wǎng)絡(luò)環(huán)境下NIO讀寫頻率不一致的問題

Netty核心內(nèi)容

Netty中最核心的內(nèi)容主要有以下四個方面：

• Reactor線程模型：一種高性能的多線程程序設(shè)計思路

• Netty中自己定義的Channel概念：增強版的通道概念

• ChannelPipeline職責(zé)鏈設(shè)計模式：事件處理機制

• 內(nèi)存管理：增強的ByteBuf緩沖區(qū)

Netty整體結(jié)構(gòu)圖

Netty核心組件

EventLoop：EventLoop維護了一個線程和任務(wù)隊列，支持異步提交執(zhí)行任務(wù)。EventLoop自身實現(xiàn)了Executor接口，當調(diào)用executor方法提交任務(wù)時，則判斷是否啟動，未啟動則調(diào)用內(nèi)置的executor創(chuàng)建新線程來觸發(fā)run方法執(zhí)行，其大致流程參考Netty源碼SingleThreadEventExecutor如下：

EventLoopGroup：EventLoopGroup主要是管理eventLoop的生命周期，可以將其看作是一個線程池，其內(nèi)部維護了一組EventLoop，每個eventLoop對應(yīng)處理多個Channel，而一個Channel只能對應(yīng)一個EventLoop

Bootstrap：BootStrap 是客戶端的引導(dǎo)類，主要用于客戶端連接遠程主機，有1個EventLoopGroup。Bootstrap 在調(diào)用 bind()（連接UDP）和 connect()（連接TCP）方法時，會新創(chuàng)建一個單獨的、沒有父 Channel 的 Channel 來實現(xiàn)所有的網(wǎng)絡(luò)交換。

ServerBootstrap：ServerBootstrap 是服務(wù)端的引導(dǎo)類，主要用戶服務(wù)端綁定本地端口，有2個EventLoopGroup。ServerBootstarp 在調(diào)用 bind() 方法時會創(chuàng)建一個 ServerChannel 來接受來自客戶端的連接，并且該 ServerChannel 管理了多個子 Channel 用于同客戶端之間的通信。

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

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

ChannelHandler：ChannelHandler用來處理業(yè)務(wù)邏輯，分別有入站和出站的實現(xiàn)。

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

Netty線程模型

Netty的線程模型是基于Reactor模式的線程實現(xiàn)。關(guān)于Reactor模式可以參考Reactor模式 ，Netty中依據(jù)用戶的配置可以支持單線程的Reactor模型，多線程的Reactor模型以及主從多Reactor的模型。在Netty中其大致流程如下如下：

Netty入門代碼示例

服務(wù)端代碼示例：

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線程組， 用于處理業(yè)務(wù)邏輯
    EventLoopGroup workerGroup = new NioEventLoopGroup(1);

    try {
      // 服務(wù)端啟動引導(dǎo)
      ServerBootstrap b = new ServerBootstrap();

      b.group(bossGroup, workerGroup) // 綁定兩個線程組
          .channel(NioServerSocketChannel.class) // 指定通道類型
          .option(ChannelOption.SO_BACKLOG, 100) // 設(shè)置TCP連接的緩沖區(qū)
          .handler(new LoggingHandler(LogLevel.INFO)) // 設(shè)置日志級別
          .childHandler(
              new ChannelInitializer<SocketChannel>() {
                @Override
                protected void initChannel(SocketChannel socketChannel) throws Exception {
                  ChannelPipeline pipeline = socketChannel.pipeline(); // 獲取處理器鏈
                  pipeline.addLast(new EchoServerHandler()); // 添加新的件處理器

                }

              });

      // 通過bind啟動服務(wù)
      ChannelFuture f = b.bind(8080).sync();

      // 阻塞主線程，知道網(wǎng)絡(luò)服務(wù)被關(guān)閉
      f.channel().closeFuture().sync();

    } catch (Exception e) {
      e.printStackTrace();
    } finally {
      workerGroup.shutdownGracefully();
      bossGroup.shutdownGracefully();
    }
  }

}

class EchoServerHandler extends ChannelInboundHandlerAdapter {

  // 每當從客戶端收到新的數(shù)據(jù)時，這個方法會在收到消息時被調(diào)用
  @Override
  public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
    System.out.println("收到數(shù)據(jù)：" + ((ByteBuf) msg).toString(Charset.defaultCharset()));
    ctx.write(Unpooled.wrappedBuffer("Server message".getBytes()));
    ctx.fireChannelRead(msg);
  }
  // 數(shù)據(jù)讀取完后被調(diào)用  @Override
  public void channelReadComplete(ChannelHandlerContext ctx) throws Exception {
    ctx.flush();
  }
  // 當Netty由于IO錯誤或者處理器在處理事件時拋出的異常時被調(diào)用  @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("收到數(shù)據(jù)：" + ((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();
  }
}

關(guān)于如何進行Java--Netty的入門就分享到這里了，希望以上內(nèi)容可以對大家有一定的幫助，可以學(xué)到更多知識。如果覺得文章不錯，可以把它分享出去讓更多的人看到。