FileSystem 这个类是最常用的类，因为只要是我们想连接 HDFS 进行一些操作，程序的入口必定是它。接下来主要分析一下，本地代码是如何通过 FileSystem 连接上 hadoop 并且读取数据的

本地代码示例

由于要连接的集群是一个配置 HA 的 hadoop 集群，所有需要把 core-site.xml 和 hdfs-site.xml 文件放到resources 目录中

  1. public static void main(String[] args) throws IOException {
  3.     // 构建配置, 默认加载core-site.xml hdfs-site.xml
  4.     Configuration conf = new Configuration();
  6.     // 构建FileSystem——核心
  7.     FileSystem  fs = FileSystem.get(conf);
  9.     //读取根目录上有哪些文件.
  10.     FileStatus[] list = fs.listStatus(new Path("/"));
  11.     for (FileStatus file:list ) {
  12.         System.out.println(file.getPath().getName());
  13.     }
  15. }
  1. <dependency>
  2.     <groupId>org.apache.hadoop</groupId>
  3.     <artifactId>hadoop-client</artifactId>
  4.     <version>3.2.1</version>
  5. </dependency>

解析

FileSystem.get(conf)

  1. // 代码核心
  2. FileSystem  fs = FileSystem.get(conf);
  1. public static FileSystem get(Configuration conf) throws IOException{
  2.     return get(getDefaultUri(conf), conf);
  3. }
  5. public static URI getDefaultUri(Configuration conf) {
  7.     URI uri =
  8.         URI.create(fixName(conf.getTrimmed(FS_DEFAULT_NAME_KEY, DEFAULT_FS)));
  10.     if (uri.getScheme() == null) {
  11.       throw new IllegalArgumentException("No scheme in default FS: " + uri);
  12.     }
  14.     return uri; // 返回core-site.xml中的fs.defaultFS: hdfs://master
  16. }

FileSystem get(URI uri, Configuration conf)

通过 CACHE.get(uri, conf); 获取对应的 FileSystem。默认配置是走缓存，如果设置不走缓存则自动创建一个去走，就是：createFileSystem 方法

  1. public static FileSystem get(URI uri, Configuration conf) throws IOException {
  2.     String scheme = uri.getScheme();
  3.     String authority = uri.getAuthority();
  5.     if (scheme == null && authority == null) {     // use default FS
  6.         return get(conf);
  7.     }
  9.     if (scheme != null && authority == null) {     // no authority
  10.         URI defaultUri = getDefaultUri(conf);
  11.             if (scheme.equals(defaultUri.getScheme())    // if scheme matches default
  12.                   && defaultUri.getAuthority() != null) {  // & default has authority
  13.             return get(defaultUri, conf);              // return default
  14.         }
  15.     }
  17.     String disableCacheName = String.format("fs.%s.impl.disable.cache", scheme);
  18.     if (conf.getBoolean(disableCacheName, false)) {
  19.         LOGGER.debug("Bypassing cache to create filesystem {}", uri);
  20.     return createFileSystem(uri, conf);
  21.     }
  23.     return CACHE.get(uri, conf);
  24. }
  1. FileSystem get(URI uri, Configuration conf) throws IOException{
  2.     FileSystem.Cache.Key key = new FileSystem.Cache.Key(uri, conf);
  3.     return this.getInternal(uri, conf, key);
  4. }
  6. /** The objects inserted into the cache using this method are all unique. */
  7. FileSystem getUnique(URI uri, Configuration conf) throws IOException{
  8.     FileSystem.Cache.Key key = new FileSystem.Cache.Key(uri, conf, unique.getAndIncrement());
  9.     return this.getInternal(uri, conf, key);
  10. }

缓存里面的方法就是根据 uri 和 conf 生成的 key ，在缓存中查询是否有已经创建的 FileSystem ，有就复用，没有就创建。并将创建的 FileSystem 加入缓存

CreateFileSystem(uri, conf)

  1. private static FileSystem createFileSystem(URI uri, Configuration conf) throws IOException {
  2.     Tracer tracer = FsTracer.get(conf);
  3.     TraceScope scope = tracer.newScope("FileSystem#createFileSystem");
  4.     Throwable var4 = null;
  6.     FileSystem var7;
  7.     try {
  8.         scope.addKVAnnotation("scheme", uri.getScheme());
  9.         // 根据配置获取需要加载的FileSystem实现类
  10.         Class<?> clazz = getFileSystemClass(uri.getScheme(), conf);
  11.         // 实例化配置 FileSystem
  12.         FileSystem fs = (FileSystem)ReflectionUtils.newInstance(clazz, conf);
  13.         // 对 FileSystem 进行初始化
  14.         fs.initialize(uri, conf);
  15.         var7 = fs;
  16.     } catch (Throwable var16) {
  17.         var4 = var16;
  18.         throw var16;
  19.     } finally {
  20.         if (scope != null) {
  21.             if (var4 != null) {
  22.                 try {
  23.                     scope.close();
  24.                 } catch (Throwable var15) {
  25.                     var4.addSuppressed(var15);
  26.                 }
  27.             } else {
  28.                 scope.close();
  29.             }
  30.         }
  31.     }
  32.     return var7;
  33. }

这里其实就是通过配置获取到 FileSystem 的实现类，通过反射进行实例化，然后执行 initialize 方法进行初始化，然后返回 FileSystem 对象就行了

如何通过配置获取到对应的 FileSystem 的实现类

  1. Class<?> clazz = getFileSystemClass(uri.getScheme(), conf);
  1. public static Class<? extends FileSystem> getFileSystemClass(String scheme, 
  2.                                              Configuration conf) throws IOException {
  3.     if (!FILE_SYSTEMS_LOADED) {
  4.         loadFileSystems();
  5.     }
  7.     LOGGER.debug("Looking for FS supporting {}", scheme);
  8.     Class<? extends FileSystem> clazz = null;
  9.     if (conf != null) {
  10.         // 这里的 property 为:   "fs.hdfs.impl"
  11.         String property = "fs." + scheme + ".impl";
  12.         LOGGER.debug("looking for configuration option {}", property);
  13.         clazz = conf.getClass(property, (Class)null);
  14.     } else {
  15.         LOGGER.debug("No configuration: skipping check for fs.{}.impl", scheme);
  16.     }
  18.     if (clazz == null) {
  19.         LOGGER.debug("Looking in service filesystems for implementation class");
  20.         clazz = (Class)SERVICE_FILE_SYSTEMS.get(scheme);
  21.     } else {
  22.         LOGGER.debug("Filesystem {} defined in configuration option", scheme);
  23.     }
  25.     if (clazz == null) {
  26.         throw new UnsupportedFileSystemException("No FileSystem for scheme \"" + 
  27.                                                  scheme + "\"");
  28.     } else {
  29.         LOGGER.debug("FS for {} is {}", scheme, clazz);
  30.         return clazz;
  31.     }
  32. }

上面代码是获取 FileSystem 的实现类，首先去配置文件中查找，是否配置了 fs.hdfs.impl 的实现类，如果配置了直接返回，如果没有的话，会查询系统自己自带的实现。也就是 SERVICE_FILE_SYSTEMS 这里系统默认的配置，一共有一下八种：
image.png
我们就可以知道 hdfs 对应的实现类为：org.apache.hadoop.hdfs.DistributedFileSystem

Initialize(URI uri, Configuration conf)

  1.  public void initialize(URI uri, Configuration conf) throws IOException {
  2.     super.initialize(uri, conf);
  3.     this.setConf(conf);
  5.     String host = uri.getHost();
  6.     if (host == null) {
  7.       throw new IOException("Incomplete HDFS URI, no host: "+ uri);
  8.     }else{
  9.         // 采用DFSClient进行通讯
  10.         this.dfs = new DFSClient(uri, conf, this.statistics);
  11.         this.uri = URI.create(uri.getScheme()+"://"+uri.getAuthority());
  12.         this.workingDir = getHomeDirectory();
  13.         this.storageStatistics = (DFSOpsCountStatistics) GlobalStorageStatistics.
  14.             INSTANCE.put(DFSOpsCountStatistics.NAME, new StorageStatisticsProvider() {
  15.             @Override
  16.             public StorageStatistics provide() {
  17.                 return new DFSOpsCountStatistics();
  18.             }
  19.             });
  20.     }
  21. }

DFSClient 主要是负责与 hadoop 通讯的客户端类

  1. public DFSClient(URI nameNodeUri, ClientProtocol rpcNamenode, Configuration conf, Statistics stats) throws IOException {
  2.     this.clientRunning = true;
  3.     this.r = new Random();
  4.     this.filesBeingWritten = new HashMap();
  5.     this.tracer = FsTracer.get(conf);
  6.     this.dfsClientConf = new DfsClientConf(conf);
  7.     this.conf = conf;
  8.     this.stats = stats;
  10.     // 构建socket通讯工厂类:  StandardSocketFactory
  11.     this.socketFactory = NetUtils.getSocketFactory(conf, ClientProtocol.class);
  13.     this.dtpReplaceDatanodeOnFailure = ReplaceDatanodeOnFailure.get(conf);
  14.     this.smallBufferSize = DFSUtilClient.getSmallBufferSize(conf);
  15.     this.dtpReplaceDatanodeOnFailureReplication = (short)conf.getInt("dfs.client.block.write.replace-datanode-on-failure.min-replication", 0);
  16.     if (LOG.isDebugEnabled()) {
  17.         LOG.debug("Sets dfs.client.block.write.replace-datanode-on-failure.min-replication to " + this.dtpReplaceDatanodeOnFailureReplication);
  18.     }
  20.     this.ugi = UserGroupInformation.getCurrentUser();
  21.     this.namenodeUri = nameNodeUri;
  23.     // 客户端的名字 
  24.     this.clientName = "DFSClient_" + this.dfsClientConf.getTaskId() + "_" + ThreadLocalRandom.current().nextInt() + "_" + Thread.currentThread().getId();
  26.     int numResponseToDrop = conf.getInt("dfs.client.test.drop.namenode.response.number", 0);
  28.     // 通讯协议 ClientProtocol : NameNodeProxiesClient
  29.     ProxyAndInfo<ClientProtocol> proxyInfo = null;
  30.     AtomicBoolean nnFallbackToSimpleAuth = new AtomicBoolean(false);
  31.     if (numResponseToDrop > 0) {
  32.         LOG.warn("dfs.client.test.drop.namenode.response.number is set to " + numResponseToDrop + ", this hacked client will proactively drop responses");
  33.         proxyInfo = NameNodeProxiesClient.createProxyWithLossyRetryHandler(conf, nameNodeUri, ClientProtocol.class, numResponseToDrop, nnFallbackToSimpleAuth);
  34.     }
  36.     if (proxyInfo != null) {
  37.         this.dtService = proxyInfo.getDelegationTokenService();
  38.         this.namenode = (ClientProtocol)proxyInfo.getProxy();
  39.     } else if (rpcNamenode != null) {
  40.         Preconditions.checkArgument(nameNodeUri == null);
  41.         this.namenode = rpcNamenode;
  42.         this.dtService = null;
  43.     } else {
  44.         Preconditions.checkArgument(nameNodeUri != null, "null URI");
  45.         proxyInfo = NameNodeProxiesClient.createProxyWithClientProtocol(conf, nameNodeUri, nnFallbackToSimpleAuth);
  46.         this.dtService = proxyInfo.getDelegationTokenService();
  47.         this.namenode = (ClientProtocol)proxyInfo.getProxy();
  48.     }
  50.     String[] localInterfaces = conf.getTrimmedStrings("dfs.client.local.interfaces");
  51.     this.localInterfaceAddrs = getLocalInterfaceAddrs(localInterfaces);
  52.     if (LOG.isDebugEnabled() && 0 != localInterfaces.length) {
  53.         LOG.debug("Using local interfaces [" + Joiner.on(',').join(localInterfaces) + "] with addresses [" + Joiner.on(',').join(this.localInterfaceAddrs) + "]");
  54.     }
  56.     Boolean readDropBehind = conf.get("dfs.client.cache.drop.behind.reads") == null ? null : conf.getBoolean("dfs.client.cache.drop.behind.reads", false);
  57.     Long readahead = conf.get("dfs.client.cache.readahead") == null ? null : conf.getLong("dfs.client.cache.readahead", 0L);
  58.     this.serverDefaultsValidityPeriod = conf.getLong("dfs.client.server-defaults.validity.period.ms", HdfsClientConfigKeys.DFS_CLIENT_SERVER_DEFAULTS_VALIDITY_PERIOD_MS_DEFAULT);
  59.     Boolean writeDropBehind = conf.get("dfs.client.cache.drop.behind.writes") == null ? null : conf.getBoolean("dfs.client.cache.drop.behind.writes", false);
  60.     this.defaultReadCachingStrategy = new CachingStrategy(readDropBehind, readahead);
  61.     this.defaultWriteCachingStrategy = new CachingStrategy(writeDropBehind, readahead);
  62.     this.clientContext = ClientContext.get(conf.get("dfs.client.context", "default"), this.dfsClientConf, conf);
  63.     if (this.dfsClientConf.getHedgedReadThreadpoolSize() > 0) {
  64.         initThreadsNumForHedgedReads(this.dfsClientConf.getHedgedReadThreadpoolSize());
  65.     }
  67.     this.initThreadsNumForStripedReads(this.dfsClientConf.getStripedReadThreadpoolSize());
  68.     this.saslClient = new SaslDataTransferClient(conf, DataTransferSaslUtil.getSaslPropertiesResolver(conf), TrustedChannelResolver.getInstance(conf), nnFallbackToSimpleAuth);
  69. }

DFSClient 是采用 socket 进行通讯，默认的通讯工厂类为: org.apache.hadoop.net.StandardSocketFactory

FileSystem 类是通过配置参数匹配到 fs.defaultFS 对应的 FileSystem 实现类，然后由实现类来进行操作

也就是，hdfs://master 匹配的 DistributedFileSystem 实现类，然后 DistributedFileSystem 的实现类里面封装了 DFSClient 类。然后后面的操作都是通过 DFSClient 与 hadoop 进行通讯