hbase与客户端的通信过程解析

hbase通信主要涵盖了两个技术,一个是google的protobuf rpc通信框架,一个是java的NIO通信

启动入口

org.apache.hadoop.hbase.regionserver.HRegionServer这个类是regionserver的启动类;
org.apache.hadoop.hbase.master.HMaster这个类是Hmaster的启动类,继承了HRegionServer;
而HRegionServer定义了一个org.apache.hadoop.hbase.regionserver.RSRpcServices变量:

private RSRpcServices rsRpcServices;
  • rpcservices主要实现了对于所有客户端请求的核心处理过程;
  • rpcservices中使用了一个关键的类RpcServer用来和客户端通信:
      rpcServer = new RpcServer(rs, name, getServices(),
          bindAddress, // use final bindAddress for this server.
          rs.conf,
          rpcSchedulerFactory.create(rs.conf, this, rs));

因此整个通信过程最核心的就是这两个类:RSRpcServices和RpcServer

利用google的protobuf 实现rpc通信

hbase的protobuf的使用流程如下:

1.在Client.proto中定义service:ClientService,通过protobuf提供的命令生成对应服务接口和message类
service ClientService {
  rpc Get(GetRequest)
    returns(GetResponse);

  rpc Scan(ScanRequest)
    returns(ScanResponse);

  rpc BulkLoadHFile(BulkLoadHFileRequest)
    returns(BulkLoadHFileResponse);

  rpc ExecService(CoprocessorServiceRequest)
    returns(CoprocessorServiceResponse);
    
  rpc ExecRegionServerService(CoprocessorServiceRequest)
    returns(CoprocessorServiceResponse);

  rpc Multi(MultiRequest)
    returns(MultiResponse);
}
2.在服务端定义一个类RSRpcServices实现自动生成的接口:
  • org.apache.hadoop.hbase.protobuf.generated.AdminProtos.AdminService.BlockingInterface
  • org.apache.hadoop.hbase.protobuf.generated.ClientProtos.ClientService.BlockingInterface
    它们继承自:
  • org.apache.hadoop.hbase.protobuf.generated.ClientProtos.ClientService.BlockingInterface
    因此,RSRpcServices便是所有请求的具体实现类。因为它实现了所有的请求AdminService和ClientService。
3.在服务端调用 ClientService.newReflectiveBlockingService(final BlockingInterface impl)方法生成对应的com.google.protobuf.BlockingService实现类。
 public static com.google.protobuf.BlockingService
        newReflectiveBlockingService(final BlockingInterface impl) {
      return new com.google.protobuf.BlockingService() {
        public final com.google.protobuf.Descriptors.ServiceDescriptor
            getDescriptorForType() {
          return getDescriptor();
        }

        public final com.google.protobuf.Message callBlockingMethod(
            com.google.protobuf.Descriptors.MethodDescriptor method,
            com.google.protobuf.RpcController controller,
            com.google.protobuf.Message request)
            throws com.google.protobuf.ServiceException {
          if (method.getService() != getDescriptor()) {
            throw new java.lang.IllegalArgumentException(
              "Service.callBlockingMethod() given method descriptor for " +
              "wrong service type.");
          }
          switch(method.getIndex()) {
            case 0:
              return impl.get(controller, (org.apache.hadoop.hbase.protobuf.generated.ClientProtos.GetRequest)request);
            case 1:
              return impl.mutate(controller, (org.apache.hadoop.hbase.protobuf.generated.ClientProtos.MutateRequest)request);
            case 2:
              return impl.scan(controller, (org.apache.hadoop.hbase.protobuf.generated.ClientProtos.ScanRequest)request);
            case 3:
              return impl.bulkLoadHFile(controller, (org.apache.hadoop.hbase.protobuf.generated.ClientProtos.BulkLoadHFileRequest)request);
            case 4:
              return impl.execService(controller, (org.apache.hadoop.hbase.protobuf.generated.ClientProtos.CoprocessorServiceRequest)request);
            case 5:
              return impl.execRegionServerService(controller, (org.apache.hadoop.hbase.protobuf.generated.ClientProtos.CoprocessorServiceRequest)request);
            case 6:
              return impl.multi(controller, (org.apache.hadoop.hbase.protobuf.generated.ClientProtos.MultiRequest)request);
            default:
              throw new java.lang.AssertionError("Can't get here.");
          }
        }
4.解析客户端发送的请求,映射成MethodDescriptor和Message对象;

此过程主要是以下要讨论的 JAVA NIO做的工作;

5.使用BlockingService执行callBlockingMethod方法进行对客户端请求进行处理

Message callBlockingMethod(MethodDescriptor var1, RpcController var2, Message var3) throws ServiceException;

java NIO

关于 java NIO的使用,主要集中于RpcServer类中:
主要使用了一个listener,但是实际情况这不是一个常见的listener模式,而是用来监听请求的监听器。

// Start the listener here and let it bind to the port
    listener = new Listener(name);

而它的实现如下:

public Listener(final String name) throws IOException {
      super(name);
      backlogLength = conf.getInt("hbase.ipc.server.listen.queue.size", 128);
      // Create a new server socket and set to non blocking mode
      acceptChannel = ServerSocketChannel.open();
      acceptChannel.configureBlocking(false);

      // Bind the server socket to the binding addrees (can be different from the default interface)
      bind(acceptChannel.socket(), bindAddress, backlogLength);
      port = acceptChannel.socket().getLocalPort(); //Could be an ephemeral port
      address = (InetSocketAddress)acceptChannel.socket().getLocalSocketAddress();

      // create a selector;
      selector= Selector.open();

      readers = new Reader[readThreads];
      readPool = Executors.newFixedThreadPool(readThreads,
        new ThreadFactoryBuilder().setNameFormat(
          "RpcServer.reader=%d,bindAddress=" + bindAddress.getHostName() +
          ",port=" + port).setDaemon(true).build());
      for (int i = 0; i < readThreads; ++i) {
        Reader reader = new Reader();
        readers[i] = reader;
        readPool.execute(reader);
      }
      LOG.info(getName() + ": started " + readThreads + " reader(s) listening on port=" + port);

      // Register accepts on the server socket with the selector.
      acceptChannel.register(selector, SelectionKey.OP_ACCEPT);
      this.setName("RpcServer.listener,port=" + port);
      this.setDaemon(true);
    }

主要定义了一个acceptChannel,一个selector和多个readers,每个reader对应一个selector;

1.accept connection

它的主线程是监控selector中的accept请求,进行doAccept操作:

 void doAccept(SelectionKey key) throws IOException, OutOfMemoryError {
      Connection c;
      ServerSocketChannel server = (ServerSocketChannel) key.channel();

      SocketChannel channel;
      while ((channel = server.accept()) != null) {
        try {
          channel.configureBlocking(false);
          channel.socket().setTcpNoDelay(tcpNoDelay);
          channel.socket().setKeepAlive(tcpKeepAlive);
        } catch (IOException ioe) {
          channel.close();
          throw ioe;
        }

        Reader reader = getReader();
        try {
          reader.startAdd();
          SelectionKey readKey = reader.registerChannel(channel);
          c = getConnection(channel, System.currentTimeMillis());
          readKey.attach(c);
          synchronized (connectionList) {
            connectionList.add(numConnections, c);
            numConnections++;
          }
          if (LOG.isDebugEnabled())
            LOG.debug(getName() + ": connection from " + c.toString() +
                "; # active connections: " + numConnections);
        } finally {
          reader.finishAdd();
        }
      }
    }

主要是对每个accept请求创建了一个connection对象,每个connection对应一个读写数据的channel,然后注册channel给某一个reader的selector;

2.read and process操作

对于每个reader线程来说,会对自己selector绑定的所有的SelectionKey进行查看,如果接收到数据,那么对绑定的connection进行处理,最后调用connection的process方法;
解析收到的请求,然后创建请求;通过scheduler执行,

Call call = new Call(id, this.service, md, header, param, cellScanner, this, responder,
              totalRequestSize, traceInfo, this.addr);

      if (!scheduler.dispatch(new CallRunner(RpcServer.this, call))) {
        callQueueSize.add(-1 * call.getSize());

scheduler是整个regionserver处理所有请求的核心,创建scheduler需要用到参数如下,因此hbase.regionserver.handler.count参数决定了同时进行处理请求的handler个数,即regionserver的并发能力。

int handlerCount = conf.getInt(HConstants.REGION_SERVER_HANDLER_COUNT,
        HConstants.DEFAULT_REGION_SERVER_HANDLER_COUNT);
 public static final String REGION_SERVER_HANDLER_COUNT = "hbase.regionserver.handler.count";

最后再rpcserver中调用call函数:
Message result = service.callBlockingMethod(md, controller, param);

3.返回数据序列化

上边写到数据的具体执行在CallRunner中,执行结束后调用Call.setResponse方法,

 protected synchronized void setResponse(Object m, final CellScanner cells,
        Throwable t, String errorMsg) {
       ...
        Message header = headerBuilder.build();

        // Organize the response as a set of bytebuffers rather than collect it all together inside
        // one big byte array; save on allocations.
        ByteBuffer bbHeader = IPCUtil.getDelimitedMessageAsByteBuffer(header);
        ByteBuffer bbResult = IPCUtil.getDelimitedMessageAsByteBuffer(result);
        int totalSize = bbHeader.capacity() + (bbResult == null? 0: bbResult.limit()) +
          (this.cellBlock == null? 0: this.cellBlock.limit());
        ByteBuffer bbTotalSize = ByteBuffer.wrap(Bytes.toBytes(totalSize));
        bc = new BufferChain(bbTotalSize, bbHeader, bbResult, this.cellBlock);
       ...
      } catch (IOException e) {
        LOG.warn("Exception while creating response " + e);
      }
      this.response = bc;
    }

其中通过IPCUtil.getDelimitedMessageAsByteBuffer(result)把messgae数据序列化成buffer,调用google提供的com.google.protobuf.CodedOutputStream实现的序列化方法。

CodedOutputStream cos = CodedOutputStream.newInstance(buffer);
    // This will write out the vint preamble and the message serialized.
    cos.writeMessageNoTag(m);
5.response数据写回channel

以下是reponder提供的方法:

 void doRespond(Call call) throws IOException {
      boolean added = false;

      // If there is already a write in progress, we don't wait. This allows to free the handlers
      //  immediately for other tasks.
      if (call.connection.responseQueue.isEmpty() && call.connection.responseWriteLock.tryLock()) {
        try {
          if (call.connection.responseQueue.isEmpty()) {
            // If we're alone, we can try to do a direct call to the socket. It's
            //  an optimisation to save on context switches and data transfer between cores..
            if (processResponse(call)) {
              return; // we're done.
            }
            // Too big to fit, putting ahead.
            call.connection.responseQueue.addFirst(call);
            added = true; // We will register to the selector later, outside of the lock.
          }
        } finally {
          call.connection.responseWriteLock.unlock();
        }
      }

      if (!added) {
        call.connection.responseQueue.addLast(call);
      }
      call.responder.registerForWrite(call.connection);

      // set the serve time when the response has to be sent later
      call.timestamp = System.currentTimeMillis();
    }

最后将数据写进属于自己的channel中。

    原文作者:tinyMonkey
    原文地址: https://www.jianshu.com/p/6c5ea570e70a
    本文转自网络文章,转载此文章仅为分享知识,如有侵权,请联系博主进行删除。
点赞