Apache Bookkeeper 的网络框架

概述

Apache Bookkeeper采用的是Request-Response的一问一答方式，对网络请求处理这块直接采用的 Netty框架，所以其实这块可讲的不是太多;
Netty 我们在这里就不累述了，网上资料很多，其实我也是一知半解啊 :(

实现

主要实现在文件：bookkeeper-server/src/main/java/org/apache/bookkeeper/proto/BookieNettyServer.java

写法就是常规的Netty框架的使用,具体在listenOn函数中，我们来看一下

Netty.ServerBootstrap的基础设置，下面作一些简单注释

 ServerBootstrap bootstrap = new ServerBootstrap();
          // 设置Accept事件循环和work事件循环中的内存分配器，这个allocator是Bookkeeper自己实现的，我们后面会介绍到
          bootstrap.option(ChannelOption.ALLOCATOR, allocator);
          bootstrap.childOption(ChannelOption.ALLOCATOR, allocator);
          
          // 设置Accept事件循环和work事件循环组
          bootstrap.group(eventLoopGroup, eventLoopGroup);
          bootstrap.childOption(ChannelOption.TCP_NODELAY, conf.getServerTcpNoDelay());
          bootstrap.childOption(ChannelOption.SO_LINGER, conf.getServerSockLinger());
          
          // 设置自适应的ReceiveBuffer, 真正分配buffer时也是用的上面的allocator
          bootstrap.childOption(ChannelOption.RCVBUF_ALLOCATOR,
                  new AdaptiveRecvByteBufAllocator(conf.getRecvByteBufAllocatorSizeMin(),
                          conf.getRecvByteBufAllocatorSizeInitial(), conf.getRecvByteBufAllocatorSizeMax()));
                          
          //设置发送的高低水位，来控制发送
          bootstrap.option(ChannelOption.WRITE_BUFFER_WATER_MARK, new WriteBufferWaterMark(
                  conf.getServerWriteBufferLowWaterMark(), conf.getServerWriteBufferHighWaterMark()));

          if (eventLoopGroup instanceof EpollEventLoopGroup) {
              bootstrap.channel(EpollServerSocketChannel.class);
          } else {
              bootstrap.channel(NioServerSocketChannel.class);
          }

Netty中用来处理Request和Response的Handler的设置
对进来的Request，Netty采用职责链的design pattern来处依次处理, 下面代码中的Inbound都是用来处理Request的，Outbound用来处理Response

bootstrap.childHandler(new ChannelInitializer<SocketChannel>() {
              @Override
              protected void initChannel(SocketChannel ch) throws Exception {
                  synchronized (suspensionLock) {
                      while (suspended) {
                          suspensionLock.wait();
                      }
                  }
                  BookieSideConnectionPeerContextHandler contextHandler =
                      new BookieSideConnectionPeerContextHandler();
                  ChannelPipeline pipeline = ch.pipeline();

                  // Outbound
                  pipeline.addLast("bytebufList", ByteBufList.ENCODER_WITH_SIZE);
                  // Inbound
                  pipeline.addLast("lengthbaseddecoder", new LengthFieldBasedFrameDecoder(maxFrameSize, 0, 4, 0, 4));
                  // Outbound
                  pipeline.addLast("lengthprepender", new LengthFieldPrepender(4));
                  // Inbound
                  pipeline.addLast("bookieProtoDecoder", new BookieProtoEncoding.RequestDecoder(registry));
                  // Outbound
                  pipeline.addLast("bookieProtoEncoder", new BookieProtoEncoding.ResponseEncoder(registry));
                  // Inbound
                  pipeline.addLast("bookieAuthHandler", new AuthHandler.ServerSideHandler(
                              contextHandler.getConnectionPeer(), authProviderFactory));
                  ChannelInboundHandler requestHandler = isRunning.get()
                          ? new BookieRequestHandler(conf, requestProcessor, allChannels)
                          : new RejectRequestHandler();
                  // Inbound
                  pipeline.addLast("bookieRequestHandler", requestHandler);
                  // Inbound
                  pipeline.addLast("contextHandler", contextHandler);
              }
          });

Request的处理流程

network-request-handler.jpg

网络线程模型
thread-model.jpg
请求进来后经一系列的netty的处理，最后会到达BookieRequestProcessor类来真正处理业务逻辑，它又使用四类线程池来处理不同的请求：读请求，写请求, long poll和高优先级请求,这里的读和写不是指网络数据的读写，是针对ledger, entry, lac等;

自定义的ByteBuffer分配器

java的应用gc是个大问题，虽然现在gc算法一致在进步，另外一个问题是对于网络应用，网络收发包最终都是通过native的socket方法完成，如果直接使用jvm heap上的bytebuffer, 需要多次拷贝;
为解决以上问题，java提供了堆外内存，对应于heap上的类型是DirectByteBuffer(它的回收还是受到一部分地受到gc的管理，有兴趣同学可以自行google), 有了DirectByteBuffer,再加上池化技术又可以给已分配的ByteBuffer重复利用
Apache Bookkeeper利用Netty提供的 PooledByteBufAllocator和UnpooledByteBufAllocator构建了自己的ByteBuffer Allocator, 主要是提供了几种自定义的策略:
1. PoolingPolicy策略：
  a. UnpooledHeap: 从heap上分配，且不采用池化方式;
  b. PooledDirect: 从堆外分配，且采用池化方式;
2. OutOfMemoryPolicy策略：
  a. ThrowException: 不能分配内存时，抛出异常;
  b. FallbackToHeap: 如果无法分配堆外内存，则尝试从heap上分配内存
这部分代码位于：bookkeeper-common-allocator/src/main/java/org/apache/bookkeeper/common/allocator/imp/ByteBufAllocatorImpl.java