首页 > 代码库 > 多线程IO操作(扫描文件夹并计算总大小)

多线程IO操作(扫描文件夹并计算总大小)

2024-07-03 07:38:22   223人阅读

 场景为，给到一个硬盘上文件或文件夹，（当然文件夹时，多线程的优势才能越发体现出来），得到该文件或文件夹的大小和计算该结果所需要的时间。

        

            首先是单线程下的例子，这个可难不倒大家，代码如下：

01public class TotalFileSizeSequential {
02  private long getTotalSizeOfFilesInDir(final File file) {
03    if (file.isFile()) return file.length();
04     
05    final File[] children = file.listFiles();
06    long total = 0;
07    if (children != null)
08      for(final File child : children) 
09        total += getTotalSizeOfFilesInDir(child);      
10    return total;
11  }
12   
13  public static void main(final String[] args) {
14    final long start = System.nanoTime();
15    final long total = new TotalFileSizeSequential()
16      .getTotalSizeOfFilesInDir(new File("D:/idea_ws"));
17    final long end = System.nanoTime();
18    System.out.println("Total Size: " + total);
19    System.out.println("Time taken: " + (end - start)/1.0e9);
20  }
21}

        上述代码在我的机器上（win7,8g,i5）多次运行后的均值为：文件夹大小为276590351字节，即263M；耗时大概 0.25s

1Total Size: 276589881
2Time taken: 0.258706999

       


        至此，我们当然希望通过多线程扫描计算的方式，来更快的得到这个文件夹的大小，于是有个这个看似设计不错的，native版的代码：

01public class NaivelyConcurrentTotalFileSize {
02  private long getTotalSizeOfFilesInDir(
03    final ExecutorService service, final File file) 
04    throws InterruptedException, ExecutionException, TimeoutException {
05    if (file.isFile()) return file.length();
06     
07    long total = 0;
08    final File[] children = file.listFiles();
09     
10    if (children != null) {
11      final List<Future<Long>> partialTotalFutures = 
12        new ArrayList<Future<Long>>();      
13      for(final File child : children) {
14        partialTotalFutures.add(service.submit(new Callable<Long>() {
15          public Long call() throws InterruptedException, 
16            ExecutionException, TimeoutException { 
17            return getTotalSizeOfFilesInDir(service, child); 
18          }
19        }));
20      }
21       
22      for(final Future<Long> partialTotalFuture : partialTotalFutures) 
23        total += partialTotalFuture.get(100, TimeUnit.SECONDS);
24  }
25     
26    return total;
27  }
28 
29  private long getTotalSizeOfFile(final String fileName)
30    throws InterruptedException, ExecutionException, TimeoutException {
31      final ExecutorService service = Executors.newFixedThreadPool(100);
32      try {
33        return getTotalSizeOfFilesInDir(service, new File(fileName));
34      } finally {
35          service.shutdown();
36      }
37  }
38     
39  public static void main(final String[] args) 
40    throws InterruptedException, ExecutionException, TimeoutException {
41    final long start = System.nanoTime();
42    final long total = new NaivelyConcurrentTotalFileSize()
43      .getTotalSizeOfFile("D:/idea_ws");
44    final long end = System.nanoTime();
45    System.out.println("Total Size: " + total);
46    System.out.println("Time taken: " + (end - start)/1.0e9);
47  }
48}

            额，运行的结果出乎意料，多次运行该多线程代码，去扫描计算这个263M的文件夹时，仅有几次能运行成功，大部分时候，都是死锁，导致程序崩溃，根本无法计算文件夹大小。

            这段程序在getTotalSizeOfFilesInDir(final ExecutorService service, final File file) 方法中，有阻塞线程池的操作。每当扫描到一个子目录的时候，它就将该任务调度给其他的线程。这个逻辑看上去没有错，一旦它调度完了所有任务，该函数就等待任何一个任务的响应。即是代码中的 total += partialTotalFuture.get(100, TimeUnit.SECONDS);这一行。

            当然，当给到扫描的目录不是很深，文件不很多时，这样做确实能很快的得到结果，否则，就会在该行卡主，一直等待响应，若是没有我们设置的100的超时操作，这将演变成一种潜在的“线程诱发型死锁（Pool Include DeadLock）”

            此时，我们就要考虑如何改善这段代码了，目标是：计算各子目录大小的任务分配给不同线程，当我们等待其他任务响应时，又不会占住当前的主调线程（避免死锁）。

            下面，是第一种解决方案：每个任务都返回给定目录的直接子目录列表，而不是去返回计算的大小。这样的好处是，使线程被堵住的时间不会超过扫描给定目录的直接子目录的时间。在返回直接子目录的同时，计算出目录中的文件的大小一并返回。
            

            那么，第一个修复版本的代码如下：

01public class ConcurrentTotalFileSize {
02  class SubDirectoriesAndSize {
03    final public long size;
04    final public List<File> subDirectories;
05    public SubDirectoriesAndSize(
06      final long totalSize, final List<File> theSubDirs) {
07      size = totalSize;
08      subDirectories = Collections.unmodifiableList(theSubDirs);
09    }
10  }
11 
12  private SubDirectoriesAndSize getTotalAndSubDirs(final File file) {
13    long total = 0;
14  /*
15                list的作用：
16                1.先存放入参file，根据file，得到它下面的SubDirectoriesAndSize对象，即总的文件大小和总的文件夹的ls。根据getTotalSubDirs(final File file)获取
17                2.清空上一步的入参file，用于存放SubDirectoriesAndSize.dirs
18             */
19    final List<File> subDirectories = new ArrayList<File>();
20    if(file.isDirectory()) {
21      final File[] children = file.listFiles();
22      if (children != null)
23        for(final File child : children) {
24          if (child.isFile())
25            total += child.length();
26          else
27            subDirectories.add(child);
28        }
29    }
30    return new SubDirectoriesAndSize(total, subDirectories);
31  }
32 
33  private long getTotalSizeOfFilesInDir(final File file) 
34    throws InterruptedException, ExecutionException, TimeoutException {
35    final ExecutorService service = Executors.newFixedThreadPool(100);
36    try {
37      long total = 0;
38      final List<File> directories = new ArrayList<File>();
39      directories.add(file);
40      while(!directories.isEmpty()) {
41        final List<Future<SubDirectoriesAndSize>> partialResults = 
42          new ArrayList<Future<SubDirectoriesAndSize>>();
43        for(final File directory : directories) {
44          partialResults.add(
45            service.submit(new Callable<SubDirectoriesAndSize>() {
46            public SubDirectoriesAndSize call() {
47              return getTotalAndSubDirs(directory);
48            }
49          }));
50        }
51        directories.clear();     
52        for(final Future<SubDirectoriesAndSize> partialResultFuture : 
53          partialResults) {
54          final SubDirectoriesAndSize subDirectoriesAndSize = 
55            partialResultFuture.get(100, TimeUnit.SECONDS);
56          directories.addAll(subDirectoriesAndSize.subDirectories);
57          total += subDirectoriesAndSize.size;
58        }
59      }
60      return total;
61    } finally {
62      service.shutdown();
63    }
64  }
65 
66  public static void main(final String[] args) 
67    throws InterruptedException, ExecutionException, TimeoutException {
68    final long start = System.nanoTime();
69    final long total = new ConcurrentTotalFileSize()
70      .getTotalSizeOfFilesInDir(new File("D:/idea_ws"));
71    final long end = System.nanoTime();
72    System.out.println("Total Size: " + total);
73    System.out.println("Time taken: " + (end - start)/1.0e9);
74  }
75}
            以上代码多次运行后的结果为：
1Total Size: 276592101
2Time taken: 0.124100452

            比起单线程版本，速度基本提升了一倍的样子。到此，我们这个任务就完成了么？不，难道你不觉得上面这个设计，虽然说起来简单，但是实现起来却并不那么容易么？

            我们不仅需要设计一个用于保存计算任务的返回结果的不变的子类，而且还要花心思去设计如何不断的分派任务以及协调处理任务的返回结果。所以，即使我们这个设计提高了性能，却引入了相当的复杂性。

            


             所以，我们下面将会引入CountDownLatch辅助实现线程的设计和协作。上面的设计中，我们引入了Future，我们通过它获取任务的执行结果，同时，Future也隐含的帮我们完成了一些任务/线程之间的协调工作，我们因此不用考虑线程切换和并行中可能发现的一些问题。但是，大家知道，Future是需要返回值的，假使我们的任务没有返回值的话，Future就不太适合出现在我们的代码中，并且作为线程间的协作工具了。此时，我们会考虑到CountDownLatch作为替代。

            下面，我们将使用到CountDownLatch去做协调，在扫描到一个文件时，线程不在能够通过Future去接受一个计算的返回结果，而是去更新一个AtomicLong类型的共享变量totalSize。AtomicLong提供了更新并取回一个简单long类型的变量的线程安全的方法。此外，我们还会有个AtomicLong类型的变量pendingFileVisits，用于保存当前等待访问的文件或子目录的数量。而当该值为0时，我们就通过countDown()来释放线程闩。

01public class ConcurrentTotalFileSizeWLatch {  
02  private ExecutorService service;
03   //用于计算待扫描的文件/子文件夹的个数
04  final private AtomicLong pendingFileVisits = new AtomicLong();
05  final private AtomicLong totalSize = new AtomicLong();
06    //定义一个锁存器，给定计数初始化的 CountDownLatch的个数是1个
07  final private CountDownLatch latch = new CountDownLatch(1);
08  private void updateTotalSizeOfFilesInDir(final File file) {
09    long fileSize = 0;
10    if (file.isFile())
11      fileSize = file.length();
12    else {      
13      final File[] children = file.listFiles();      
14      if (children != null) {
15        for(final File child : children) {
16          if (child.isFile()) 
17            fileSize += child.length();
18          else {
19            pendingFileVisits.incrementAndGet();
20            service.execute(new Runnable() {
21              public void run() { updateTotalSizeOfFilesInDir(child); }
22            });            
23          }
24        }
25      }
26    }
27    totalSize.addAndGet(fileSize);
28    //递减锁存器的计数，如果计数到达零，则释放所有等待的线程。
29    if(pendingFileVisits.decrementAndGet() == 0) latch.countDown();
30  }
31 
32  private long getTotalSizeOfFile(final String fileName) 
33    throws InterruptedException {
34    service  = Executors.newFixedThreadPool(100);
35    pendingFileVisits.incrementAndGet();
36    try {
37     updateTotalSizeOfFilesInDir(new File(fileName));
38        // 使当前线程在锁存器倒计数至零之前一直等待，除非线程被中断或超出了指定的等待时间。
39     latch.await(100, TimeUnit.SECONDS);
40     return totalSize.longValue();
41    } finally {
42      service.shutdown();
43    }
44  }
45  public static void main(final String[] args) throws InterruptedException {
46    final long start = System.nanoTime();
47    final long total = new ConcurrentTotalFileSizeWLatch()
48      .getTotalSizeOfFile("D:/idea_ws");
49    final long end = System.nanoTime();
50    System.out.println("Total Size: " + total);
51    System.out.println("Time taken: " + (end - start)/1.0e9);
52  }
53}

        这个版本的代码比较上一个版本，简洁了很多，下面我们看看运行的结果：

1Total Size: 276592105
2Time taken: 0.114999062

        结果比较上一个版本，性能上差不多。然而，即使简洁，高性能，但是这个版本的代码存在着访问共享可变变量的风险。我们将在后面研究，如何二者兼得的方法。


         上面2个版本的代码中，我们通过Future和AtomicLong来实现数据的交换功能。当任务完成时能获取返回值时，Future就能派上用场，而AtomicLong的线程安全的原子操作对处理单个共享数据的值来说，是非常有用的。

           如果，只是想在两个线程间交换数据，我们可以用Exchanger类。它可以看做一个同步点，两个线程在改同步点上，可以线程安全的方式互换数据，如果两个线程的运行速度不一样，则运行较快的会被阻塞，直到慢的线程赶到同步点时，才能开始数据的互换。

            所以，如果想在线程间互发多组数据，则BlockingQueue接口可以派上用场。队列想必大家都比较熟悉了，队列没用空间，插入时会被阻塞；队列里没有可用数据，删除时会被阻塞。若想要插入和删除操作一一对应，可以使用SynchronousQueue类。该类的作用是本线程的每一个插入操作与其他线程相应的删除操作相匹配，以完成类似的手递手形式的数据传输。如果希望数据根据某种优先级在队列中上下浮动，则可以使用PriorityBlockingQueue。另外，如果只是要一个简单的阻塞队列，我们可以用链表的实现LinkedBlockingQueue或者数组的实现ArrayBlockingQueue。

             第3版的代码如下：

01public class ConcurrentTotalFileSizeWQueue {  
02  private ExecutorService service;
03  //定义队列，存放每个线程得到的fileSize
04  final private BlockingQueue<Long> fileSizes = 
05    new ArrayBlockingQueue<Long>(500);
06    //记录文件夹数
07  final AtomicLong pendingFileVisits = new AtomicLong();
08  private void startExploreDir(final File file) {
09    pendingFileVisits.incrementAndGet();
10    service.execute(new Runnable() {
11      public void run() { exploreDir(file); }
12    });          
13  }
14  private void exploreDir(final File file) {
15    long fileSize = 0;
16    if (file.isFile())
17      fileSize = file.length();
18    else {
19      final File[] children = file.listFiles();
20      if (children != null)
21        for(final File child : children) {
22          if (child.isFile())
23            fileSize += child.length();
24          else {
25            startExploreDir(child);
26          }
27        }
28    }
29    try { 
30      fileSizes.put(fileSize); 
31    } catch(Exception ex) { throw new RuntimeException(ex); }
32    pendingFileVisits.decrementAndGet();          
33  }
34 
35  private long getTotalSizeOfFile(final String fileName) 
36    throws InterruptedException {
37    service  = Executors.newFixedThreadPool(100);
38    try {
39      startExploreDir(new File(fileName));
40      long totalSize = 0;
41      while(pendingFileVisits.get() > 0 || fileSizes.size() > 0)
42      {
43        final Long size = fileSizes.poll(10, TimeUnit.SECONDS);
44        totalSize += size;
45      }
46      return totalSize;
47    } finally {
48      service.shutdown();
49    }
50  }   
51  public static void main(final String[] args) throws InterruptedException {
52    final long start = System.nanoTime();
53    final long total = new ConcurrentTotalFileSizeWQueue()
54      .getTotalSizeOfFile("D:/idea_ws");
55    final long end = System.nanoTime();
56    System.out.println("Total Size: " + total);
57    System.out.println("Time taken: " + (end - start)/1.0e9);
58  }
59}

            这个版本的运行结果如下：

1Total Size: 276591906
2Time taken: 0.124069344
            这一版的性能与之前一版相仿，但是代码简化方面又有提升了，主要归功与阻塞队列帮我们完成了线程之间的数据交换和同步的操作。由于oschina博客的字数限制，之后一篇文章，将利用java7引入的新的api来进一步改进上面的方案


联系
我们