Netty组件源码分析

public class NettyServer {

    private static final String IP = "127.0.0.1";
    private static final int port = 8080;
    private static final int BIZGROUPSIZE = Runtime.getRuntime().availableProcessors() * 2;
    private static final int BIZTHREADSIZE = 100;

    private static final EventLoopGroup bossGroup = new NioEventLoopGroup(BIZGROUPSIZE);
    private static final EventLoopGroup workGroup = new NioEventLoopGroup(BIZTHREADSIZE);

    public static void main(String[] args) throws Exception {
        NettyServer.start();
    }
  
    public static void start() throws Exception {
        ServerBootstrap serverBootstrap = initServerBootstrap();
        ChannelFuture channelFuture = serverBootstrap.bind(IP, port).sync();
        channelFuture.channel().closeFuture().sync();
    }

    private static ServerBootstrap initServerBootstrap() {
        ServerBootstrap serverBootstrap = new ServerBootstrap();
        serverBootstrap.group(bossGroup, workGroup)
                .channel(NioServerSocketChannel.class)
                .childHandler(new ChannelInitializer<Channel>() {
                    @Override
                    protected void initChannel(Channel ch) {
                        ChannelPipeline pipeline = ch.pipeline();
                        pipeline.addLast(new LengthFieldBasedFrameDecoder(Integer.MAX_VALUE, 0, 4, 0, 4));
                        pipeline.addLast(new StringDecoder(CharsetUtil.UTF_8));
                        pipeline.addLast(new StringEncoder(CharsetUtil.UTF_8));
                        pipeline.addLast(new TcpServerHandler());
                    }
                });
        return serverBootstrap;
    }
}

public class TcpServerHandler extends ChannelInboundHandlerAdapter {
    @Override
    public void channelActive(ChannelHandlerContext ctx) {
        System.out.println("get new client connection ");
    }

    @Override
    public void channelRead(ChannelHandlerContext ctx, Object msg) {
      // 使用之后需要进行释放回到内存池中方便回收,以便于下次使用的时候可重复利用
      // 如果不释放将会产生新的ByteBuf
       ((ByteBuf) msg).release();
    }

    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) {
      ctx.close();
    }
}

// ServerBootstrap.java

    @Override
    @SuppressWarnings("CloneDoesntCallSuperClone")
    public ServerBootstrap clone() {
        // 创建一个新的Boostrap,但是新的Boostrap对象属性沿用先前的bootsrtap属性,属于对象属性浅拷贝
        return new ServerBootstrap(this);
    }
    
    private ServerBootstrap(ServerBootstrap bootstrap) {
        super(bootstrap);
        childGroup = bootstrap.childGroup;
        childHandler = bootstrap.childHandler;
        synchronized (bootstrap.childOptions) {
            childOptions.putAll(bootstrap.childOptions);
        }
        synchronized (bootstrap.childAttrs) {
            childAttrs.putAll(bootstrap.childAttrs);
        }
    }

// 1. 当启动调用`bind()`方法时将会创建一个ServerChannel并监听客户端连接
// 2. 接收到客户端连接时将会新创建一个Channel并监听Channel读写事件

// EventLoop是具备有序性,是一个有序事件执行器
public interface EventLoop extends OrderedEventExecutor, EventLoopGroup {
    @Override
    EventLoopGroup parent();
}

// EventLoopGroup 是一个事件执行器组,在原有的基础上增强事件功能方法
public interface EventLoopGroup extends EventExecutorGroup{}

// 对于OrderedEventExecutor,通过查看开发工具查看父类最终也是一个事件执行器组,在原有的基础上增强功能,即EventExecutor
// OrderedEventExecutor -> EventExecutor -> EventExecutorGroup

// 事件执行器具备两个功能:一个是迭代器功能,一个是计划任务的线程池
interface EventExecutorGroup extends ScheduledExecutorService, Iterable<EventExecutor>{}

ChannelUnregistered: Channel已将被创建,但是还未被注册到EventLoop组件上
ChannelRegistered: Channel已经被注册到EventLoop组件上
ChannelActive: Channel已经与服务端建立连接状态,处于活跃状态,可以进行发送和写出数据
ChannelInActive: Channel与服务端没有建立连接状态

// 与ChannelFuture相关的
interface ChannelFuture extends Future<Void>{}

// 特殊的ChannelFuture,表示具备可写特征,即客户端获取当前类之后是具备可更改属性操作的.
interface Promise<V> extends Future<V>{}	// Future为java并发库下
interface ChannelPromise extends ChannelFuture, Promise<Void>{}

Channel channel = new NioSocketChannel();
// 以异步的方式与远程服务建立连接
ChannelFuture future = channel.connect(new InetSocketAddress("192.168.10.110", 8080));
future.addListener(new ChannelFutureListener(){
   void operationComplete(ChannelFuture future) throws Exception{
      if(future.future.isSuccess()){
         // 建立连接成功
        
      }else{
        //建立连接失败
        
      }
   }
});

// 1. 交由具体子接口定义出入站事件处理方法
// ChannelHandler并没有提供很多的方法声明,同时通过上述的入站和出站事件处理,我们也很容易想到ChannelHandler存在处理入站事件的ChannelInboundHandler以及出站事件的ChannelOutboundHandler,同时在netty中默认有对应实现方式
// ChannelInboundHandlerAdapter实现入站事件的处理
// ChannelOutboundHandlerAdapter实现出站事件的处理
// ChannelDuplexHandler能够同时处理入站和出站事件

// 2. 上下文对象,ChannelHandlerContext
// 要想在上述ChannelHandler的链式事件处理流程,就必须满足两个条件,一个是如何在每个单独ChannelHandler的处理器传递事件,二是每个ChannelHandler是如何通过链式绑定关联的
// ChannelHandler通过ChannelHandlerContext为每个对应的Handler传递事件,因此ChannelHandler必然存在一个上下文对象负责事件传递,类似于EDA的事件通道
// netty事件触发就会创建响应事件的ChannelHandler,并添加到ChannelPipeline中,通过链表的数据结构来维护每个Handler之间关联,同时将ChannelPipeline存储在上下文中,可以通过上下文对象获取管道对象

// 现定义一个需求:需要进行登录才能够获取数据,于是就有了以下的定义
interface Message{
  // methods
}

class DataServerHandler extends SimpleChannelInboundHandler<Message> {
  
  // 存储有状态数据
  private boolean isLogined;
  
  public void channelRead0(ChannelHandlerContext ctx, Message msg){
     if(message instanceof LoginMessage){
       auth((LoginMessage)message);
       isLogined = true;
  	}else if(message instanceof GetDataMessage){
       if(isLogined){
          ctx.writeAndFlush(fetchSecret((GetDataMessage) message));
       }else{
         fail();
       }
  }
}

// 为了避免数据竞争产生数据不一致问题,避免上述需求中的非法登录用户获取到登录数据,必须为每个请求处理连接在提交给handler处理前保证handler是针对当前的连接是1:1的处理方式,即一个连接对应一个channel处理器,对此,需要在添加channel时通过以下的方式进行添加
serverBootstrap.childHandler(new ChannelInitializer<Channel>(){
  public void initChannel(Channel channel){
    channel.pipeline().addLast(new DataServerHandler());
    // 如果一个连接需要用到多个handler协同处理,则只需要调用addLast添加即可
    // 这样每次有事件发生的时候,对应的一个请求处理连接的事件响应处理都会重新创建handler,即保证每个请求连接的事件响应handler都是新创建的,避免了数据竞争产生的数据不一致问题
  }
});

// 如果要保证handler只创建一次,那么就只需要进行调用childHandler添加对应实现的具体Handler实例

interface Message{
  // methods
}

@Sharable
class DataServerHandler extends SimpleChannelInboundHandler<Message> {
  
  // 存储有状态数据
  private boolean sharedObject;
  public void channelRead0(ChannelHandlerContext ctx, Message msg){
    	// ...
  	}
}

serverBootstrap.childHandler(new ChannelInitializer<Channel>(){
  private static final DataServerHandler SHARED = new DataServerHandler();
  public void initChannel(Channel channel){
    // 这样保证在处理不同连接的处理链pipeline中存储的handler都是相同的,并且是线程安全的
    channel.pipeline().addLast(SHARED);
  }
});

// handlerAdded: 当ChannelHandler添加到pipeline被调用
// handlerRemoved: 当ChannelHandler从pipeline中移除时被调用
// exceptionCaught: 在处理过程中ChannelPipeline发生异常时被调用

/**
 * 责任链设计:
 *    channel1 -> channel2  -> channel3 -> ...
 */
// channel1处理完之后的结果将会传递给channel2进行处理,然后将channel2的处理结果传递给channel3再进行处理,那么什么时候结束呢?
//要么遍历到没有下一个channel节点为止就结束
//要么就是直接在当前节点中断不进行往下传播事件
// 根据netty的channel可知,pipeline需要依赖到一个上下文对象,通过上下文对象来实现责任链的数据传输,于是就有以下的定义.

class Main{
 public static void main(String[] args){
     // 类比netty添加方式
     HandlerPipeline pipeline = new HandlerPipeline();
     pipeline.addLast(new HandlerTest());
     pipeline.addLast(new HandlerTest());
  }
}

// 基于链表结构存储handler
class HandlerPipeline {
  private HandlerContext head = new HandlerContext(new Handler(){
    	void doHandler(HandlerContext context, Object val){
            context.nextRun(val);
      }
  });
  
  public void addLast(Handler handler){
     HandlerContext ctx = head;
     while(ctx.next != null){
        ctx = ctx.next;
     }
    ctx.next = new HandlerContext(handler);
  }
}

// 通过上下文保存当前handler并传递数据到下一个context进行处理
class HandlerContext {
   HandlerContext next;
   Handler handler;
  
  HandlerContext(Handler handler){
    this.handler = handler;
  }
  
  void nextRun(Object val){
    if(this.next != null){
      this.next.handler(val);
    }
  }
  
  void handler(Object val){
    this.handler.doHandler(this, val);
  }
}

// handler处理器
interface Handler{
   void doHandler(HandlerContext context, Object val);
}

class HandlerTest implements Handler{
   public void doHandler(HandlerContext context, Object val){
       // 传播给下一个handler
       context.nextRun(val);
      // 如果不传播,在当前handler停止,就无需调用nextRun方法
   }
}

 *  +---------------------------------------------------+---------------+
 *  |                           ChannelPipeline         |               |
 *  |                                                  \|/              |
 *  |    +---------------------+            +-----------+----------+    |
 *  |    | Inbound Handler  N  |            | Outbound Handler  1  |    |
 *  |    +----------+----------+            +-----------+----------+    |
 *  |              /|\                                  |               |
 *  |               |                                  \|/              |
 *  |    +----------+----------+            +-----------+----------+    |
 *  |    | Inbound Handler N-1 |            | Outbound Handler  2  |    |
 *  |    +----------+----------+            +-----------+----------+    |
 *  |              /|\                                  .               |
 *  |               .                                   .               |
 *  | ChannelHandlerContext.fireIN_EVT() ChannelHandlerContext.OUT_EVT()|
 *  |        [ method call]                       [method call]         |
 *  |               .                                   .               |
 *  |               .                                  \|/              |
 *  |    +----------+----------+            +-----------+----------+    |
 *  |    | Inbound Handler  2  |            | Outbound Handler M-1 |    |
 *  |    +----------+----------+            +-----------+----------+    |
 *  |              /|\                                  |               |
 *  |               |                                  \|/              |
 *  |    +----------+----------+            +-----------+----------+    |
 *  |    | Inbound Handler  1  |            | Outbound Handler  M  |    |
 *  |    +----------+----------+            +-----------+----------+    |
 *  |              /|\                                  |               |
 *  +---------------+-----------------------------------+---------------+
 *                  |                                  \|/
 *  +---------------+-----------------------------------+---------------+
 *  |               |                                   |               |
 *  |       [ Socket.read() ]                    [ Socket.write() ]     |
 *  |                                                                   |
 *  |  Netty Internal I/O Threads (Transport Implementation)            |
 *  +-------------------------------------------------------------------+

// 1.pipeline的创建
// 每个Channel都有自己的pipeline,并且当一个新的channel被创建的时候会自动创建

// 2. 责任链流程 -- 见上述责任链流程

// 3. pipeline在上下文进行事件传播的方法
// 入站事件与出站事件
 /* <ul>
   * <li>Inbound event propagation methods:
   *     <ul>
   *     <li>{@link ChannelHandlerContext#fireChannelRegistered()}</li>
   *     <li>{@link ChannelHandlerContext#fireChannelActive()}</li>
   *     <li>{@link ChannelHandlerContext#fireChannelRead(Object)}</li>
   *     <li>{@link ChannelHandlerContext#fireChannelReadComplete()}</li>
   *     <li>{@link ChannelHandlerContext#fireExceptionCaught(Throwable)}</li>
   *     <li>{@link ChannelHandlerContext#fireUserEventTriggered(Object)}</li>
   *     <li>{@link ChannelHandlerContext#fireChannelWritabilityChanged()}</li>
   *     <li>{@link ChannelHandlerContext#fireChannelInactive()}</li>
   *     <li>{@link ChannelHandlerContext#fireChannelUnregistered()}</li>
   *     </ul>
   * </li>
   * <li>Outbound event propagation methods:
   *     <ul>
   *     <li>{@link ChannelHandlerContext#bind(SocketAddress, ChannelPromise)}</li>
   *     <li>{@link ChannelHandlerContext#connect(SocketAddress, SocketAddress, ChannelPromise)}</li>
   *     <li>{@link ChannelHandlerContext#write(Object, ChannelPromise)}</li>
   *     <li>{@link ChannelHandlerContext#flush()}</li>
   *     <li>{@link ChannelHandlerContext#read()}</li>
   *     <li>{@link ChannelHandlerContext#disconnect(ChannelPromise)}</li>
   *     <li>{@link ChannelHandlerContext#close(ChannelPromise)}</li>
   *     <li>{@link ChannelHandlerContext#deregister(ChannelPromise)}</li>
   *     </ul>
   * </li>
   * </ul>
 */

// 4. 构建一个ChannelPipeline
// 当我们的web服务为每个请求处理对应的decode-process-encode时,对于执行比较耗时的操作需要将线程隔离处理,也就是需要有针对Group对process进行处理,而其他线程仍然可以处理decode以及encode非耗时逻辑,可以通过以下的方式:

static final EventExecutorGroup group = new DefaultEventExecutorGroup();
pipeline = ch.pipeline();
pipeline.addLast("decoder", new MyProtocolDecoder());
pipeline.addLast("encoder", new MyProtocolEncoder());
// 这个时候的process处理会单独放在以下的group进行处理
pipeline.addLast(group, "handler", new MyBusinessLogicHandler());


// 5. ChannelPipeline是属于线程安全类

// 1. 通过唤醒回调后在pipeline流程链中向不同的handler传递信息

// 2. 上下文存储的数据可以实现事件触发执行传递到不同的handler方法中,甚至可以是在不同线程中实现数据的共享,比如以下代码:

public class MyHandler extends ChannelDuplexHandler{
  	private ChannelHandlerContext ctx;
    
    public void beforeAdd(ChannelHandlerContext ctx){
       // 可以在添加到pipeline之前保存ctx信息
       this.ctx = ctx;
    }
  
   public void login(String username, password) {
     		// 将保存的ctx存储登录信息并将登录信息传递到责任链pipeline下后续的handler获取使用
        ctx.write(new LoginMessage(username, password));
     }
}

// 3. 存储有状态的信息,详细可以查看上述的ChannelHandler使用

// 4. 一个handler可以拥有多个context信息,因为一个handler可以添加到一个或者多个pipeline中,而每个pipeline都会对应着一个context,因而一个handler是可以拥有一个或者多个context,比如计算handler被添加到pipeline的次数

public class FactorialHandler extends ChannelInboundHandlerAdapter{
   private final AttributeKey<Integer> counter = AttributeKey.valueOf("counter");
   
   public void channelRead(ChannelHandlerContext ctx, Object msg){
      Integer a = ctx.attr(counter).get();
      if (a == null) {
        a = 1;
      }
     attr.set(a * (Integer)msg);
   }
}

// 下面将会进行4次计数器的计算,也就是一个handler实例添加到不同或者相同的active的pipeline中,其上下文对象是不一样的
ChannelPipeline p1 = channel.pipeline();
p1.addLast("f1", fh);
p1.addLast("f2", fh);

ChannelPipeline p2 = channel.pipeline();
p1.addLast("f3", fh);
p1.addLast("f4", fh);

for(int i=0; i<buffer.capacity(); i++){
   byte b = buffer.getByte(i);
   logger.info("char s is " + (char)b);
}

/*
 *      +-------------------+------------------+------------------+
 *      | discardable bytes |  readable bytes  |  writable bytes  |
 *      |                   |     (CONTENT)    |                  |
 *      +-------------------+------------------+------------------+
 *      |                   |                  |                  |
 *      0      <=      readerIndex   <=   writerIndex    <=    capacity
 *.
 */
// 可读字节区域: 代表ByteBuf还未读取到ByteBuf的数据区域,在netty的ByteBuf中以read或者skip开头的读取数据的方法都会在指针readerIndex实现计数的自增加操作,对于可读指针取值范围: 0<=readerIndex<=writeIndex

// 可写字节区域: 代表从writeIndex - capacity之间的区域为空闲区域,能够继续存储数据来填充区域,如果满足writeIndex < capacity代表该ByteBuf拥有足够的区域进行写入数据.

// 丢弃的字节区域: 表示已经被读取过的ByteBuf片段区域,初始化状态的时候,区域大小为0,如果数据一直在被读取,那么对应的区域大小会增加到writeIndex,也就是对于该区域满足: 0<= size <= writeIndex.

// 调用discardReadBytes()方法的区域变化:
/*
 *  BEFORE discardReadBytes()
 *
 *      +-------------------+------------------+------------------+
 *      | discardable bytes |  readable bytes  |  writable bytes  |
 *      +-------------------+------------------+------------------+
 *      |                   |                  |                  |
 *      0      <=      readerIndex   <=   writerIndex    <=    capacity
 *
 *
 *  AFTER discardReadBytes()
 *
 *      +------------------+--------------------------------------+
 *      |  readable bytes  |    writable bytes (got more space)   |
 *      +------------------+--------------------------------------+
 *      |                  |                                      |
 * readerIndex (0) <= writerIndex (decreased)        <=        capacity
 */

// 关于清除方法直接通过查看区域的变化即可
/*
 * <pre>
 *  BEFORE clear()
 *
 *      +-------------------+------------------+------------------+
 *      | discardable bytes |  readable bytes  |  writable bytes  |
 *      +-------------------+------------------+------------------+
 *      |                   |                  |                  |
 *      0      <=      readerIndex   <=   writerIndex    <=    capacity
 *
 *
 *  AFTER clear()
 *
 *      +---------------------------------------------------------+
 *      |             writable bytes (got more space)             |
 *      +---------------------------------------------------------+
 *      |                                                         |
 *      0 = readerIndex = writerIndex            <=            capacity
 * </pre>
 */

// 对于一些普通字节的查询,可以调用indexOf(int, int, byte)或者是bytesBefore(int, int, byte)完成,bytesBefore这个方法尤其对于以一些特殊字符结尾的字符串尤其有用.

// 对于更为复杂的操作,比较存在不同系统的特殊符号查询,可以通过ByteProcessor接口指定查询特定的符号内容,调用forEachByte(int, int, ByteProcessor)方法完成字节的搜索

// 可以通过以下的方法创建一个新的ByteBuf缓冲区,每个派生出来的缓冲区都拥有自己的readerIndex,writeIndex以及标记索引,与NIO的buffer议案具备数据共享,因而当派生的缓冲区数据发生变化的时候,对应的源缓冲区也会发生变化.

// 如果要让数据不具备共享,可以通过使用copy()方法来实现一个新的数据副本,这个时候与派生的数据缓冲区区分在于copy()拥有独立的数据副本信息,可以通过以下的图示来分析,假设现在申请的一块源bytebuf是使用堆外内存存储数据的方式(堆内内存也是同理),这个时候派生的缓冲与copy的缓冲内存区域分布如下:

// 1. 转换为byte[]
if(byteBuff.hasArray()){
   byte[] byteArr = byteBuff.array();
}

// 2. 转换为NIO的Buffers
if(byteBuff.nioBufferCount() > 0){
   ByteBuffer nioByteBuffer = byteBuff.nioBuffer();
}

// 3. 转换为String
String str = byteBuff.toString(Charset.forName("utf8"));

// 4. 转换为IO的字节流
ByteBufInputStream in = new ByteBufInputStream(byteBuff);
ByteBufOutputStream out = new ByteBufOutputStream(byteBuff);

// 现在有一数组数据
byte[] arr = [1,2,3,4,5];
// 这个时候要发送数据出去,可考虑使用堆外内存数据缓冲,避免数据缓冲多一次内存复制,将数据发送到网卡中
// 接收网卡数据的时候,如果这个时候知道接收的数据为数组且需要堆数组数据进行遍历,那么这个时候直接使用堆外内存存储,还需要再进行一次数据内容的复制,而如果使用堆内内存存储,直接可以通过直接获取到数据转换为数组进行操作

// 默认创建池化且为堆外内存存储的ByteBuf,如果当前环境支持Unsafe底层操作,那么默认就会使用Unsafe+堆外存储+池化技术方式来创建ByteBuf
// 如果显示使用PreferHeapByteBufAllocator方式进行分配,则会创建堆内数据存储的ByteBuf

// 核心方法
// 1. hexDump: 以16进制形式打印ByteBuf内容
// ByteBuf内容为: [0, 0, 0, 1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11]
// 16进制打印出来的数据为: 00 00 00 01 02 03 04 05 06 07 08 09 0a 0b
// (空格是手动加的,实际为: 0000000102030405060708090a0b)

// 2. 比较两个ByteBuf的存储的数据是否一致(内存 + 数据大小)