有没有办法在 Java 中暂停和恢复 Files.walkFileTree？

Posted 2023-03-15

【中文标题】有没有办法在 Java 中暂停和恢复 Files.walkFileTree？

【英文标题】：Is there any way to pause and resume Files.walkFileTree in Java?

【发布时间】：2021-11-23 21:09:42

【问题描述】：

这个类遍历一个路径，并且对于它遇到的每个文件，它都会将一个线程抛出到一个 ThreadPoolExecutor 中，该线程将提供给它的对象收集到一个 HashMap 中。我有另一个线程来监控 HashMap，当该 Map 中有 5,000 个元素时，它会被转储到 mysql 数据库，然后记录的记录会从 HashMap 中清除，一切都会继续。

但是，一旦 walker 到达它已经达到超过 200 万个文件的位置，HashMap 的实际清除已经滞后到它几乎落后一百万条记录的位置，所以我希望能够暂停文件夹行走直到数据转储已经赶上，然后恢复...冲洗重复...

这门课开始后可以暂停吗？或者，有什么办法可以减慢速度吗？

public class WalkFilePaths implements Runnable

    public WalkFilePaths(Path rootPath, ThreadPoolExecutor executor) 
        this.rootPath           = rootPath;
        this.executor           = executor;
    

    private        final Path               rootPath;
    private static       ThreadPoolExecutor executor;
    private static final FileDataManager    fileDataManager = new FileDataManager();

    @Override public void run() 
        try 
            FolderWalker folderWalker = new FolderWalker();
            Files.walkFileTree(rootPath,folderWalker);
        
        catch (IOException e) e.printStackTrace();
    

    public static class FolderWalker extends SimpleFileVisitor<Path> 
        @Override
        public FileVisitResult preVisitDirectory(Path dir, BasicFileAttributes attrs) 
            return FileVisitResult.CONTINUE;
        

        @Override
        public FileVisitResult visitFile(Path path, BasicFileAttributes attrs) 
            if(attrs.isRegularFile()) 
                executor.execute(fileDataManager.addFileMap(new FileDataModel(path.toFile(), attrs.creationTime().toInstant(), attrs.lastAccessTime().toInstant())));
            
            return FileVisitResult.CONTINUE;
        

        @Override
        public FileVisitResult postVisitDirectory(Path dir, IOException e) 
            return FileVisitResult.CONTINUE;
        

        @Override
        public FileVisitResult visitFileFailed(Path file, IOException exc) 
            return FileVisitResult.CONTINUE;
        
    

【参考方案1】：

您可以通过使用计数Semaphore 来控制进程。

这背后的概念很简单：您从信号量上释放的 5_000 个许可的初始值开始；每次您的代码将一个新文件排入队列时，它都需要一个许可。当信号量用完许可时，您的代码将等待（阻止对semaphore.acquire() 的调用），直到再次释放许可。每次使用累积的数据时，您现有的代码都必须release() 允许。

上述概念的简单实现是：

class FileDataManager 

    private static final int BATCH_SIZE = 5_000;
    private final Semaphore semaphore = new Semaphore(BATCH_SIZE);
    private Map<String, String> data = new HashMap<>(BATCH_SIZE);

    Runnable addFileMap(FileDataModel fileDataModel) 
        try 
            //Try to acquire a permit, or wait (blocking call) until a permit is available
            semaphore.acquire();
            return new Runnable() 
                @Override
                public void run() 
                    //Process file...
                    data.put(fileDataModel.toString(), fileDataModel.toString());
                
            ;
         catch (InterruptedException ex) 
            Logger.getLogger(FileDataManager.class.getName()).log(Level.SEVERE, null, ex);
            throw new RuntimeException(ex);
        

    

    public int accumulatedFileCount() 
        return data.size();
    

    public void releasePermits() 
        semaphore.release(BATCH_SIZE);
    

监控地图的另一个线程会：

//...
            if (fileDataManager.accumulatedFileCount() >= 5_000) 
                // store data in RDBMS
                fileDataManager.releasePermits();
            
//...