流式传输文件并在读取后移动它们

Question

我想流式传输文件中包含的行，但在处理每个文件后将其移动到另一个文件夹。

目前的流程是这样的：

解释：

1. 我创建了Stream 的Files
2. 我为每个人创建一个BufferedReader
3. 我flatMap 到BufferedReader 的行Stream
4. 我打印每一行。

代码（为简单起见省略了例外）：

(1)    Stream.generate(localFileProvider::getNextFile)
(2)       .map(file -> new BufferedReader(new InputStreamReader(new FileInputStream(file))))
(3)       .flatMap(BufferedReader::lines)
(4)       .map(System.out::println)
          .MOVE_EACH_FILE_FROM_INPUT_FOLDER_TO_SOME_OTHER_FOLDER;

是否可以在每个文件被完全读取后移动它并继续处理流中的其他文件？

Answer 1

我只想创建两个方法：

public void processFile(File f);
public void moveFile(File f, File dstFolder);

然后在 lambda 中：

Stream.generate(localFileProvider::getNextFile).forEach(file->
   {
     processFile(file);
     moveFile(file, dstFolder);
   }
);

Answer 2

其实如果你能把逻辑分成不同的方法就很容易了

 public Path readFile(File eachFile) {
   BufferedReader reader = new BufferedReader(new InputStreamReader(new FileInputStream(file)));

  //try-with-resources
 try (Stream<String> lines = reader.lines()) {
lines.forEach(System.out::println);

 } 
catch (IOException e) {
   e.printStackTrace();
 }
   return eachFile.toPath();
  }

然后对每个文件调用这个方法

(1)    Stream.generate(localFileProvider::getNextFile)
(2)          .map(this::readFile) //process each file
(3)          .forEach(path->Files.move(path,Paths.get("new path"))); //then move each file

Answer 3

你可以chain a close action到一个流，如果flatMap会自动执行：

Stream.generate(localFileProvider::getNextFile).takeWhile(Objects::nonNull)

    .flatMap(file -> {
        try {
            Path p = file.toPath();
            return Files.lines(p, Charset.defaultCharset()).onClose(() -> {
                try { // move path/x/y/z to path/x/y/z.moved
                    Files.move(p, p.resolveSibling(p.getFileName()+".moved"));
                } catch(IOException ex) { throw new UncheckedIOException(ex); }
            });
        } catch(IOException ex) { throw new UncheckedIOException(ex); }
    })

    .forEach(System.out::println);

the documentation of onClose 声明很重要：

在流上调用close() 方法时运行关闭处理程序，并按照添加它们的顺序执行。

因此，移动关闭处理程序在已经存在的关闭处理程序之后执行，该处理程序将关闭用于读取行的文件句柄。

我使用Charset.defaultCharset() 来模仿问题代码的嵌套构造函数new InputStreamReader(new FileInputStream(file))) 的行为，但通常，您应该尽可能使用固定字符集，例如Files.lines 的默认UTF-8。

Answer 4

可以这样做：

    files
        .map( file -> {
            getBufferedReader( file ).lines()
                .forEach( System.out::println );
            return file;
        } )
        .forEach( this::moveFile );

已检查异常和Reader.close的更新：

诚然，这不会在 finally 块中运行 close()，所以这是一个缺点。本次更新的重点主要是说明在 Java 8 流中处理已检查异常的一种方式。

假设您有以下可用的实用程序代码：

private interface ThrowingFunction<I, O>
{
    O apply( I input ) throws Exception;
}

private <I, O> Function<I, O> unchecked( ThrowingFunction<I, O> checked )
{
    return i -> {
        try {
            return checked.apply( i );
        }
        catch ( Exception e ) {
            throw new RuntimeException();
        }
    };
}

private interface ThrowingConsumer<T>
{
    void consume( T input ) throws Exception;
}

private <T> Consumer<T> unchecked( ThrowingConsumer<T> checked )
{
    return t -> {
        try {
            checked.consume( t );
        }
        catch ( Exception e ) {
            throw new RuntimeException();
        }
    };
}

private BufferedReader getBufferedReader( File file ) throws FileNotFoundException
{
    return new BufferedReader( new InputStreamReader( new FileInputStream( file )));
}

编写实际的代码就变成了：

    files
        .map( file -> {
            Stream.of( file )
                .map( unchecked( this::getBufferedReader ))
                .map( reader -> {
                    reader.lines().forEach( System.out::println );
                    return reader;
                } )
                .forEach( unchecked( Reader::close ));
            return file;
        } )
        .forEach( this::moveFile );