Spring 集成nioLocker读取异常

Question

我有一个集成流程可以扫描文件进行处理。由于可能有多个处理器扫描同一目录，我添加了“.nioLocker()”以防止来自其他 JVM 的处理器处理该文件。

流程配置如下：

IntegrationFlows.from( // Scan files from input dir
                    s -> s.file(new File(fileInputDir))
                            .preventDuplicates(true)
                            .nioLocker()
                            .regexFilter("(.)*\.[xX][mM][lL]|(.)+\.[dD][nN][eE]"),        // to match any case of the letters XML
                    p -> p.poller(Pollers.fixedRate(filePollerInterval)
                            .taskExecutor(new ScheduledThreadPoolExecutor(filePoolSize))
                    )

现在，问题是即使使用一个处理器运行，当我调用 BufferedReader.readLine 时，我也会收到一个异常，指出文件已锁定

java.io.IOException: The process cannot access the file because another process has locked a portion of the file
    at java.io.FileInputStream.readBytes(Native Method)
    at java.io.FileInputStream.read(FileInputStream.java:255)
    at sun.nio.cs.StreamDecoder.readBytes(StreamDecoder.java:284)
    at sun.nio.cs.StreamDecoder.implRead(StreamDecoder.java:326)
    at sun.nio.cs.StreamDecoder.read(StreamDecoder.java:178)
    at java.io.InputStreamReader.read(InputStreamReader.java:184)
    at java.io.BufferedReader.fill(BufferedReader.java:161)
    at java.io.BufferedReader.readLine(BufferedReader.java:324)
    at java.io.BufferedReader.readLine(BufferedReader.java:389)

我试图通过调用

来释放锁

 private NioFileLocker fileLocker = new NioFileLocker();
 fileLocker.unlock(file);

但这不起作用！（我怀疑是因为它是从与储物柜不同的线程调用的，但我不确定）

正确的获取锁的方法是什么？有没有更好的方法来确保只有一个处理器获得对资源的访问权？

------------------------编辑---------------- ------------------

所以我采取了额外的步骤来确保锁定文件的线程与从其文件通道读取的线程相同。为此，我使用了直接频道。（以前，传递给 fileSplitter 的消息是通过 QueueChannel 执行的，它会在不同的线程上执行 send() ）。我仍然收到错误

2017-07-21 11:22:03.316  INFO 336488 --- [           main] c.f.e.m.i.MailerInboundApplication       : Started MailerInboundApplication in 13.541 seconds (JVM running for 14.419)
2017-07-21 11:22:09.946  INFO 336488 --- [ask-scheduler-5] o.s.i.file.FileReadingMessageSource      : Created message: [GenericMessage [payload=input\EMAIL92770.9352177.20170617.xml, headers={id=5dba6d62-b0a5-508e-48a9-cfddfa3b331f, timestamp=1500654129946}]]

2017-07-21 11:22:09.962 DEBUG 336488 --- [ask-scheduler-5] c.f.edd.mailer.inbound.core.FileRouter   : fileRouter received message: GenericMessage [payload=input\EMAIL92770.9352177.20170617.xml, headers={CORRELATION_ID=92770.9352177.20170617, id=32a8846d-5425-0b
ee-657e-8767e1fb6105, timestamp=1500654129962}]
2017-07-21 11:22:09.962 DEBUG 336488 --- [ask-scheduler-5] c.f.e.mailer.inbound.core.FileSplitter   : fileSplitter received message: GenericMessage [payload=input\EMAIL92770.9352177.20170617.xml, headers={CORRELATION_ID=92770.9352177.20170617, id=32a8846d-5425-
0bee-657e-8767e1fb6105, timestamp=1500654129962}]

java.io.IOException: The process cannot access the file because another process has locked a portion of the file
        at sun.nio.ch.FileDispatcherImpl.read0(Native Method)
        at sun.nio.ch.FileDispatcherImpl.read(FileDispatcherImpl.java:61)
        at sun.nio.ch.IOUtil.readIntoNativeBuffer(IOUtil.java:223)
        at sun.nio.ch.IOUtil.read(IOUtil.java:197)
        at sun.nio.ch.FileChannelImpl.read(FileChannelImpl.java:159)
        at com.fiserv.edd.mailer.inbound.core.FileSplitter.splitMessage(FileSplitter.java:93)

FileSplitter.java处的代码：

    @Override
protected Object splitMessage(Message<?> message) {
    String correlationId = (String) message.getHeaders().get("CORRELATION_ID"); //Save the correlation ID so we can use it to send the DNE/CLP file later
    File file = (File) message.getPayload();
    String inputFileName = file.getName();


    log.info(LogEvent.getBuilder().withMessageId(inputFileName)
            .withMessage("Processing file: " + inputFileName).build());

    long startTime = System.currentTimeMillis();
    Optional<InputHeader> inputHeader = Optional.empty();// headerParser.parse(file);

    ParsingReport pr = new ParsingReport(inputFileName);
    try (RandomAccessFile lfs = new RandomAccessFile(file.getAbsolutePath(), "rw")){
        FileChannel fc = lfs.getChannel();
        byte[] bytes = new byte[1024];
        fc.read(ByteBuffer.wrap(bytes));
        System.out.println(new String(bytes));

    } catch (FileNotFoundException e) {
            e.printStackTrace();
    } catch (IOException e) {
            e.printStackTrace();
    }
    return Collections.EMPTY_LIST;
}

Answer 1

当我们使用 java.nio.channels.FileLock 时，我们只能通过与该锁关联的 FileChannel 或 InputStream 访问文件内容：

FileInputStream in = new FileInputStream(file);
try {
    java.nio.channels.FileLock lock = in.getChannel().lock();
    try {
        Reader reader = new InputStreamReader(in, charset);
        ...
    } finally {
        lock.release();
    }
} finally {
    in.close();
}

NioFileLocker 不允许轻易访问 FileLock。所以，你应该在你的代码中使用这样的东西：

 new DirectFieldAccessor(this.nioFileLocker).getPropertyValue("lockCac‌he");

并将其转换为 Map<File, FileLock> 以获得为文件创建的 FileLock。

同时，请就此事提出 JIRA。这个 NioFileLocker 会导致很多问题。应该以某种方式进行修改。谢谢

Spring 集成nioLocker读取异常

Spring Integration nioLocker Exception when reading

java

synchronization

spring-integration

file-locking