从HDFS源码学习接口重试机制

最近博主，接触到了以前的老项目需要维护。需要从一个接口获取数据，但是这个数据很重要很重要，没有这个数据的话当前系统无法初始化。那么和这个时候我们就小心处理了，也就是需要重试当初的接口访问。

static DFSOutputStream newStreamForCreate(DFSClient dfsClient, String src,
      FsPermission masked, EnumSet flag, boolean createParent,
      short replication, long blockSize, Progressable progress, int buffersize,
      DataChecksum checksum, String[] favoredNodes) throws IOException {
    TraceScope scope = dfsClient.getPathTraceScope("newStreamForCreate", src);
    try {
      HdfsFileStatus stat = null;

      // Retry the create if we get a RetryStartFileException up to a maximum
      // number of times
      boolean shouldRetry = true;
      int retryCount = CREATE_RETRY_COUNT;

      while (shouldRetry) {
        shouldRetry = false;
        try {
          stat = dfsClient.namenode.create(src, masked, dfsClient.clientName,
              new EnumSetWritable(flag), createParent, replication,
              blockSize, SUPPORTED_CRYPTO_VERSIONS);
          break;
        } catch (RemoteException re) {
          IOException e = re.unwrapRemoteException(
              AccessControlException.class,
              DSQuotaExceededException.class,
              FileAlreadyExistsException.class,
              FileNotFoundException.class,
              ParentNotDirectoryException.class,
              NSQuotaExceededException.class,
              RetryStartFileException.class,
              SafeModeException.class,
              UnresolvedPathException.class,
              SnapshotAccessControlException.class,
              UnknownCryptoProtocolVersionException.class);
          if (e instanceof RetryStartFileException) {
            if (retryCount > 0) {
              shouldRetry = true;
              retryCount--;
            } else {
              throw new IOException("Too many retries because of encryption" +
                  " zone operations", e);
            }
          } else {
            throw e;
          }
        }
      }
      Preconditions.checkNotNull(stat, "HdfsFileStatus should not be null!");

      //TODO 里面初始化了DataStreamer,DataStreamer是写数据流程里面重要的对象
      final DFSOutputStream out = new DFSOutputStream(dfsClient, src, stat,
          flag, progress, checksum, favoredNodes);
      //TODO 里面启动了DataStreamer
      out.start();
      return out;
    } finally {
      scope.close();
    }
  }

其实这段代码的意思也就是说，通过RPC服务调用namenode的create方法，来创建一个文件，并且创建一个dfs流，如果抛异常了，则进行重试。那么我们是否可以抽取一下这个方法呢。

    public  T fetchData() throws IOException {
        try {
            boolean shouldRetry = true;
            int retryCount = CREATE_RETRY_COUNT;

            while (shouldRetry) {
                shouldRetry = false;
                try {
                   //执行调用的方法
                    break;
                } catch (Exception e) {
                    if (retryCount > 0) {
                        shouldRetry = true;
                        retryCount--;
                    } else {
                        e.printStackTrace();
                        log.error("错误处理,可以存起来，或者其他的一些操作，根据业务来看");
                    }
                }
            }
            return out;
        } finally {
            //关闭流或者其他一些操作
        }
    }

那么这样我们就把一些优秀项目的一些经验吸收了，那么我们这样获取数据或者写出数据。假如我们有很多数据源，很多的结果呢，那么我们是否可以抽取成为模板方法呢。

public abstract class TemplateRetryPattern {
    private static final int RETRY_COUNT = 3;

    public abstract  T execute(Class clazz);

    public  T fetchData(Class clazz) throws Exception {
        boolean shouldRetry = true;
        int retryCount = RETRY_COUNT;
        while (shouldRetry) {
            shouldRetry = false;
            try {
                return execute(clazz);
            } catch (Exception e) {
                if (retryCount > 0) {
                    shouldRetry = true;
                    retryCount--;
                } else {
                    throw new RuntimeException("拉取数据失败,");
                }
            }
        }
        return null;
    }
}

那么以上就是我们的模板设计模式了，当然方法参数可以根据我们的业务来进行修改。

以上就是根据HDFS源码来吸取经验，生成我们自己的经验，当然本人才疏学浅，假如有不对的地方，望各位大佬不吝斧正。