從netty-example分析Netty組件

本文轉載自查看原文 2015-12-14 19:16 15688 高並發及性能

分析netty從源碼開始

准備工作：

1.下載源代碼：https://github.com/netty/netty.git

我下載的版本為4.1

2. eclipse導入maven工程。

netty提供了一個netty-example工程，

分類如下：

Fundamental

Echo ‐ the very basic client and server
Discard ‐ see how to send an infinite data stream asynchronously without flooding the write buffer
Uptime ‐ implement automatic reconnection mechanism

Text protocols

Telnet ‐ a classic line-based network application
Quote of the Moment ‐ broadcast a UDP/IP packet
SecureChat ‐ an TLS-based chat server, derived from the Telnet example

Binary protocols

ObjectEcho ‐ exchange serializable Java objects
Factorial ‐ write a stateful client and server with a custom binary protocol
WorldClock ‐ rapid protocol protyping with Google Protocol Buffers integration

HTTP

Snoop ‐ build your own extremely light-weight HTTP client and server
File server ‐ asynchronous large file streaming in HTTP
Web Sockets (Client & Server) ‐ add a two-way full-duplex communication channel to HTTP using Web Sockets
SPDY (Client & Server) ‐ implement SPDY protocol
CORS demo ‐ implement cross-origin resource sharing

Advanced

Proxy server ‐ write a highly efficient tunneling proxy server
Port unification ‐ run services with different protocols on a single TCP/IP port

UDT

Byte streams ‐ use UDT in TCP-like byte streaming mode
Message flow ‐ use UDT in UDP-like message delivery mode
Byte streams in symmetric peer-to-peer rendezvous connect mode
Message flow in symmetric peer-to-peer rendezvous connect mode

我們的分析從這里開始，netty是client-server形式的，我們以最簡單的discard示例開始，其服務器端代碼如下：

/**
 * Discards any incoming data.
 */
public final class DiscardServer {

    static final boolean SSL = System.getProperty("ssl") != null;
    static final int PORT = Integer.parseInt(System.getProperty("port", "8009"));

    public static void main(String[] args) throws Exception {
        // Configure SSL.
        final SslContext sslCtx;
        if (SSL) {
            SelfSignedCertificate ssc = new SelfSignedCertificate();
            sslCtx = SslContextBuilder.forServer(ssc.certificate(), ssc.privateKey()).build();
        } else {
            sslCtx = null;
        }

        EventLoopGroup bossGroup = new NioEventLoopGroup(1);
        EventLoopGroup workerGroup = new NioEventLoopGroup();
        try {
            ServerBootstrap b = new ServerBootstrap();
            b.group(bossGroup, workerGroup)
             .channel(NioServerSocketChannel.class)
             .handler(new LoggingHandler(LogLevel.INFO))
             .childHandler(new ChannelInitializer<SocketChannel>() {
                 @Override
                 public void initChannel(SocketChannel ch) {
                     ChannelPipeline p = ch.pipeline();
                     if (sslCtx != null) {
                         p.addLast(sslCtx.newHandler(ch.alloc()));
                     }
                     p.addLast(new DiscardServerHandler());
                 }
             });

            // Bind and start to accept incoming connections.
            ChannelFuture f = b.bind(PORT).sync();

            // Wait until the server socket is closed.
            // In this example, this does not happen, but you can do that to gracefully
            // shut down your server.
            f.channel().closeFuture().sync();
        } finally {
            workerGroup.shutdownGracefully();
            bossGroup.shutdownGracefully();
        }
    }
}

上面的代碼中使用了下面幾個類：

1. EventLoopGroup

實現類為NioEventLoopGroup，其層次結構為：

　　EventExecutorGroup為所有類的父接口，它通過next()方法來提供EventExecutor供使用。除此以外，它還負責處理它們的生命周期，允許以優雅的方式關閉。

　　EventExecutor是一種特殊的EventExcutorGroup，它提供了一些便利的方法來查看一個線程是否在一個事件循環中執行過，除此以外，它也擴展了EventExecutorGroup，從而提供了一個通用的獲取方法的方式。

　　EventLoopGroup是一種特殊的EventExcutorGroup，它提供了channel的注冊功能，channel在事件循環中將被后面的selection來獲取到。

　　NioEventLoopGroup繼承自MultithreadEventLoopGroup，基於channel的NIO selector會使用該類。

2.ServerBootstrap使ServerChannel容易自舉。

　 group(EventLoopGroup parentGroup, EventLoopGroup childGroup)方法設置父EventLoopGroup和子EventLoopGroup。這些EventLoopGroup用來處理所有的事件和ServerChannel和Channel的IO。

channel(Class<? extends C> channelClass)方法用來創建一個Channel實例。創建Channel實例要不使用此方法，如果channel實現是無參構造要么可以使用channelFactory來創建。

　　handler(ChannelHandler handler)方法，channelHandler用來處理請求的。

　　childHandler(ChannelHandler childHandler)方法，設置用來處理請求的channelHandler。

3. ChannelInitializer一種特殊的ChannelInboundHandler

　　當Channel注冊到它的eventLoop中時，ChannelInitializer提供了一個方便初始化channel的方法。該類的實現通常用來設置ChannelPipeline的channel，通常使用在Bootstrap#handler(ChannelHandler),ServerBootstrap#handler(ChannelHandler)和ServerBootstrap#childHandler(ChannelHandler)三個場景中。示例：

 public class MyChannelInitializer extends ChannelInitializer{
      public void initChannel({@link Channel} channel) {
          channel.pipeline().addLast("myHandler", new MyHandler());
      }
  }

然后：

ServerBootstrap bootstrap = ...;
...
bootstrap.childHandler(new MyChannelInitializer());

注意：這個類標示為可共享的，因此實現類重用時需要時安全的。

4. ChannelPipeline相關

　　理解ChannelPipeline需要先理解ChannelHandler，

4.1 ChannelHandler

處理一個IO事件或者翻譯一個IO操作，並且傳遞給ChannelPineline的下一個handler。

　　你可以使用ChannelHandlerAdapter來替代它

　　因為這個接口有太多接口需要實現，因此你只有實現ChannelHandlerAdapter就可以代替實現這個接口。

　　Context對象

　　ChannelHandlerContext封裝了ChannelHandler。ChannelHandler應該通過context對象與它所屬的ChannelPipeLine進行交互。通過使用context對象，ChannelHandler可以傳遞上行或者下行事件，或者動態修改pipeline，或者存儲特定handler的信息(使用AttributeKey)。

　　狀態管理

　　一個channelHandler通常需要存儲一些狀態信息。最簡單最值得推薦的方法是使用member變量：

 public interface Message {
      // your methods here
  }
 
  public class DataServerHandler extends  SimpleChannelInboundHandler<Message> {
 
      private boolean loggedIn;
 
      {@code @Override}
      protected void messageReceived( ChannelHandlerContext ctx, Message message) {
           Channel ch = e.getChannel();
          if (message instanceof LoginMessage) {
              authenticate((LoginMessage) message);
              loggedIn = true;
          } else (message instanceof GetDataMessage) {
              if (loggedIn) {
                  ch.write(fetchSecret((GetDataMessage) message));
              } else {
                  fail();
              }
          }
      }
      ...
  }

注意：handler的狀態附在ChannePipelineContext上，因此可以增加相同的handler實例到不同的pipeline上：

  public class DataServerInitializer extends ChannelInitializer<Channel> {
 
      private static final DataServerHandler SHARED = new DataServerHandler();
 
      @Override
      public void initChannel(Channel channel) {
          channel.pipeline().addLast("handler", SHARED);
      }
  }

@Sharable注解

　　在上面的示例中，使用了一個AttributeKey，你可能注意到了@Sharable注解。

　　如果一個ChannelHandler使用@sharable進行注解，那就意味着你僅僅創建了一個handler一次，可以添加到一個或者多個ChannelPipeline多次而不會產生競爭。

　　如果沒有指定該注解，你必須每次都創建一個新的handler實例，並且增加到一個ChannelPipeline，因為它沒有像member變量一樣，它有一個非共享的狀態。

4.2 ChannelPipeline

ChanelPipeline是一組ChanelHandler的集合，它處理或者解析Channel的Inbound事件和OutBound操作。ChannelPipeline的實現是Intercepting Filter的一種高級形式，這樣用戶可以控制事件如何處理，一個pipeline內部ChannelHandler如何交互。

　　pipeline事件流程

　　上圖描述了IO事件如何被一個ChannelPipeline的ChannelHandler處理的。一個IO事件被一個ChannelInBoundHandler處理或者ChannelOutboundHandler，然后通過調用在ChannelHandlerContext中定義的事件傳播方法傳遞給最近的handler，傳播方法有ChannelHandlerContext#filreChannelRead(Object)和ChannelHandlerContext#write(Object)。

　　一個Inbound事件通常由Inbound handler來處理，如上如左上。一個Inbound handler通常處理在上圖底部IO線程產生的Inbound數據。Inbound數據通過真實的輸入操作如SocketChannel#read(ByteBuffer)來獲取。如果一個inbound事件越過了最上面的inbound handler，該事件將會被拋棄到而不會通知你或者如果你需要關注，打印出日志。

　　一個outbound事件由上圖的右下的outbound handler來處理。一個outbound handler通常由outbound流量如寫請求產生或者轉變的。如果一個outbound事件越過了底部的outbound handler，它將由channel關聯的IO線程處理。IO線程通常運行的是真實的輸出操作如SocketChannel#write(byteBuffer).

示例，假設我們創建下面這樣一個pipeline：

 ChannelPipeline} p = ...;
 p.addLast("1", new InboundHandlerA());
 p.addLast("2", new InboundHandlerB());
 p.addLast("3", new OutboundHandlerA());
 p.addLast("4", new OutboundHandlerB());
 p.addLast("5", new InboundOutboundHandlerX());

　　在上例中，inbound開頭的handler意味着它是一個inbound handler。outbound開頭的handler意味着它是一個outbound handler。上例的配置中當一個事件進入inbound時handler的順序是1,2,3,4,5;當一個事件進入outbound時，handler的順序是5,4,3,2,1.在這個最高准則下，ChannelPipeline跳過特定handler的處理來縮短stack的深度：

　　3,4沒有實現ChannelInboundHandler，因而一個inbound事件的處理順序是1,2,5.

　　1,2沒有實現ChannelOutBoundhandler，因而一個outbound事件的處理順序是5,4,3

　　若5同時實現了ChannelInboundHandler和channelOutBoundHandler，一個inbound和一個outbound事件的執行順序分別是125和543.

　　一個事件跳向下一個handler

　　如上圖所示，一個handler觸發ChannelHandlerContext中的事件傳播方法，然后傳遞到下一個handler。這些方法有：

　　inbound 事件傳播方法：

      ChannelHandlerContext#fireChannelRegistered()
      ChannelHandlerContext#fireChannelActive()
      ChannelHandlerContext#fireChannelRead(Object)
      ChannelHandlerContext#fireChannelReadComplete()
      ChannelHandlerContext#fireExceptionCaught(Throwable)
      ChannelHandlerContext#fireUserEventTriggered(Object)
      ChannelHandlerContext#fireChannelWritabilityChanged()
      ChannelHandlerContext#fireChannelInactive()
      ChannelHandlerContext#fireChannelUnregistered()

　　outbound事件傳播方法：

ChannelHandlerContext#bind(SocketAddress, ChannelPromise)
ChannelHandlerContext#connect(SocketAddress, SocketAddress, ChannelPromise)
ChannelHandlerContext#write(Object, ChannelPromise)
ChannelHandlerContext#flush()
ChannelHandlerContext#read()
ChannelHandlerContext#disconnect(ChannelPromise)
ChannelHandlerContext#close(ChannelPromise)
ChannelHandlerContext#deregister(ChannelPromise)

下面的示例展示了事件是如何傳播的：

  public class MyInboundHandler extends ChannelInboundHandlerAdapter {
      @Override
      public void channelActive(ChannelHandlerContext} ctx) {
          System.out.println("Connected!");
          ctx.fireChannelActive();
      }
  }
 
  public clas MyOutboundHandler extends ChannelOutboundHandlerAdapter {
      @Override
      public void close(ChannelHandlerContext} ctx, ChannelPromise} promise) {
          System.out.println("Closing ..");
          ctx.close(promise);
      }
  }

　　創建一個pipeline

　　在pipeline中，一個用戶一般由一個或者多個ChannelHandler來接收IO事件(例如讀)和IO操作請求(如寫或者close)。例如，一個典型的服務器pipeline通常具有以下幾個handler，但最多有多少handler取決於協議和業務邏輯的復雜度：

Protocol Decoder--將二進制數據(如ByteBuffer)轉換成一個java對象

Protocol Encoder--將一個java對象轉換成二進制數據。

Business Logic Handler--處理真實的業務邏輯(如數據庫訪問)。

讓我們用下面的示例展示：

 static final  EventExecutorGroup group = new  DefaultEventExecutorGroup(16);
  ...
 
   ChannelPipeline} pipeline = ch.pipeline();
 
  pipeline.addLast("decoder", new MyProtocolDecoder());
  pipeline.addLast("encoder", new MyProtocolEncoder());
 
  // Tell the pipeline to run MyBusinessLogicHandler's event handler methods
  // in a different thread than an I/O thread so that the I/O thread is not blocked by
  // a time-consuming task.
  // If your business logic is fully asynchronous or finished very quickly, you don't
  // need to specify a group.
  pipeline.addLast(group, "handler", new MyBusinessLogicHandler());

　　線程安全

　　因為ChannelPipeline是線程安全的，一個channelhandler可以在任意時間內增加或者刪除。例如，當有敏感信息交換時，你可以插入一個加密handler，然后當信息交換結束后刪除該handler。

4.3 Channel

Channel是網絡socket的一個紐帶或者一個處理IO操作如讀、寫、連接、綁定的組件。一個Channel提供如下信息：

　　　　當前channel的狀態，如它是否開啟？是否連接？

　　　　Channel的ChannelConfig的配置參數，如接受緩存大小；

　　　　channel支持的IO操作，如讀、寫、連接、綁定；

　　　　channel支持的ChannelPipeline，它處理所有的IO事件和channel關聯的請求。

　　所有的IO操作都是異步的。

　　在Netty中所有的IO操作都是異步的。這意味着所有的IO調用將立即返回，但不保證在調用結束時請求的IO操作都已經執行完畢。而是在請求操作處理完成、失敗或者取消時返回一個ChannelFuture來通知。

Channel是繼承性的。

　　一個Channel可以它如何創建的來獲取它的父Channel(#parent()方法)。例如：一個由ServerSocketChannel接受的SocketChannel調用parent()方法時返回ServerSocketChannel。

　　繼承的結構依賴於Channel的所屬transport實現。例如，你可以新寫一個Channel實現，它創建了一個共享同一個socket連接的子channel，如BEEP和SSH

　向下去獲取特定transport操作。

　　一些transport會暴露一些該transport特定的操作。Channel向下轉換到子類型可以觸發這些操作。例如：老的IO datargram transport，DatagramChannel提供了多播的join和leave操作。

　　釋放資源

　　當Channel處理完后，一定記得調用close()或者close(ChannelPromise)來釋放資源。

5. channelFuture

channelFuture是異步IO操作的返回值。

　　當一個IO操作開始時，創建一個新的future。ChannelFuture要么是uncompleted，要么是completed。新的future開始時是uncompleted---既不是成功、失敗，也不是取消，因為IO操作還沒有開始呢。若IO操作結束時future要么成功，要么失敗或者取消，標記為completed的future有更多特殊的意義，例如失敗的原因。請注意：即使失敗和取消也屬於completed狀態。

　　有很多方法可以查詢IO操作是否完成：等待完成，檢索IO操作的結果。同樣也允許你增加ChannelFutureListenner，這樣你可以在IO操作完成后獲得通知。

　　在盡可能的情況下，推薦addListenner()方法而不是await()方法，當IO操作完成后去完成接下來的其它任務時去獲取通知。

6.ChannelHandlerContext

　　對ChannelHandler相關信息的包裝。

小結

netty處理請求的總流程是經過ChannelPipeline中的多個ChannelHandler后，返回結果ChannelFuture。如下圖所示：

具體I/O操作調用的流程，
應用->Channel的I/O操作->調用Pipeline相應的I/O操作->調用ChannelHandlerContext的相應I/O操作->調用ChannelHandler的相應操作->Channel.UnSafe中相關的I/O操作。
應用為什么不直接調用Channel.UnSafe接口中的I/O操作呢，而要繞一個大圈呢？因為它是框架，要支持擴展。
執行者完成操作后，是如何通知命令者的呢？一般流程是這樣的:
Channel.UnSafe中執行相關的I/O操作，根據操作結果->調用ChannelPipeline中相應發fireXXXX()接口->調用ChannelHandlerContext中相應的fireXXXX()接口->調用ChannelHandler中相應方法->調用應用中的相關邏輯。

@Configuration
public class WebConfig extends WebMvcConfigurerAdapter {

@Override
public void configureContentNegotiation(ContentNegotiationConfigurer configurer) {
    configurer
            .defaultContentType(MediaType.APPLICATION_JSON)
            .favorPathExtension(true);
}

/*
 * Configure ContentNegotiatingViewResolver
 */
@Bean
public ViewResolver contentNegotiatingViewResolver(ContentNegotiationManager manager) {
    ContentNegotiatingViewResolver resolver = new ContentNegotiatingViewResolver();
    resolver.setContentNegotiationManager(manager);

    // Define all possible view resolvers
    List<ViewResolver> resolvers = new ArrayList<>();

    resolvers.add(csvViewResolver());
    resolvers.add(excelViewResolver());
    resolvers.add(pdfViewResolver());

    resolver.setViewResolvers(resolvers);
    return resolver;
}

/*
 * Configure View resolver to provide XLS output using Apache POI library to
 * generate XLS output for an object content
 */
@Bean
public ViewResolver excelViewResolver() {
    return new ExcelViewResolver();
}

/*
 * Configure View resolver to provide Csv output using Super Csv library to
 * generate Csv output for an object content
 */
@Bean
public ViewResolver csvViewResolver() {
    return new CsvViewResolver();
}

/*
 * Configure View resolver to provide Pdf output using iText library to
 * generate pdf output for an object content
 */
@Bean
public ViewResolver pdfViewResolver() {
    return new PdfViewResolver();
}
}

參考文獻：

【1】http://www.jiancool.com/article/71493268111/

【2】http://blog.csdn.net/pingnanlee/article/details/11973769

免責聲明！

本站轉載的文章為個人學習借鑒使用，本站對版權不負任何法律責任。如果侵犯了您的隱私權益，請聯系本站郵箱yoyou2525@163.com刪除。

猜您在找 從netty-example分析Netty組件續【Netty】Netty核心組件介紹 Netty專題(四)-----Netty介紹、Netty組件介紹 netty基本組件介紹 Netty 框架學習 —— Netty 組件與設計 Netty4.x分析 netty ByteToMessageDecoder 分析 netty ByteBuf分析【Netty】源碼分析目錄 Netty 源碼分析之ByteBuf