Java 8 流通过前缀连接日志文件

Question

我有一个滚动日志文件列表，比如说:

thread1.2018-04-09.log
thread1.2018-04-10.log
thread1.2018-04-11.log
thread2.2018-04-09.log
thread2.2018-04-10.log
thread2.2018-04-11.log

我想连接每个线程的所有日志文件来处理这些文件，就好像每个线程只有一个文件一样。

我从单独解析所有文件开始:

Files.newDirectoryStream(Paths.get("path/to/log/folder"),
                    path -> path.toString().endsWith(".log"))
                    .forEach(this::parseLog);

然后通过手动检查生成文件的线程来合并输出。不是最优的...

是否可以直接在流操作中拼接具有相同前缀的文件流？

编辑:

根据评论中的建议，这是我想出的:

public class Test {

    public static void main(String[] args) {
        new Test().readLogs();
    }

    public void readLogs() {
        try (Stream<Path> stream = Files.list(Paths.get("."))
                .filter(path -> path.toString().endsWith(".log"))) {

            Map<String, List<Path>> pathsByThread = stream.collect(Collectors.groupingBy(this::getThreadName));

            for (String threadName : pathsByThread.keySet()) {
                pathsByThread.get(threadName).stream().flatMap(this::readAllLines).forEach(this::parseLogLine);
            }

        } catch (IOException e) {
            e.printStackTrace();
        }
    }

    private String getThreadName(Path path) {
        int index = path.getFileName().toString().indexOf(".");

        return path.getFileName().toString().substring(0, index);
    }

    private Stream<String> readAllLines(Path path) {
        try (Stream<String> fileContent = Files.lines(path)) {
            return fileContent;
        } catch (IOException e) {
            e.printStackTrace();
        }
        return null;
    }

    private void parseLogLine(String line) {
        // Do something smart
        System.out.println(line);
    }
}

我有一个运行时错误:

Exception in thread "main" java.lang.IllegalStateException: stream has already been operated upon or closed

Answer 1

问题是你的方法

private Stream<String> readAllLines(Path path) {
    try (Stream<String> fileContent = Files.lines(path)) {
        return fileContent;
    } catch (IOException e) {
        e.printStackTrace();
    }
    return null;
}

try( … ) 构造的真正目的是在离开 block 时关闭资源。您正在返回一个关闭的流。

在这个地方，你应该考虑 documentation of flatMap :

Each mapped stream is closed after its contents have been placed into this stream.

因此，您无需关闭 flatMap 函数中返回的流，Stream 实现即可。

所以在这种情况下，使用

private Stream<String> readAllLines(Path path) {
    try {
        return Files.lines(path);
    }
    catch(IOException e) {
        throw new UncheckedIOException(e);
    }
}

顺便说一句，您正在遍历 Map 的 keySet() 来为每个执行 get 查找，只是为了处理只有值(value)观。考虑到您可以遍历 values()，这很低效首先，如果这是你唯一感兴趣的事情。(你可以使用 entrySet() 如果你需要两者)你甚至可以在这个地方使用流，将整个操作简化为:

public void readLogs() {
    try(Stream<Path> stream = Files.list(Paths.get("."))) {
        stream.filter(path -> path.toString().endsWith(".log"))
              .collect(Collectors.groupingBy(path -> {
                  final String s = path.getFileName().toString();
                  return s.substring(0, s.indexOf("."));
              }))
              .values().stream()
              .flatMap(List::stream)
              .flatMap(path -> {
                  try { return Files.lines(path); }
                  catch (IOException e) { throw new UncheckedIOException(e); }
              })
              .forEachOrdered(this::parseLogLine);
    } catch(IOException|UncheckedIOException e) {
        e.printStackTrace();
    }
}

请注意，由于您仅使用分组来确定顺序，因此您也可以使用 sort 代替:

public void readLogs() {
    try(Stream<Path> stream = Files.list(Paths.get("."))) {
        stream.filter(path -> path.toString().endsWith(".log"))
              .sorted(Comparator.comparing(path -> {
                  final String s = path.getFileName().toString();
                  return s.substring(0, s.indexOf("."));
              }))
              .flatMap(path -> {
                  try { return Files.lines(path); }
                  catch (IOException e) { throw new UncheckedIOException(e); }
              })
              .forEachOrdered(this::parseLogLine);
    } catch(IOException|UncheckedIOException e) {
        e.printStackTrace();
    }
}

您可以通过按整个文件名排序来简化此操作，因为这意味着按它们的公共(public)前缀排序，包括第一个点之前的部分:

public void readLogs() {
    try(Stream<Path> stream = Files.list(Paths.get("."))) {
        stream.filter(path -> path.toString().endsWith(".log"))
              .sorted(Comparator.comparing(path -> path.getFileName().toString()))
              .flatMap(path -> {
                  try { return Files.lines(path); }
                  catch (IOException e) { throw new UncheckedIOException(e); }
              })
              .forEachOrdered(this::parseLogLine);
    } catch(IOException|UncheckedIOException e) {
        e.printStackTrace();
    }
}