一、问题背景

当客户端发送RPC请求给服务端时，基于各种原因，服务器的响应很可能会超时。如果客户端只是在那等待，针对数据的操作，很可能出现服务器端已处理完毕，但是无法通知客户端，此时，客户端只能重新发起请求，可是又可能造成服务端重复处理请求。该如何解决该问题呢？

二、解决方案

实际上，客户端发送RPC请求给服务器后，如果响应超时，那么客户端会重复发送请求，直到达到参数配置的重试次数上限。而且，客户端第一次发送和以后重发请求时，会附带相同的nonce，服务端只要根据nonce进行判断，就能得知是否为同一请求，并根据之前请求处理的结果，决定是等待、拒绝还是直接处理。

三、HBase如何实现的

在HRegionServer中，有一个ServerNonceManager类型的成员变量nonceManager，由它负责管理该RegionServer上的nonce。其定义如下：

final ServerNonceManager nonceManager;

ServerNonceManager中有一个十分重要的方法，用于当一个操作在服务端执行后未及时反馈响应给客户端，客户端重新发起携带相同nonceGroup和nonce的同一操作的请求时，服务端根据nonceGroup和nonce做相应的判断。定义如下：

* Starts the operation if operation with such nonce has not already succeeded. If the * operation is in progress, waits for it to end and checks whether it has succeeded. * 如果操作未执行成功，重新开始一个操作。如果该操作在进行过程中，等待它完成并判断它是否成功。 * @param group Nonce group. * @param nonce Nonce. * @param stoppable Stoppable that terminates waiting (if any) when the server is stopped. * @return true if the operation has not already succeeded and can proceed; false otherwise. public boolean startOperation(long group, long nonce, Stoppable stoppable) throws InterruptedException { // 如果传入的nonce为0，则返回true，表明操作可以进行 if (nonce == HConstants.NO_NONCE) return true; // 构造NonceKey实例nk NonceKey nk = new NonceKey(group, nonce); // 构造OperationContext实例ctx，初始状态为WAIT OperationContext ctx = new OperationContext(); while (true) { // 将NonceKey到OperationContext的映射，添加到ConcurrentHashMap类型的nonces中去 OperationContext oldResult = nonces.putIfAbsent(nk, ctx); // 如果之前没有，则说明该操作可以直接执行 if (oldResult == null) return true; // Collision with some operation - should be extremely rare. // 如果之前存在该操作，则取出该操作nonce对应的OperationContext synchronized (oldResult) { // 获得该nonce对应的OperationContext状态 int oldState = oldResult.getState(); LOG.debug("Conflict detected by nonce: " + nk + ", " + oldResult); // 如果之前的状态不是WAIT if (oldState != OperationContext.WAIT) { // 如果之前的状态是PROCEED，说明之前的操作执行完成且以失败告终，此处返回true，表示操作可以再次执行 return oldState == OperationContext.PROCEED; // operation ended // 等待一段时间后继续循环 oldResult.setHasWait(); oldResult.wait(this.conflictWaitIterationMs); // operation is still active... wait and loop // 判断RegionServer的状态 if (stoppable.isStopped()) { throw new InterruptedException("Server stopped"); }        在RSRpcServices的append()方法中，有如下代码：

if (r == null) {
      long nonce = startNonceOperation(m, nonceGroup);
      boolean success = false;
      try {
        r = region.append(append, nonceGroup, nonce);
        success = true;
      } finally {
        endNonceOperation(m, nonceGroup, success);
      if (region.getCoprocessorHost() != null) {
        region.getCoprocessorHost().postAppend(append, r);
    }

其中，startNonceOperation()方法源码如下： * Starts the nonce operation for a mutation, if needed. * 如果需要的话，为mutation开启一个nonce操作 * @param mutation Mutation. * @param nonceGroup Nonce group from the request. * @returns Nonce used (can be NO_NONCE). private long startNonceOperation(final MutationProto mutation, long nonceGroup) throws IOException, OperationConflictException { // 如果RegionServer上的nonceManager为null，或者该mutation不存在nonce，那么直接返回HConstants.NO_NONCE，即0 if (regionServer.nonceManager == null || !mutation.hasNonce()) return HConstants.NO_NONCE; // 标志位，是否可以运行 boolean canProceed = false; try { // 调用RegionServer上nonceManager的startOperation()方法，确定是否可以执行该操作 canProceed = regionServer.nonceManager.startOperation( nonceGroup, mutation.getNonce(), regionServer); } catch (InterruptedException ex) { throw new InterruptedIOException("Nonce start operation interrupted"); if (!canProceed) {// 如果不能运行，抛出OperationConflictException异常，即操作冲突异常 // TODO: instead, we could convert append/increment to get w/mvcc String message = "The operation with nonce {" + nonceGroup + ", " + mutation.getNonce() + "} on row [" + Bytes.toString(mutation.getRow().toByteArray()) + "] may have already completed"; throw new OperationConflictException(message); // 最后，返回mutation的nonce return mutation.getNonce();
它会调用RegionServer上nonceManager的startOperation()方法，确定是否可以执行该操作。 当客户端发送RPC请求给服务端时，基于各种原因，服务器的响应很可能会超时。如果客户端只是在那等待，针对数据的操作，很可能出现服务器端已处理完毕，但是无法通知客户端，此时，客户端只能重新发起请求，可是又可能造成服务端重复处理请求。该如何解决该问题呢？ 首先，定义appid，appsecret， nonce ，timestamp。 然后，获取parmas和Body中的参数，并将两者组合成在一个字典中。 此外，根据字典的key的ASCII码升序排序，把字典格式转换为key=value&key=value形式，生成一个字符串。 在这个字符串前面加上appid和appsecret，后面加上 nonce 和timestamp。 对该字符串做MD5加密并大写，生成sign，sign保存为全局变量。 最后，把sign放在请求头中发送给服务器做鉴权 import org.apache.hadoop.conf.Configuration; import org.apache.hadoop. hbase . HBase Configuration; import org.apache.hadoop. hbase .client.Connection; import org.apache.hadoop. hbase .client.ConnectionFactory; import java.io.IOException; public class HBase Connection { public static void main(String[] args) throws IOException { // 定义 HBase 配置文件 Configuration conf = HBase Configuration.create(); // 设置ZooKeeper集群地址 conf.set(" hbase .zookeeper.quorum","zk1.example.com,zk2.example.com,zk3.example.com"); // 创建 HBase 连接 Connection connection = ConnectionFactory.createConnection(conf); // 输出连接状态 System.out.println(" HBase 连接状态：" + connection.isClosed()); // 关闭连接 connection.close(); 在实际使用中，您需要根据自己的环境和需要进行一些设置。我们也可以从配置文件中读取相应的参数。 书忆江南: 连接Alluxio master的方法API可以参考：https://docs.alluxio.io/os/user/stable/en/api/FS-API.html#java-client 我是这样连接的：[code=java] InstancedConfiguration conf = InstancedConfiguration.defaults(); // 如果用PropertyKey.MASTER_HOSTNAME参数设置Alluxio master连接，应该是只能设置一个host，如果该host从active变成stand by，可能client会连不上，所以用如下方式设置多个master host，client自己会去找到active的Alluxio master去连接 conf.set(PropertyKey.MASTER_EMBEDDED_JOURNAL_ADDRESSES,"alluxioMasterHostname1:19200,alluxioMasterHostname2:1920"); conf.set(PropertyKey.SECURITY_LOGIN_USERNAME, "userName"); FileSystem customizedFs = FileSystem.Factory.create(conf); // 拿到Alluxio Client代表的FileSystem后，call这个类里的各种API，例如： // 去掉"hdfs://clusterName"前缀 String plainPath = new Path(hdfsLocation).toUri().getPath(); AlluxioURI path = new AlluxioURI(plainPath); customizedFs.delete(path, DeletePOptions.newBuilder().setAlluxioOnly(true).setRecursive(true).build()); [/code] ThreadPoolExecutor源码分析（一）：重要成员变量 哦哦是的 搞混了