Netty學(xué)習(xí)筆記（一）

lncwwn 發(fā)布于2019-08-14 16:59 / 740人閱讀

摘要：是一個(gè)高性能事件驅(qū)動(dòng)的異步的非堵塞的框架，用于建立等底層的連接，基于可以建立高性能的服務(wù)器。在中提供了兩套，一套是針對(duì)標(biāo)準(zhǔn)輸入輸出，另一套就是網(wǎng)絡(luò)編程。和是標(biāo)準(zhǔn)中的核心對(duì)象是對(duì)原中流的模擬，任何來(lái)源和目的數(shù)據(jù)都必須通過(guò)一個(gè)對(duì)象。

Netty是一個(gè)高性能事件驅(qū)動(dòng)的異步的非堵塞的IO(NIO)框架，用于建立TCP等底層的連接，基于Netty可以建立高性能的Http服務(wù)器。
1、首先來(lái)復(fù)習(xí)下非堵塞IO(NIO)
NIO這個(gè)庫(kù)是在JDK1.4中才引入的。NIO和IO有相同的作用和目的，但實(shí)現(xiàn)方式不同，NIO主要用到的是塊，所以NIO的效率要比IO高很多。
在Java API中提供了兩套NIO，一套是針對(duì)標(biāo)準(zhǔn)輸入輸出NIO，另一套就是網(wǎng)絡(luò)編程N(yùn)IO。
*Buffer和Channel是標(biāo)準(zhǔn)NIO中的核心對(duì)象*
Channel是對(duì)原IO中流的模擬，任何來(lái)源和目的數(shù)據(jù)都必須通過(guò)一個(gè)Channel對(duì)象。一個(gè)Buffer實(shí)質(zhì)上是一個(gè)容器對(duì)象，發(fā)給Channel的所有對(duì)象都必須先放到Buffer中；同樣的，從Channel中讀取的任何數(shù)據(jù)都要讀到Buffer中。

網(wǎng)絡(luò)編程N(yùn)IO中還有一個(gè)核心對(duì)象Selector，它可以注冊(cè)到很多個(gè)Channel上，監(jiān)聽(tīng)各個(gè)Channel上發(fā)生的事件，并且能夠根據(jù)事件情況決定Channel讀寫(xiě)。這樣，通過(guò)一個(gè)線程管理多個(gè)Channel，就可以處理大量網(wǎng)絡(luò)連接了。

Selector 就是注冊(cè)對(duì)各種 I/O 事件興趣的地方，而且當(dāng)那些事件發(fā)生時(shí)，就是這個(gè)對(duì)象告訴你所發(fā)生的事件。
Selector selector = Selector.open(); //創(chuàng)建一個(gè)selector
為了能讓Channel和Selector配合使用，我們需要把Channel注冊(cè)到Selector上。通過(guò)調(diào)用channel.register（）方法來(lái)實(shí)現(xiàn)注冊(cè)：
channel.configureBlocking(false); //設(shè)置成異步IO
SelectionKey key =channel.register(selector,SelectionKey.OP_READ); //對(duì)所關(guān)心的事件進(jìn)行注冊(cè)（connet,accept,read,write）
SelectionKey 代表這個(gè)通道在此 Selector 上的這個(gè)注冊(cè)。

2、異步
CallBack：回調(diào)是異步處理經(jīng)常用到的編程模式，回調(diào)函數(shù)通常被綁定到一個(gè)方法上，并且在方法完成之后才執(zhí)行，這種處理方式在javascript當(dāng)中得到了充分的運(yùn)用。回調(diào)給我們帶來(lái)的難題是當(dāng)一個(gè)問(wèn)題處理過(guò)程中涉及很多回調(diào)時(shí)，代碼是很難讀的。
Futures：Futures是一種抽象，它代表一個(gè)事情的執(zhí)行過(guò)程中的一些關(guān)鍵點(diǎn)，我們通過(guò)Future就可以知道任務(wù)的執(zhí)行情況，比如當(dāng)任務(wù)沒(méi)完成時(shí)我們可以做一些其它事情。它給我們帶來(lái)的難題是我們需要去判斷future的值來(lái)確定任務(wù)的執(zhí)行狀態(tài)。
3、netty到底怎么工作的呢？
直接來(lái)看個(gè)最簡(jiǎn)單的實(shí)例

服務(wù)器端

public class EchoServer {
 
 private final static int port = 8007;
 
 public void start() throws InterruptedException{
  ServerBootstrap bootstrap = new ServerBootstrap(); //引導(dǎo)輔助程序
  EventLoopGroup group = new NioEventLoopGroup(); //通過(guò)nio的方式接受連接和處理連接
  try {
   bootstrap.group(group)
    .channel(NioServerSocketChannel.class) //設(shè)置nio類(lèi)型的channel
    .localAddress(new InetSocketAddress(port)) //設(shè)置監(jiān)聽(tīng)端口
    .childHandler(new ChannelInitializer() { //有連接到達(dá)時(shí)會(huì)創(chuàng)建一個(gè)channel
     // pipline 管理channel中的handler,在channel隊(duì)列中添加一個(gè)handler來(lái)處理業(yè)務(wù)
     @Override
     protected void initChannel(SocketChannel ch) throws Exception {
      ch.pipeline().addLast("myHandler", new EchoServerHandler());
      //ch.pipeline().addLast("idleStateHandler",new? IdleStateHandler(0, 0, 180));
      
     }
    });
   ChannelFuture future = bootstrap.bind().sync(); //配置完成，綁定server，并通過(guò)sync同步方法阻塞直到綁定成功
   System.out.println(EchoServer.class.getName() + " started and listen on " + future.channel().localAddress());
   future.channel().closeFuture().sync(); //應(yīng)用程序會(huì)一直等待，直到channel關(guān)閉
  } catch (Exception e) {
   e.getMessage();
  }finally {
   group.shutdownGracefully().sync();
  }
 }

創(chuàng)建一個(gè)ServerBootstrap實(shí)例

創(chuàng)建一個(gè)EventLoopGroup來(lái)處理各種事件，如處理鏈接請(qǐng)求，發(fā)送接收數(shù)據(jù)等。

定義本地InetSocketAddress( port)好讓Server綁定

創(chuàng)建childHandler來(lái)處理每一個(gè)鏈接請(qǐng)求

所有準(zhǔn)備好之后調(diào)用ServerBootstrap.bind()方法綁定Server

handler 處理核心業(yè)務(wù)

@Sharable //注解@Sharable可以讓它在channels間共享??
public class EchoServerHandler extends ChannelInboundHandlerAdapter{
 
 @Override
 public void channelRead(ChannelHandlerContext ctx,Object msg) throws Exception {
  ByteBuf buf = ctx.alloc().buffer();
  buf.writeBytes("Hello World".getBytes());
  ctx.write(buf);
 }
 
 @Override
 public void channelReadComplete(ChannelHandlerContext ctx) throws Exception {
  ctx.flush();
 }
 
 @Override
 public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
  cause.printStackTrace();
  ctx.close();
 }
}

客戶端連接

public class EchoClient {

 private final int port;
 private final String hostIp;

 public EchoClient(int port, String hostIp) {
  this.port = port;
  this.hostIp = hostIp;
 }

 public void start() throws InterruptedException {
  Bootstrap bootstrap = new Bootstrap();
  EventLoopGroup group = new NioEventLoopGroup();
  try {
   bootstrap.group(group).channel(NioSocketChannel.class)
     .remoteAddress(new InetSocketAddress(hostIp, port))
     .handler(new ChannelInitializer() {
      @Override
      protected void initChannel(SocketChannel ch) throws Exception {
       ch.pipeline().addLast(new EchoClientHandler());
      }
     });
   ChannelFuture future = bootstrap.connect().sync();
   future.addListener(new ChannelFutureListener() {

    public void operationComplete(ChannelFuture future) throws Exception {
     if (future.isSuccess()) {
      System.out.println("client connected");
     } else {
      System.out.println("server attemp failed");
      future.cause().printStackTrace();
     }

    }
   });
   future.channel().closeFuture().sync();
  } catch (InterruptedException e) {
   e.printStackTrace();
  } finally {
   group.shutdownGracefully().sync();
  }
 }

客戶端handler

@Sharable
public class EchoClientHandler extends SimpleChannelInboundHandler{
 
 /**?
?? ? *此方法會(huì)在連接到服務(wù)器后被調(diào)用 ?
?? ? * */ ?
? ? public void channelActive(ChannelHandlerContext ctx) {?
? ?  System.out.println("Netty rocks!");
? ? ? ? ctx.write(Unpooled.copiedBuffer("Netty rocks!", CharsetUtil.UTF_8)); ?
? ? } ?
 
 /**
  * 接收到服務(wù)器數(shù)據(jù)時(shí)調(diào)用
  */
 @Override
 protected void channelRead0(ChannelHandlerContext ctx, ByteBuf msg) throws Exception {
  System.out.println("Client received: " + ByteBufUtil.hexDump(msg.readBytes(msg.readableBytes()))); ?
 }
 
  /**?
?? ? *捕捉到異常 ?
?? ? * */ ?
? ? public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) { ?
? ? ? ? cause.printStackTrace(); ?
? ? ? ? ctx.close(); ?
? ? } ?
 
}