我有一段代码监视目录以添加文件.每当将新文件添加到目录时,将挑选该文件的内容并在kafka上发布,然后删除该文件.

这在我发出单个请求时有效,但是一旦我将代码从jMeter请求5或10个用户请求,内容就会在kafka上成功发布,但代码无法删除该文件.我收到一个FileSystemException,其中包含一个消息,该进程无法访问该文件,因为它正由另一个进程使用.

我想有一些我无法看到的并发问题.

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)