org.apache.hadoop.hbase.HConstants#NORMAL_QOS源码实例Demo

@Override
public boolean dispatch(CallRunner callTask) throws InterruptedException {
  RpcCall call = callTask.getRpcCall();
  int level = priority.getPriority(call.getHeader(), call.getParam(),
      call.getRequestUser().orElse(null));
  if (level == HConstants.PRIORITY_UNSET) {
    level = HConstants.NORMAL_QOS;
  }
  if (metaTransitionExecutor != null &&
    level == MasterAnnotationReadingPriorityFunction.META_TRANSITION_QOS) {
    return metaTransitionExecutor.dispatch(callTask);
  } else if (priorityExecutor != null && level > highPriorityLevel) {
    return priorityExecutor.dispatch(callTask);
  } else if (replicationExecutor != null && level == HConstants.REPLICATION_QOS) {
    return replicationExecutor.dispatch(callTask);
  } else {
    return callExecutor.dispatch(callTask);
  }
}
 

@Test
public void test() throws IOException, InterruptedException {
  int rpcHeaderLen = HConstants.RPC_HEADER.length;
  byte[] preamble = new byte[rpcHeaderLen + 2];
  System.arraycopy(HConstants.RPC_HEADER, 0, preamble, 0, rpcHeaderLen);
  preamble[rpcHeaderLen] = HConstants.RPC_CURRENT_VERSION;
  preamble[rpcHeaderLen + 1] = AuthMethod.SIMPLE.code;
  socket.getOutputStream().write(preamble, 0, rpcHeaderLen + 1);
  socket.getOutputStream().flush();
  Thread.sleep(5000);
  socket.getOutputStream().write(preamble, rpcHeaderLen + 1, 1);
  socket.getOutputStream().flush();

  ConnectionHeader header = ConnectionHeader.newBuilder()
      .setServiceName(TestRpcServiceProtos.TestProtobufRpcProto.getDescriptor().getFullName())
      .setVersionInfo(ProtobufUtil.getVersionInfo()).build();
  DataOutputStream dos = new DataOutputStream(socket.getOutputStream());
  dos.writeInt(header.getSerializedSize());
  header.writeTo(dos);
  dos.flush();

  int callId = 10;
  Call call = new Call(callId, TestProtobufRpcProto.getDescriptor().findMethodByName("ping"),
      EmptyRequestProto.getDefaultInstance(), null, EmptyResponseProto.getDefaultInstance(), 1000,
      HConstants.NORMAL_QOS, null, MetricsConnection.newCallStats());
  RequestHeader requestHeader = IPCUtil.buildRequestHeader(call, null);
  dos.writeInt(IPCUtil.getTotalSizeWhenWrittenDelimited(requestHeader, call.param));
  requestHeader.writeDelimitedTo(dos);
  call.param.writeDelimitedTo(dos);
  dos.flush();

  DataInputStream dis = new DataInputStream(new BufferedInputStream(socket.getInputStream()));
  int size = dis.readInt();
  ResponseHeader responseHeader = ResponseHeader.parseDelimitedFrom(dis);
  assertEquals(callId, responseHeader.getCallId());
  EmptyResponseProto.Builder builder = EmptyResponseProto.newBuilder();
  builder.mergeDelimitedFrom(dis);
  assertEquals(size, IPCUtil.getTotalSizeWhenWrittenDelimited(responseHeader, builder.build()));
}
 

public AsyncServerRequestRpcRetryingCaller(Timer retryTimer, AsyncConnectionImpl conn,
    long pauseNs, long pauseForCQTBENs, int maxAttempts, long operationTimeoutNs,
    long rpcTimeoutNs, int startLogErrorsCnt, ServerName serverName, Callable<T> callable) {
  super(retryTimer, conn, HConstants.NORMAL_QOS, pauseNs, pauseForCQTBENs, maxAttempts,
    operationTimeoutNs, rpcTimeoutNs, startLogErrorsCnt);
  this.serverName = serverName;
  this.callable = callable;
}
 

static int getPriority(TableName tableName) {
  if (tableName.isSystemTable()) {
    return HConstants.SYSTEMTABLE_QOS;
  } else {
    return HConstants.NORMAL_QOS;
  }
}
 

public InterRegionServerRpcController(HBaseRpcController delegate, Configuration conf) {
    super(delegate);
    // Set priority higher that normal, but lower than high
    this.priority = (HConstants.HIGH_QOS + HConstants.NORMAL_QOS) / 2;
}
 

public InterRegionServerRpcController(PayloadCarryingRpcController delegate, Configuration conf) {
    super(delegate);
    // Set priority higher that normal, but lower than high
    this.priority = (HConstants.HIGH_QOS + HConstants.NORMAL_QOS) / 2;
}
 

/**
 * Get the priority for a given request from the header and the param
 * This doesn't consider which user is sending the request at all.
 * This doesn't consider annotations
 */
protected int getBasePriority(RequestHeader header, Message param) {
  if (param == null) {
    return HConstants.NORMAL_QOS;
  }

  // Trust the client-set priorities if set
  if (header.hasPriority()) {
    return header.getPriority();
  }

  String cls = param.getClass().getName();
  Class<? extends Message> rpcArgClass = argumentToClassMap.get(cls);
  RegionSpecifier regionSpecifier = null;
  //check whether the request has reference to meta region or now.
  try {
    // Check if the param has a region specifier; the pb methods are hasRegion and getRegion if
    // hasRegion returns true.  Not all listed methods have region specifier each time.  For
    // example, the ScanRequest has it on setup but thereafter relies on the scannerid rather than
    // send the region over every time.
    Method hasRegion = methodMap.get("hasRegion").get(rpcArgClass);
    if (hasRegion != null && (Boolean)hasRegion.invoke(param, (Object[])null)) {
      Method getRegion = methodMap.get("getRegion").get(rpcArgClass);
      regionSpecifier = (RegionSpecifier)getRegion.invoke(param, (Object[])null);
      Region region = rpcServices.getRegion(regionSpecifier);
      if (region.getRegionInfo().getTable().isSystemTable()) {
        if (LOG.isTraceEnabled()) {
          LOG.trace("High priority because region=" +
            region.getRegionInfo().getRegionNameAsString());
        }
        return HConstants.SYSTEMTABLE_QOS;
      }
    }
  } catch (Exception ex) {
    // Not good throwing an exception out of here, a runtime anyways.  Let the query go into the
    // server and have it throw the exception if still an issue.  Just mark it normal priority.
    if (LOG.isTraceEnabled()) LOG.trace("Marking normal priority after getting exception=" + ex);
    return HConstants.NORMAL_QOS;
  }

  if (param instanceof ScanRequest) { // scanner methods...
    ScanRequest request = (ScanRequest)param;
    if (!request.hasScannerId()) {
      return HConstants.NORMAL_QOS;
    }
    RegionScanner scanner = rpcServices.getScanner(request.getScannerId());
    if (scanner != null && scanner.getRegionInfo().getTable().isSystemTable()) {
      if (LOG.isTraceEnabled()) {
        // Scanner requests are small in size so TextFormat version should not overwhelm log.
        LOG.trace("High priority scanner request " + TextFormat.shortDebugString(request));
      }
      return HConstants.SYSTEMTABLE_QOS;
    }
  }

  return HConstants.NORMAL_QOS;
}
 

@Override
public int getPriority() {
  return priority < 0 ? HConstants.NORMAL_QOS : priority;
}