zoukankan      html  css  js  c++  java

  • MapReduce源码分析之LocatedFileStatusFetcher

    LocatedFileStatusFetcher是MapReduce中一个针对给定输入路径数组，使用配置的线程数目来获取数据块位置的实用类。它的主要作用就是利用多线程技术，每个线程对应一个任务，每个任务针对给定输入路径数组Path[]，解析出文件状态列表队列BlockingQueue<List<FileStatus>>。其中，输入数据输入路径只不过是一个Path，而输出数据则是文件状态列表队列BlockingQueue<List<FileStatus>>，文件状态FileStatus包含文件路径、长度、数据块大小、数据块副本数、文件所属用户、文件所属组、文件权限、文件最近修改时间、文件最近访问时间、是否为目录等信息。

            LocatedFileStatusFetcher采用了google并发编程包中的可监听Future模式ListenableFuture、可监听线程池ListeningExecutorService、回调函数FutureCallback，并使用了Java并发包中的可重入互斥锁ReentrantLock、多线程间协调通信工具Condition等实现了处理过程的多线程并发执行，并通过阻塞队列、回调函数等解决了目录的递归解析问题，是一种非常好的多线程环境下递归任务、可监听任务的实现。

            那么，MapReduce中LocatedFileStatusFetcher是如何实现的呢？本文将为你带来LocatedFileStatusFetcher的源码分析。

            首先，看下LocatedFileStatusFetcher的成员变量，代码如下：

    [java] view plain copy
     
     在CODE上查看代码片派生到我的代码片
    1. // 输入路径数组  
    2. private final Path[] inputDirs;  
    3. // 输入路径过滤器  
    4. private final PathFilter inputFilter;  
    5. // 配置信息  
    6. private final Configuration conf;  
    7. // 递归标志位  
    8. private final boolean recursive;  
    9. // 使用MR新API标志位  
    10. private final boolean newApi;  
    11.   
    12. // 底层线程池rawExec  
    13. private final ExecutorService rawExec;  
    14.   
    15. // 可监听线程池，基于底层线程池rawExec  
    16. private final ListeningExecutorService exec;  
    17.   
    18. // 文件状态列表阻塞队列  
    19. private final BlockingQueue<List<FileStatus>> resultQueue;  
    20.   
    21. // 无效输入路径错误相关IO异常列表  
    22. private final List<IOException> invalidInputErrors = new LinkedList<IOException>();  
    23.   
    24. // 处理原始输入路径回调函数  
    25. private final ProcessInitialInputPathCallback processInitialInputPathCallback =   
    26.     new ProcessInitialInputPathCallback();  
    27.   
    28. // 处理输入路径回调函数  
    29. private final ProcessInputDirCallback processInputDirCallback =   
    30.     new ProcessInputDirCallback();  
    31.   
    32. // 正在运行任务数原子计数器  
    33. private final AtomicInteger runningTasks = new AtomicInteger(0);  
    34.   
    35. // 可重入互斥锁  
    36. private final ReentrantLock lock = new ReentrantLock();  
    37.   
    38. // 多线程间协调通信工具Condition  
    39. private final Condition condition = lock.newCondition();  
    40.   
    41. // 任务执行过程中未知错误  
    42. private volatile Throwable unknownError;  

            LocatedFileStatusFetcher的成员变量比较多，但是大体可以分为以下几类：

            一、实现基本功能的输入、输出成员变量

            1、Path[] inputDirs：输入路径数组，其作为整体输入数据，每个最终路径都会被LocatedFileStatusFetcher解析成文件状态FileStatus；

            2、PathFilter inputFilter：输入路径过滤器，内置boolean accept(Path path)方法，对输入路径继续过滤，选取符合业务规则的路径；

            3、Configuration conf：配置信息，可以从中获取执行任务的线程数；

            4、boolean recursive：递归标志位，true表示对目录中的目录进行递归处理；

            5、boolean newApi：使用MR新API标志位；

            6、BlockingQueue<List<FileStatus>> resultQueue：文件状态列表阻塞队列，输出数据，即最终返回结果；

            二、多线程需要使用的成员变量

            1、ExecutorService rawExec：底层线程池；

            2、ListeningExecutorService exec：基于底层线程池rawExec的可监听线程池，利用google的并发编程包实现；

            3、ProcessInitialInputPathCallback processInitialInputPathCallback：处理原始输入路径回调函数；

            4、ProcessInputDirCallback processInputDirCallback：处理输入路径回调函数；

            5、AtomicInteger runningTasks：正在运行任务数原子计数器；

            6、ReentrantLock lock：ReentrantLock lock；

            7、Condition condition：多线程间协调通信工具；

            三、存放中间结果或异常的成员变量

            1、List<IOException> invalidInputErrors：无效输入路径错误相关IO异常列表；

            2、Throwable unknownError：任务执行过程中未知错误；

            再看下LocatedFileStatusFetcher的构造函数，代码如下：

    [java] view plain copy
     
     在CODE上查看代码片派生到我的代码片
    1.  /** 
    2.   * 构造函数 
    3.   *  
    4.   * @param conf configuration for the job 
    5.   * @param dirs the initial list of paths 
    6.   * @param recursive whether to traverse the patchs recursively 
    7.   * @param inputFilter inputFilter to apply to the resulting paths 
    8.   * @param newApi whether using the mapred or mapreduce API 
    9.   * @throws InterruptedException 
    10.   * @throws IOException 
    11.   */  
    12.  public LocatedFileStatusFetcher(Configuration conf, Path[] dirs,  
    13.      boolean recursive, PathFilter inputFilter, boolean newApi) throws InterruptedException,  
    14.      IOException {  
    15.     
    16. // 获取配置信息中的任务使用线程数numThreads，取参数mapreduce.input.fileinputformat.list-status.num-threads，参数未配置默认为1，  
    17. // 这里很明显应该会大于1  
    18.    int numThreads = conf.getInt(FileInputFormat.LIST_STATUS_NUM_THREADS,  
    19.        FileInputFormat.DEFAULT_LIST_STATUS_NUM_THREADS);  
    20.      
    21.    // 使用Executors.newFixedThreadPool方式构造线程池rawExec，线程个数为numThreads，并且设置为后台线程，线程名格式为GetFileInfo #数字  
    22.    rawExec = Executors.newFixedThreadPool(  
    23.        numThreads,  
    24.        new ThreadFactoryBuilder().setDaemon(true)  
    25.            .setNameFormat("GetFileInfo #%d").build());  
    26.      
    27.    // 使用MoreExecutors.listeningDecorator方式利用rawExec构造可监听线程池exec  
    28.    exec = MoreExecutors.listeningDecorator(rawExec);  
    29.      
    30.    // 初始化最终返回结果数据结构，即文件状态列表的链式阻塞队列resultQueue  
    31.    resultQueue = new LinkedBlockingQueue<List<FileStatus>>();  
    32.      
    33.    // 根据构造函数入参初始化类成员变量，这些成员变量包括输入路径数组、配置信息、递归标志位等全部是外部输入数据  
    34.    this.conf = conf;  
    35.    this.inputDirs = dirs;  
    36.    this.recursive = recursive;  
    37.    this.inputFilter = inputFilter;  
    38.    this.newApi = newApi;  
    39.  }  

            LocatedFileStatusFetcher构造函数逻辑很清晰，大体如下：

            1、首先获取配置信息中的任务使用线程数numThreads：

                  取参数mapreduce.input.fileinputformat.list-status.num-threads，参数未配置默认为1，这里很明显应该会大于1；

            2、使用Executors.newFixedThreadPool方式构造线程池rawExec，线程个数为numThreads，并且设置为后台线程，线程名格式为GetFileInfo #数字；

            3、使用MoreExecutors.listeningDecorator方式利用rawExec构造可监听线程池exec；

            4、初始化最终返回结果数据结构，即文件状态列表的链式阻塞队列resultQueue；

            5、根据构造函数入参初始化类成员变量，这些成员变量包括输入路径数组、配置信息、递归标志位等全部是外部输入数据。

            到了这里，您已经大概了解了LocatedFileStatusFetcher的结构。但是，您可能对Java并发编程或者google的可监听并发编程不是很了解，为此，这里有必要做个简单介绍，详细信息，读者可通过相关搜索引擎或书籍自行补脑。

            首先说下Future，Future表示一个异步计算任务，当任务完成时可以得到任务执行结果。您可能需要借助Future，通过启用另外的线程不断的查询任务状态，在任务完成时，获取任务执行结果通知或者展示给用户。而google的ListenableFuture顾名思义就是可以监听的Future，通过它在任务完成后自动调用配置好的回调函数，您就可以很方便的及时获取任务执行结果，采取下一步处理，这些回调函数统一都需要实现FutureCallback接口。

            再来说下可重入互斥锁ReentrantLock，它是一个独占锁，即互斥的，意即当前线程获取该锁后，其他线程此时如果想要获取该锁，就必须等待当前线程释放锁。何谓可重入呢？也很简单，当前线程获取该锁后，未释放前，还可以再次获得或者说进入该锁。

            第三个要说的是Condition，它是一个多线程间协调通信的工具类。通过其await()方法，当前线程会释放锁，进入睡眠，等待被唤醒；而其他线程借助Condition的signal()或signalAll()方法，则可以唤醒等待的线程，继续进行相关逻辑处理。

            最后一个要说的是ListeningExecutorService，它是一个可以返回ListenableFuture的接口，其借助Java并发包中的ExecutorService，就可以实现一个可监听的线程池，而本例中的底层线程池是Executors.newFixedThreadPool，它是一个可重用固定线程数的线程池,以共享的无界队列方式来运行这些线程。ListeningExecutorService中可以提交一些实现了Callable接口的线程任务，这些线程任务会被线程池调度，借助其call()方法完成任务执行逻辑。

            截至到目前，相信您应该LocatedFileStatusFetcher使用的并发编程的一些基础知识有一个大致了解了吧！

            好了，我们继续往下分析吧！看先LocatedFileStatusFetcher实现其核心功能的getFileStatuses()方法，代码如下：

    [java] view plain copy
     
     在CODE上查看代码片派生到我的代码片
    1.  /** 
    2.   * Start executing and return FileStatuses based on the parameters specified 
    3.   * 基于指定参数开始执行任务，并返回文件状态迭代器 
    4.   *  
    5.   * @return fetched file statuses 
    6.   * @throws InterruptedException 
    7.   * @throws IOException 
    8.   */  
    9.  public Iterable<FileStatus> getFileStatuses() throws InterruptedException,  
    10.      IOException {  
    11.    // Increment to make sure a race between the first thread completing and the  
    12.    // rest being scheduled does not lead to a termination.  
    13.     
    14. // 正在运行任务数原子计数器runningTasks加1  
    15.    runningTasks.incrementAndGet();  
    16.      
    17.    // 遍历输入路径inputDirs  
    18.    for (Path p : inputDirs) {  
    19.       
    20.      // 正在运行任务数原子计数器runningTasks加1  
    21.      runningTasks.incrementAndGet();  
    22.        
    23.      // 将处理原始输入路径任务ProcessInitialInputPathCallable提交到线程池exec中去执行，并获取可监听Future，即ListenableFuture，  
    24.      // 监听任务执行结果ProcessInitialInputPathCallable.Result  
    25.      ListenableFuture<ProcessInitialInputPathCallable.Result> future = exec  
    26.          .submit(new ProcessInitialInputPathCallable(p, conf, inputFilter));  
    27.        
    28.      // future中添加回调函数ProcessInitialInputPathCallback实例processInitialInputPathCallback  
    29.      Futures.addCallback(future, processInitialInputPathCallback);  
    30.    }  
    31.   
    32.    // 正在运行任务数原子计数器runningTasks减1  
    33.    runningTasks.decrementAndGet();  
    34.   
    35.    // 获取可重入互斥锁ReentrantLock实例lock  
    36.    lock.lock();  
    37.    try {  
    38.       
    39.      // 正在运行任务数原子计数器runningTasks不为0，且未知错误unknownError没有发生时  
    40.      while (runningTasks.get() != 0 && unknownError == null) {  
    41.         
    42.     // 等待所有任务运行完成  
    43.        condition.await();  
    44.      }  
    45.    } finally {  
    46.       
    47.      // 释放可重入互斥锁ReentrantLock  
    48.      lock.unlock();  
    49.    }  
    50.      
    51.    // 停止线程池exec  
    52.    this.exec.shutdownNow();  
    53.      
    54.    // 有未知错误unknownError的话处理未知错误  
    55.    if (this.unknownError != null) {  
    56.      if (this.unknownError instanceof Error) {  
    57.        throw (Error) this.unknownError;  
    58.      } else if (this.unknownError instanceof RuntimeException) {  
    59.        throw (RuntimeException) this.unknownError;  
    60.      } else if (this.unknownError instanceof IOException) {  
    61.        throw (IOException) this.unknownError;  
    62.      } else if (this.unknownError instanceof InterruptedException) {  
    63.        throw (InterruptedException) this.unknownError;  
    64.      } else {  
    65.        throw new IOException(this.unknownError);  
    66.      }  
    67.    }  
    68.      
    69.    // 有无效路径错误invalidInputErrors的话处理无效路径错误  
    70.    if (this.invalidInputErrors.size() != 0) {  
    71.      if (this.newApi) {  
    72.        throw new org.apache.hadoop.mapreduce.lib.input.InvalidInputException(  
    73.            invalidInputErrors);  
    74.      } else {  
    75.        throw new InvalidInputException(invalidInputErrors);  
    76.      }  
    77.    }  
    78.      
    79.    // 将结果队列resultQueue转换成迭代器并返回  
    80.    return Iterables.concat(resultQueue);  
    81.  }  

            getFileStatuses()方法的执行逻辑大体如下：

            1、首先，正在运行任务数原子计数器runningTasks加1，这个是针对主线程任务的计数；

            2、接着遍历输入路径inputDirs：

                  2.1、正在运行任务数原子计数器runningTasks加1，这个是针对每个待处理输入路径的子线程任务的计数；

                  2.2、将处理原始输入路径任务提交到线程池exec中去执行，并获取可监听Future，即ListenableFuture，监听任务执行结果：

                           这里，原始输入路径任务为ProcessInitialInputPathCallable，它实现了Callable接口，并有一个内部静态类Result，作为任务处理结果，稍后我们对它做详细分析；

                  2.3、future中添加回调函数，待任务处理完成后通过回调函数做进一步处理：

                           这里，回调函数为ProcessInitialInputPathCallback，即处理原始输入路径的回调函数，其实现了FutureCallback接口，并对上述任务执行结果ProcessInitialInputPathCallable.Result进行回调处理；

            3、正在运行任务数原子计数器runningTasks减1，这个是针对主线程任务的计数，含义是主线程任务在其它子线程任务全部执行完成的情况下可以标记为处理完成；

            4、获取可重入互斥锁ReentrantLock实例lock；

            5、当正在运行任务数原子计数器runningTasks不为0，且未知错误unknownError没有发生时，通过condition.await()方法，释放当前锁，进入睡眠，等待被唤醒，直到其他线程唤醒它，并且正在运行任务数原子计数器runningTasks为0，或者未知错误unknownError发生，才说明所有任务已执行完成或不得不终止运行；

            6、释放可重入互斥锁ReentrantLock；

            7、停止线程池exec；

            8、有未知错误unknownError的话处理未知错误；

            9、有无效路径错误invalidInputErrors的话处理无效路径错误；

            10、将结果队列resultQueue转换成迭代器并返回。

            我们先说下这个原始输入路径任务为ProcessInitialInputPathCallable，它实现了Callable接口，并有一个内部静态类Result，作为任务处理结果，代码如下：

    [java] view plain copy
     
     在CODE上查看代码片派生到我的代码片
    1. /** 
    2.  * Processes an initial Input Path pattern through the globber and PathFilter 
    3.  * to generate a list of files which need further processing. 
    4.  * 通过globber和路径过滤器PathFilter处理一个初始输入路径模式，产生一个需要进一步处理的文件列表。 
    5.  */  
    6. private static class ProcessInitialInputPathCallable implements  
    7.     Callable<ProcessInitialInputPathCallable.Result> {  
    8.   
    9. / 待处理路径  
    10.   private final Path path;  
    11.     
    12.   // 配置信息  
    13.   private final Configuration conf;  
    14.     
    15.   // 输入路径过滤器  
    16.   private final PathFilter inputFilter;  
    17.   
    18.   public ProcessInitialInputPathCallable(Path path, Configuration conf,  
    19.       PathFilter pathFilter) {  
    20.     this.path = path;  
    21.     this.conf = conf;  
    22.     this.inputFilter = pathFilter;  
    23.   }  
    24.   
    25.   @Override  
    26.   public Result call() throws Exception {  
    27.       
    28.     // 构造任务结果Result实例result  
    29.     Result result = new Result();  
    30.       
    31.     // 从路径path中获取文件系统FileSystem实例fs  
    32.     FileSystem fs = path.getFileSystem(conf);  
    33.       
    34.     // 设置任务结果Result实例result中的fs变量  
    35.     result.fs = fs;  
    36.       
    37.     // 通过文件系统FileSystem实例fs的globStatus()方法，将路径path依据输入路径过滤器inputFilter解析成文件状态FileStatus数组matches  
    38.     FileStatus[] matches = fs.globStatus(path, inputFilter);  
    39.       
    40.     if (matches == null) {  
    41.     // 如果文件状态FileStatus数组matches为null，说明路径根本不存在，将IO异常通过addError()方法添加到result中  
    42.       result.addError(new IOException("Input path does not exist: " + path));  
    43.     } else if (matches.length == 0) {  
    44.     // 如果文件状态FileStatus数组matches不为null，但长度为0，说明路径存在但是没有通过过滤器过滤规则，将IO异常通过addError()方法添加到result中  
    45.       result.addError(new IOException("Input Pattern " + path  
    46.           + " matches 0 files"));  
    47.     } else {  
    48.         
    49.     // 将符合过滤规则的文件状态FileStatus数组matches赋值给任务结果result的matchedFileStatuses  
    50.       result.matchedFileStatuses = matches;  
    51.     }  
    52.     return result;  
    53.   }  
    54.   
    55.   private static class Result {  
    56.       
    57.     // 处理过程中发生的IO异常列表errors  
    58.     private List<IOException> errors;  
    59.       
    60.     // 匹配的文件状态数组matchedFileStatuses  
    61.     private FileStatus[] matchedFileStatuses;  
    62.       
    63.     // 文件系统实例  
    64.     private FileSystem fs;  
    65.   
    66.     // 添加IO异常到errors列表  
    67.     void addError(IOException ioe) {  
    68.       if (errors == null) {  
    69.         errors = new LinkedList<IOException>();  
    70.       }  
    71.       errors.add(ioe);  
    72.     }  
    73.   }  
    74. }  

            我们看到，它有三个成员变量，待处理路径path、配置信息conf、输入路径过滤器inputFilter，并且构造方法就是简单的根据入参初始化这三个成员变量。ProcessInitialInputPathCallable还提供了一个表示任务结果的内部静态类Result，它也有三个成员变量，处理过程中发生的IO异常列表errors、匹配的文件状态匹配的文件状态数组matchedFileStatuses数组matchedFileStatuses、文件系统实例fs，并提供了添加IO异常到errors列表的addError()方法。

            重点看下 ProcessInitialInputPathCallable的call()方法，它是任务得以执行的入口方法，其大体逻辑如下：

            1、构造任务结果Result实例result；

            2、从路径path中获取文件系统FileSystem实例fs；

            3、设置任务结果Result实例result中的fs变量；

            4、通过文件系统FileSystem实例fs的globStatus()方法，将路径path依据输入路径过滤器inputFilter解析成文件状态FileStatus数组matches：

                  这里，限于篇幅及主题明确性，我们不做过多介绍，你只要知道它的主要作用就行，我们将在单线程处理的博文中进行详细介绍；

            5、根据matches分别处理任务执行结果：

                  5.1、如果文件状态FileStatus数组matches为null，说明路径根本不存在，将IO异常通过addError()方法添加到result中；

                  5.2、如果文件状态FileStatus数组matches不为null，但长度为0，说明路径存在但是没有通过过滤器过滤规则，将IO异常通过addError()方法添加到result中；

                  5.3、否则将符合过滤规则的文件状态FileStatus数组matches赋值给任务结果result的matchedFileStatuses；

            6、返回任务结果result。

            原始路径处理任务执行完成的回调函数则是通过ProcessInitialInputPathCallback来定义的，代码如下：

    [java] view plain copy
     
     在CODE上查看代码片派生到我的代码片
    1. /** 
    2.  * The callback handler to handle results generated by 
    3.  * {@link ProcessInitialInputPathCallable} 
    4.  *  
    5.  */  
    6. private class ProcessInitialInputPathCallback implements  
    7.     FutureCallback<ProcessInitialInputPathCallable.Result> {  
    8.   
    9. / 任务执行成功时：不是说结果对错，而是说任务能完整的执行下来  
    10.   @Override  
    11.   public void onSuccess(ProcessInitialInputPathCallable.Result result) {  
    12.     try {  
    13.         
    14.     // 如果任务结果有IO异常  
    15.       if (result.errors != null) {  
    16.           
    17.         // 通过registerInvalidInputError()方法，将IO异常列表errors全部添加到无效输入路径错误相关IO异常列表invalidInputErrors中  
    18.         registerInvalidInputError(result.errors);  
    19.       }  
    20.         
    21.       // 如果任务结果得到了匹配的文件状态数组  
    22.       if (result.matchedFileStatuses != null) {  
    23.           
    24.         // 遍历匹配的文件状态数组matchedFileStatuses，取出每个文件状态FileStatus实例matched，做以下处理：  
    25.         for (FileStatus matched : result.matchedFileStatuses) {  
    26.             
    27.         // 正在运行任务数原子计数器runningTasks加1，这里标识的是子任务数加1  
    28.           runningTasks.incrementAndGet();  
    29.             
    30.           // 将处理输入路径任务ProcessInputDirCallable提交到线程池exec中去执行，并获取可监听Future，即ListenableFuture，  
    31.           // 监听任务执行结果ProcessInputDirCallable.Result  
    32.           ListenableFuture<ProcessInputDirCallable.Result> future = exec  
    33.               .submit(new ProcessInputDirCallable(result.fs, matched,  
    34.                   recursive, inputFilter));  
    35.             
    36.           // future中添加回调函数ProcessInputDirCallback实例processInputDirCallback  
    37.           Futures.addCallback(future, processInputDirCallback);  
    38.         }  
    39.       }  
    40.         
    41.       // 解析原始路径的任务完成，调用decrementRunningAndCheckCompletion()做后续处理工作：  
    42.       // 正在运行任务数原子计数器减1，并判断是否为0，为0，说明全部任务运行完成，通过condition.signal()通知主线程进行处理  
    43.       decrementRunningAndCheckCompletion();  
    44.     } catch (Throwable t) { // Exception within the callback  
    45.         
    46.     // 有异常的话，调用registerError()方法，重置任务执行过程中未知错误unknownError，并通过condition.signal()通知主线程，  
    47.       // 有未知错误发生，交由主线程处理（主线程在有位置错误unknownError的情况下会结束整个流程），潜台词就是第一次发生未知错误时，不会通知主线程结束整个流程，后续再发生时才会通知  
    48.       registerError(t);  
    49.     }  
    50.   }  
    51.   
    52.   // 任务执行失败时：不是说结果对错，而是说任务不能完整的执行下来  
    53.   @Override  
    54.   public void onFailure(Throwable t) {  
    55.     // Any generated exceptions. Leads to immediate termination.  
    56.     // 调用registerError()方法，重置任务执行过程中未知错误unknownError，并通过condition.signal()通知主线程，  
    57.     // 有未知错误发生，交由主线程处理（主线程在有位置错误unknownError的情况下会结束整个流程），潜台词就是第一次发生未知错误时，不会通知主线程结束整个流程，后续再发生时才会通知  
    58.     registerError(t);  
    59.   }  
    60. }  

              原始路径处理任务执行完成的回调函数ProcessInitialInputPathCallback实现了FutureCallback接口，并对原始路径处理任务结果ProcessInitialInputPathCallable.Result进行检测处理，主要分为两种情况：

            1、任务执行成功时：不是说结果对错，而是说任务能完整的执行下来

                  通过onSuccess()方法来处理，大体逻辑如下：

                  1.1、如果任务结果有IO异常，通过registerInvalidInputError()方法，将IO异常列表errors全部添加到无效输入路径错误相关IO异常列表invalidInputErrors中；

                  1.2、如果任务结果得到了匹配的文件状态数组，遍历匹配的文件状态数组matchedFileStatuses，取出每个文件状态FileStatus实例matched，做以下处理：

                            1.2.1、正在运行任务数原子计数器runningTasks加1，这里标识的是子任务数加1；

                            1.2.2、将处理输入路径任务ProcessInputDirCallable提交到线程池exec中去执行，并获取可监听Future，即ListenableFuture，监听任务执行结果ProcessInputDirCallable.Result：

                                         这里的ProcessInputDirCallable任务，主要是为给定文件状态FileStatus获取数据块位置，如有必要（即需要递归目录进行处理），添加额外的路径到处理队列，后续递归处理，而给定文件状态FileStatus则是通过解析原始路径任务ProcessInitialInputPathCallable来获得的；

                            1.2.3、future中添加回调函数ProcessInputDirCallback实例processInputDirCallback；

                 1.3、解析原始路径的任务完成，调用decrementRunningAndCheckCompletion()做后续处理工作：正在运行任务数原子计数器减1，并判断是否为0，为0，说明全部任务运行完成，通过condition.signal()通知主线程进行处理；

                  需要说明的是，上述逻辑执行期间，如果有Throwable发生，则会调用registerError()方法，至于如何处理，参见2任务执行失败时的处理；

            2、任务执行失败时：不是说结果对错，而是说任务不能完整的执行下来

                  通过onFailure()方法来处理，调用registerError()方法，重置任务执行过程中未知错误unknownError，并通过condition.signal()通知主线程，有未知错误发生，交由主线程处理（主线程在有位置错误unknownError的情况下会结束整个流程），潜台词就是第一次发生未知错误时，不会通知主线程结束整个流程，后续再发生时才会通知。

            decrementRunningAndCheckCompletion()方法代码如下：

    [java] view plain copy
     
     在CODE上查看代码片派生到我的代码片
    1.  private void decrementRunningAndCheckCompletion() {  
    2.      
    3. // 获取可重入互斥锁lock  
    4. lock.lock();  
    5.    try {  
    6.       
    7.      // 正在运行任务数原子计数器减1，并判断是否为0，为0，说明全部任务运行完成，通过condition.signal()通知主线程进行处理  
    8.      if (runningTasks.decrementAndGet() == 0) {  
    9.        condition.signal();  
    10.      }  
    11.    } finally {  
    12.       
    13.      // 释放可重入互斥锁lock  
    14.      lock.unlock();  
    15.    }  
    16.  }  

            而registerError()方法代码如下：

    [java] view plain copy
     
     在CODE上查看代码片派生到我的代码片
    1.  /** 
    2.   * Register fatal errors - example an IOException while accessing a file or a 
    3.   * full exection queue 
    4.   */  
    5.  private void registerError(Throwable t) {  
    6.      
    7. // 获取可重入互斥锁lock    
    8. lock.lock();  
    9.    try {  
    10.       
    11.      // 重置任务执行过程中未知错误unknownError，并通过condition.signal()通知主线程，  
    12.      // 有未知错误发生，交由主线程处理（主线程在有位置错误unknownError的情况下会结束整个流程）  
    13.      if (unknownError != null) {  
    14.        unknownError = t;  
    15.        condition.signal();  
    16.      }  
    17.   
    18.    } finally {  
    19.       
    20.      // 释放可重入互斥锁lock  
    21.      lock.unlock();  
    22.    }  
    23.  }  

            两个方法功能很明确，注释也很详细，且上面已经提到过，这里不再赘述！

            接下来，我们再看下ProcessInputDirCallable任务，它主要是为给定文件状态FileStatus获取数据块位置，如有必要（即需要递归目录进行处理），添加额外的路径到处理队列，后续递归处理，其实现如下：

    [java] view plain copy
     
     在CODE上查看代码片派生到我的代码片
    1. /** 
    2.  * Retrieves block locations for the given @link {@link FileStatus}, and adds 
    3.  * additional paths to the process queue if required. 
    4.  * 为给定文件状态获取数据块位置，如有必要，添加额外的路径到处理队列。 
    5.  */  
    6. private static class ProcessInputDirCallable implements  
    7.     Callable<ProcessInputDirCallable.Result> {  
    8.   
    9. / 文件系统实例  
    10.   private final FileSystem fs;  
    11.     
    12.   // 文件状态实例  
    13.   private final FileStatus fileStatus;  
    14.     
    15.   // 递归标志位  
    16.   private final boolean recursive;  
    17.     
    18.   // 输入路径过滤器  
    19.   private final PathFilter inputFilter;  
    20.   
    21.   // 构造函数  
    22.   ProcessInputDirCallable(FileSystem fs, FileStatus fileStatus,  
    23.       boolean recursive, PathFilter inputFilter) {  
    24.     this.fs = fs;  
    25.     this.fileStatus = fileStatus;  
    26.     this.recursive = recursive;  
    27.     this.inputFilter = inputFilter;  
    28.   }  
    29.   
    30.   // 任务执行主方法  
    31.   @Override  
    32.   public Result call() throws Exception {  
    33.       
    34.     // 构造结果Result  
    35.     Result result = new Result();  
    36.       
    37.     // 初始化结果中的文件系统实例fs  
    38.     result.fs = fs;  
    39.   
    40.     // 如果文件状态fileStatus对应为目录  
    41.     if (fileStatus.isDirectory()) {  
    42.         
    43.     // 通过文件系统FileSystem实例fs的listLocatedStatus()方法获取fileStatus对应的带数据块位置信息文件状态迭代器iter  
    44.     RemoteIterator<LocatedFileStatus> iter = fs  
    45.           .listLocatedStatus(fileStatus.getPath());  
    46.         
    47.     // 通过迭代器iter遍历每个带数据块位置信息文件状态stat  
    48.     while (iter.hasNext()) {  
    49.         LocatedFileStatus stat = iter.next();  
    50.           
    51.         // 通过输入路径过滤器的accept()方法进行过滤  
    52.         if (inputFilter.accept(stat.getPath())) {  
    53.             
    54.         // 如果需要递归，且stat为目录  
    55.         if (recursive && stat.isDirectory()) {  
    56.               
    57.             // 添加到结果result的dirsNeedingRecursiveCalls列表  
    58.             result.dirsNeedingRecursiveCalls.add(stat);  
    59.           } else {  
    60.               
    61.             // 否则添加到结果result的locatedFileStatuses列表  
    62.             result.locatedFileStatuses.add(stat);  
    63.           }  
    64.         }  
    65.       }  
    66.     } else {  
    67.         
    68.     // 如果文件状态fileStatus对应为文件，直接添加到结果result的locatedFileStatuses列表  
    69.       result.locatedFileStatuses.add(fileStatus);  
    70.     }  
    71.     return result;  
    72.   }  
    73.   
    74.   // 处理结果  
    75.   private static class Result {  
    76.       
    77.     // 已处理完的文件状态链表locatedFileStatuses  
    78.     private List<FileStatus> locatedFileStatuses = new LinkedList<FileStatus>();  
    79.     // 需要递归的文件状态链表dirsNeedingRecursiveCalls  
    80.     private List<FileStatus> dirsNeedingRecursiveCalls = new LinkedList<FileStatus>();  
    81.       
    82.     // 文件系统实例  
    83.     private FileSystem fs;  
    84.   }  
    85. }  

            首先，ProcessInputDirCallable内部有四个成员变量，分别是文件系统实例fs、文件状态实例fileStatus、递归标志位recursive、输入路径过滤器inputFilter，意义都很明确，而构造方法也是根据入参初始化这四个成员变量，不再详述。

            任务执行结果由其静态内部类Result来表示，它包含三个成员变量，已处理完的文件状态链表locatedFileStatuses、需要递归再处理的文件状态链表dirsNeedingRecursiveCalls、文件系统实例fs，意义都很明确，不再详述。

            接下来，我们再看下任务执行的入口方法call()的运行逻辑，归纳如下：

            1、构造任务运行结果Result实例result；

            2、初始化结果中的文件系统实例fs；

            3、如果文件状态fileStatus对应为目录：

                  3.1、通过文件系统FileSystem实例fs的listLocatedStatus()方法获取fileStatus对应的带数据块位置信息文件状态迭代器iter：

                           文件系统FileSystem实例fs的listLocatedStatus()方法我们会在单线程任务重点描述，这里你只要记住它的主要功能就是根据文件状态获取数据块位置信息，并返回带数据块位置信息文件状态迭代器，而带数据块位置信息文件状态LocatedFileStatus是文件状态FileStatus的子类，其内部多了一个成员变量BlockLocation[] locations，表示文件所含数据块的位置信息；

                  3.2、通过迭代器iter遍历每个带数据块位置信息文件状态stat：通过输入路径过滤器的accept()方法进行过滤，如果需要递归，且stat为目录，添加到结果result的dirsNeedingRecursiveCalls列表，否则添加到结果result的locatedFileStatuses列表；

            4、如果文件状态fileStatus对应为文件，直接添加到结果result的locatedFileStatuses列表；

            5、返回任务执行结果result。

            如同上面提到的解析原始路径任务ProcessInitialInputPathCallable一样，ProcessInputDirCallable任务也需要在任务执行完成后有回调函数做进一步处理，而这个回调函数是通过ProcessInputDirCallback来实现的，代码如下：

    [java] view plain copy
     
     在CODE上查看代码片派生到我的代码片
    1. /** 
    2.  * The callback handler to handle results generated by 
    3.  * {@link ProcessInputDirCallable}. This populates the final result set. 
    4.  *  
    5.  */  
    6. private class ProcessInputDirCallback implements  
    7.     FutureCallback<ProcessInputDirCallable.Result> {  
    8.   
    9. / 任务执行完成时：不是说结果对错，而是说任务能完整的执行下来  
    10.   @Override  
    11.   public void onSuccess(ProcessInputDirCallable.Result result) {  
    12.     try {  
    13.         
    14.     // 如果任务执行结果中已处理完的文件状态链表locatedFileStatuses有数据的话，将其添加到最终返回结果队列resultQueue中  
    15.       if (result.locatedFileStatuses.size() != 0) {  
    16.         resultQueue.add(result.locatedFileStatuses);  
    17.       }  
    18.         
    19.       // 如果任务执行结果中需要递归再处理的文件状态链表dirsNeedingRecursiveCalls，再次提交ProcessInputDirCallable任务到线程池ProcessInputDirCallable，  
    20.       // runningTasks计数器加1，添加回调函数ProcessInputDirCallback，以实现迭代处理  
    21.       if (result.dirsNeedingRecursiveCalls.size() != 0) {  
    22.         for (FileStatus fileStatus : result.dirsNeedingRecursiveCalls) {  
    23.           runningTasks.incrementAndGet();  
    24.           ListenableFuture<ProcessInputDirCallable.Result> future = exec  
    25.               .submit(new ProcessInputDirCallable(result.fs, fileStatus,  
    26.                   recursive, inputFilter));  
    27.           Futures.addCallback(future, processInputDirCallback);  
    28.         }  
    29.       }  
    30.         
    31.       // 解析路径的任务完成，调用decrementRunningAndCheckCompletion()做后续处理工作：  
    32.       // 正在运行任务数原子计数器减1，并判断是否为0，为0，说明全部任务运行完成，通过condition.signal()通知主线程进行处理  
    33.       decrementRunningAndCheckCompletion();  
    34.     } catch (Throwable t) { // Error within the callback itself.  
    35.         
    36.     // 有异常的话，调用registerError()方法，重置任务执行过程中未知错误unknownError，并通过condition.signal()通知主线程，  
    37.       // 有未知错误发生，交由主线程处理（主线程在有位置错误unknownError的情况下会结束整个流程），潜台词就是第一次发生未知错误时，不会通知主线程结束整个流程，后续再发生时才会通知  
    38.       registerError(t);  
    39.     }  
    40.   }  
    41.   
    42.   // 任务执行失败时：不是说结果对错，而是说任务不能完整的执行下来  
    43.   @Override  
    44.   public void onFailure(Throwable t) {  
    45.     // Any generated exceptions. Leads to immediate termination.  
    46.       
    47.     // 调用registerError()方法，重置任务执行过程中未知错误unknownError，并通过condition.signal()通知主线程，  
    48.     // 有未知错误发生，交由主线程处理（主线程在有位置错误unknownError的情况下会结束整个流程），潜台词就是第一次发生未知错误时，不会通知主线程结束整个流程，后续再发生时才会通知  
    49.     registerError(t);  
    50.   }  
    51. }  

            ProcessInputDirCallbacky如同上面介绍的ProcessInitialInputPathCallback一样，也分成功、失败两种情况分别进行处理：

            1、任务执行成功时：不是说结果对错，而是说任务能完整的执行下来

                  通过onSuccess()方法来处理，大体逻辑如下：

                  1.1、如果任务执行结果中已处理完的文件状态链表locatedFileStatuses有数据的话，将其添加到最终返回结果队列resultQueue中；

                  1.2、如果任务执行结果中需要递归再处理的文件状态链表dirsNeedingRecursiveCalls，再次提交ProcessInputDirCallable任务到线程池ProcessInputDirCallable，runningTasks计数器加1，添加回调函数ProcessInputDirCallback，以实现迭代处理；

                 1.3、解析路径的任务完成，调用decrementRunningAndCheckCompletion()做后续处理工作：正在运行任务数原子计数器减1，并判断是否为0，为0，说明全部任务运行完成，通过condition.signal()通知主线程进行处理；

                  需要说明的是，上述逻辑执行期间，如果有Throwable发生，则会调用registerError()方法，至于如何处理，参见2任务执行失败时的处理；

            2、任务执行失败时：不是说结果对错，而是说任务不能完整的执行下来

                  通过onFailure()方法来处理，调用registerError()方法，重置任务执行过程中未知错误unknownError，并通过condition.signal()通知主线程，有未知错误发生，交由主线程处理（主线程在有位置错误unknownError的情况下会结束整个流程），潜台词就是第一次发生未知错误时，不会通知主线程结束整个流程，后续再发生时才会通知。

            至此，整个LocatedFileStatusFetcher的源码分析介绍完毕。

            总结

            LocatedFileStatusFetcher通过多线程的方式，实现了针对给定输入路径数组，使用配置的线程数目来获取数据块位置的核心功能。它通过google的可监听并发技术ListenableFuture、ListeningExecutorService，实现了两层级别的子任务的并发执行、结果监听与回调处理，第一层任务是ProcessInitialInputPathCallable，根据输入路径获取对应文件状态，第二层任务是ProcessInputDirCallable，根据文件状态获取带数据块位置信息的文件状态，每层任务都有一个静态内部类Result来很好的抽象任务运行结果。每层任务都有一个回调函数，在获得任务执行结果后做进一步处理，并且第一层任务执行结束后，在回调函数里提交第二层任务，且第二层任务会根据是否递归的标志位和实际路径情况，在在回调函数里决定是否递归提交第二层任务。另外，LocatedFileStatusFetcher还使用了可重入互斥锁ReentrantLock、多线程间协调通信工具Condition来解决多线程之间的并发同步问题，特别是主任务线程与子任务线程间的主从协调、通信等。不得不说，LocatedFileStatusFetcher是多线程处理递归任务一种非常好的实现，值得我们借鉴和学习！
  • 相关阅读:
    AOP
    session的一些原理
    HttpContext.Cache和Application的区别
    什么是COM组件
    MS—SQL数据库索引的应用
    会话状态(转)
    LINQ体验(6)——LINQ语句之Join和Order By(转)
    永远不停止学习，才是程序员最大的财富，欢迎加入CMS研究室
    面向对象
    从瀑布型开发到迭代型开发的转变
  • 原文地址：https://www.cnblogs.com/jirimutu01/p/5556353.html
Copyright © 2011-2022 走看看