java - 将 FileWatcher 与多线程结合使用

Question

我正在尝试将多线程与java中的FileWatcher服务集成。即，我不断地监听特定目录 -> 每当创建新文件时，我需要生成一个处理该文件的新线程(假设它打印文件内容)。我设法编写了一个可以编译并运行的代码(但没有达到预期)。它按顺序工作，意味着文件 2 在文件 1 之后处理，文件 3 在文件 2 之后处理。我希望它并行执行。

添加代码片段:

while(true) {
            WatchKey key;
            try {
                key = watcher.take();
                Path dir = keys.get(key);
                for (WatchEvent<?> event: key.pollEvents()) {
                    WatchEvent.Kind<?> kind = event.kind();
                    if (kind == StandardWatchEventKinds.OVERFLOW) {
                        continue;
                    }
                    if(kind == StandardWatchEventKinds.ENTRY_CREATE){
                        boolean valid = key.reset();
                        if (!valid) {
                            break;
                        }
                        log.info("New entry is created in the listening directory, Calling the FileProcessor");
                        WatchEvent<Path> ev = (WatchEvent<Path>)event;
                        Path newFileCreatedResolved = dir.resolve(ev.context());
                        try{
                        FileProcessor processFile = new FileProcessor(newFileCreatedResolved.getFileName().toString());
                        Future<String> result = executor.submit(processFile);
                            try {
                                System.out.println("Processed File" + result.get());
                            } catch (ExecutionException e) {
                                e.printStackTrace();
                            }
                        //executor.shutdown(); add logic to shut down
                        }   
                    }
                }
            }
        }

和 FileProcessor 类

public class FileProcessor implements Callable <String>{
    FileProcessor(String triggerFile) throws FileNotFoundException, IOException{
        this.triggerFile = triggerFile;
    }
    public String call() throws Exception{
        //logic to write to another file, this new file is specific to the input file
        //returns success
    }

现在发生了什么 -> 如果我一次传输 3 个文件，它们是按顺序传输的。首先将 file1 写入其目标文件，然后是 file2、file3 等。

我说得有道理吗？我需要更改哪一部分才能使其并行？或者 Executor 服务就是被设计成这样工作的。

Answer 1

调用Future.get()正在阻塞。当然，在处理完成之前结果才可用，并且您的代码在此之前不会提交另一个任务。

将您的Executor包装在 CompletionService 中和 submit()而是给它任务。让另一个线程使用 CompletionService 的结果来执行任务完成后所需的任何处理。

或者，您可以使用 CompletableFuture 的辅助方法设置等效的操作管道。

第三种更简单但可能不太灵活的选择是将后处理合并到任务本身中。我demonstrated一个简单的任务包装器来展示如何完成此操作。