19

我有一段代码可以监视目录以添加文件。每当将新文件添加到目录中时，文件的内容都会被挑选并发布到 kafka 上，然后删除该文件。

这在我发出单个请求时有效，但只要我的代码接受来自 jMeter 的 5 或 10 个用户请求，内容就会成功发布在 kafka 上，但代码无法删除文件。我收到一条FileSystemException消息The process cannot access the file because it is being used by another process.。

我想有一些我看不到的并发问题。

public void monitor() throws IOException, InterruptedException {
    Path faxFolder = Paths.get(TEMP_FILE_LOCATION);
    WatchService watchService = FileSystems.getDefault().newWatchService();
    faxFolder.register(watchService, StandardWatchEventKinds.ENTRY_CREATE);
    boolean valid = true;
    do {
        WatchKey watchKey = watchService.take();
        for (WatchEvent<?> event : watchKey.pollEvents()) {
            if (StandardWatchEventKinds.ENTRY_CREATE.equals(event.kind())) {
                String fileName = event.context().toString();
                publishToKafka(new File(TEMP_FILE_LOCATION + fileName).toPath(), "topic");
            }
        }
        valid = watchKey.reset();
    } while (valid);
}

private void publishToKafka(Path path, String topic) {
    try (BufferedReader reader = Files.newBufferedReader(path)) {
        String input = null;
        while ((input = reader.readLine()) != null) {
            kafkaProducer.publishMessageOnTopic(input, topic);
        }
    } catch (IOException e) {
        LOG.error("Could not read buffered file to send message on kafka.", e);
    } finally {
        try {
            Files.deleteIfExists(path); // This is where I get the exception
        } catch (IOException e) {
            LOG.error("Problem in deleting the buffered file {}.", path.getFileName(), e);
        }
    }
}

异常日志：

java.nio.file.FileSystemException: D:\upload\notif-1479974962595.csv: The process cannot access the file because it is being used by another process.

    at sun.nio.fs.WindowsException.translateToIOException(Unknown Source)
    at sun.nio.fs.WindowsException.rethrowAsIOException(Unknown Source)
    at sun.nio.fs.WindowsException.rethrowAsIOException(Unknown Source)
    at sun.nio.fs.WindowsFileSystemProvider.implDelete(Unknown Source)
    at sun.nio.fs.AbstractFileSystemProvider.deleteIfExists(Unknown Source)
    at java.nio.file.Files.deleteIfExists(Unknown Source)
    at com.panasonic.mdw.core.utils.MonitorDirectory$FileContentPublisher.publishToKafka(MonitorDirectory.java:193)
    at com.panasonic.mdw.core.utils.MonitorDirectory$FileContentPublisher.sendData(MonitorDirectory.java:125)
    at com.panasonic.mdw.core.utils.MonitorDirectory$FileContentPublisher.run(MonitorDirectory.java:113)
    at java.util.concurrent.ThreadPoolExecutor.runWorker(Unknown Source)
    at java.util.concurrent.ThreadPoolExecutor$Worker.run(Unknown Source)
    at java.lang.Thread.run(Unknown Source)
4

4 回答 4

3

查看您的代码，似乎当一个文件被线程选择再次发布时，另一个线程正在选择它进行发布。这就是为什么没有人能够删除它的原因。它必须只是并发问题。您应该根据标准重新设计代码：可以同时运行的步骤和不能同时运行的步骤。所以整个过程的步骤是：

  1. 拿起一个文件（主线程应该这样做）
  2. 发布文件（调用其他线程来做）
  3. 删除文件（调用线程应该删除它）
  4. 检查是否存在任何文件（再次主线程可以做到）

同样，在选择文件的那一刻，您可以将其读入 buffer ，将其删除，然后继续发布。这将确保主线程不会将此文件分配给其他线程。

于 2016-11-24T09:48:48.763 回答
2

在 WatchService 事件中添加睡眠时间总是一个更好的主意：

if (StandardWatchEventKinds.ENTRY_CREATE.equals(event.kind())) {

       try {
            Thread.sleep(3000);
       } catch (InterruptedException e) {
            e.printStackTrace();
       }

// now do your intended jobs ...

我需要添加睡眠时间，否则它不适用于多个请求并用于获取错误：

The process cannot access the file because it is being used by another process
于 2020-04-27T23:57:25.003 回答
1

我在以下线程中也遇到了类似的问题： 尝试将动态创建的文件上传到队列服务时队列上的多线程，我需要 2 天才能解决。感谢Holger给出了上述答案，锁定的发生可能是由于另一个线程读取时尚未完全完成创建，它为我节省了很多时间。

我从互联网上找到了很多最初的解决方案是：

WatchEvent<Path> ev = cast(event);
Path name = ev.context();
Path child = dir.resolve(name);
//queueUploadFile(child);
if (kind == ENTRY_CREATE) {
    uploadToQueue(this.queueId, child);
}

我将其更改为：

WatchEvent<Path> ev = cast(event);
Path name = ev.context();
Path child = dir.resolve(name);
//queueUploadFile(child);
if (kind == ENTRY_MODIFY) {
    uploadToQueue(this.queueId, child);
}

一切都很完美。为了处理“以某种方式”触发的多个 ENTRY_MODIFY 事件（上传了重复的文件），我在上传后对 uploadToQueue() 方法中的文件执行删除。

我希望我基于上述贡献采取的方法也能帮助其他有类似问题的人。

于 2017-06-09T20:24:41.037 回答
-1

在删除该文件之前，您必须关闭所有正在访问该文件的连接。

于 2016-11-24T06:50:12.093 回答