Schlussströme in der Mitte der Rohrleitungen

Question

Wenn ich diesen Code ausführen, die eine Menge von Dateien während einer Stream-Pipeline eröffnet:

public static void main(String[] args) throws IOException { 
    Files.find(Paths.get("JAVA_DOCS_DIR/docs/api/"), 
      100, (path, attr) -> path.toString().endsWith(".html")) 
     .map(file -> runtimizeException(() -> Files.lines(file, StandardCharsets.ISO_8859_1))) 
     .map(Stream::count) 
     .forEachOrdered(System.out::println); 
} 

Ich erhalte eine Ausnahme:

java.nio.file.FileSystemException: /long/file/name: Too many open files 

Das Problem ist, dass Stream.count den Stream nicht schließt, wenn es durchquert wird. Aber ich sehe nicht, warum es nicht sollte, da es eine Terminaloperation ist. Dasselbe gilt für andere Terminaloperationen wie reduce und forEach. flatMap schließt dagegen die Ströme, aus denen es besteht.

Die Dokumentation sagt mir, eine try-with-resouces-Anweisung zu verwenden, um Streams bei Bedarf zu schließen.

.map(s -> { long c = s.count(); s.close(); return c; }) 

Aber das ist laut und hässlich und könnte eine echte Unannehmlichkeit in einigen Fällen mit großen, komplexen Pipelines sein: In meinem Fall die count Linie mit so etwas wie diesem, ich könnte ersetzen.

Also folgend meine Fragen sind:

1. Warum die Ströme wurden nicht so konzipiert, dass Operationen, die Ströme in der Nähe auf sie arbeiten? Dadurch würden sie besser mit IO-Streams arbeiten.
2. Was ist die beste Lösung zum Schließen von IO-Streams in Pipelines?

---

runtimizeException ist eine Methode, die Ausnahme in RuntimeException s geprüft einwickelt.

Answer 1

Hier gibt es zwei Probleme: die Behandlung von geprüften Ausnahmen wie IOException und das rechtzeitige Schließen von Ressourcen.

Keine der vordefinierten funktionalen Schnittstellen deklariert irgendwelche checked Ausnahmen, was bedeutet, dass sie innerhalb des Lambdas behandelt oder in eine ungeprüfte Exception gehüllt werden müssen. Es sieht so aus, als ob Ihre runtimizeException Funktion das tut. Wahrscheinlich mussten Sie dafür auch eine eigene funktionale Schnittstelle deklarieren. Wie Sie wahrscheinlich festgestellt haben, ist dies ein Schmerz.

Beim Schließen von Ressourcen wie Dateien wurde untersucht, ob Streams automatisch geschlossen werden, wenn das Ende des Streams erreicht wurde. Das wäre praktisch, aber es geht nicht um das Schließen, wenn eine Ausnahme ausgelöst wird. In Streams gibt es dafür keinen magischen "Do-the-Right-Thing" -Mechanismus.

Wir sind mit den Standard-Java-Techniken des Umgangs mit Ressourcen schließen, nämlich die Try-mit-Ressourcen Konstrukt in Java 7 eingeführt. TWR möchte Ressourcen auf der gleichen Ebene in der Aufrufliste geschlossen werden als sie geöffnet wurden. Das Prinzip "wer es öffnet, muss es schließen" gilt. TWR befasst sich auch mit der Ausnahmebehandlung, die es normalerweise erleichtert, mit der Ausnahmebehandlung und dem Schließen von Ressourcen an derselben Stelle umzugehen.

In diesem Beispiel ist der Stream insofern ungewöhnlich, als er eine Stream<Path> auf eine Stream<Stream<String>> abbildet. Diese verschachtelten Datenströme sind diejenigen, die nicht geschlossen sind, was zu einer eventuellen Ausnahme führt, wenn das System keine offenen Dateideskriptoren mehr hat.Was dies schwierig macht, ist, dass Dateien durch eine Stream-Operation geöffnet und dann stromabwärts weitergeleitet werden; Dies macht es unmöglich, TWR zu verwenden.

Ein alternativer Ansatz zur Strukturierung dieser Pipeline ist wie folgt.

Der Aufruf Files.lines ist derjenige, der die Datei öffnet, also muss dies die Ressource in der TWR-Anweisung sein. Die Verarbeitung dieser Datei ist, wo (einige) IOExceptions geworfen werden, so dass wir die Ausnahme Wrapping in der gleichen TWR-Anweisung tun können. Dies deutet auf eine einfache Funktion, die den Weg zu einer Zeilenzahl abbildet, während Schließen Ressourcen Handhabung und Ausnahme Verpackung:

long lineCount(Path path) { 
    try (Stream<String> s = Files.lines(path, StandardCharsets.ISO_8859_1)) { 
     return s.count(); 
    } catch (IOException ioe) { 
     throw new UncheckedIOException(ioe); 
    } 
} 

Sobald Sie diese Hilfsfunktion haben, sieht die Hauptleitung wie folgt aus:

Files.find(Paths.get("JAVA_DOCS_DIR/docs/api/"), 
      100, (path, attr) -> path.toString().endsWith(".html")) 
    .mapToLong(this::lineCount) 
    .forEachOrdered(System.out::println); 

Answer 2

Sie müssen in dieser Stream-Operation close() aufrufen, was dazu führt, dass alle zugrunde liegenden close-Handler aufgerufen werden.

Besser noch, wäre es, Ihre ganze Aussage in einen try-with-resources Block zu wickeln, da dann automatisch der Close-Handler aufgerufen wird.

Dies ist möglicherweise keine Möglichkeit in Ihrer Situation, das bedeutet, dass Sie es in einem Vorgang selbst behandeln müssen. Ihre aktuellen Methoden sind möglicherweise nicht für Streams geeignet.

Es scheint, als ob Sie es tatsächlich in Ihrem zweiten map() Betrieb tun müssen.

Answer 3

Die schließen der Schnittstelle AutoCloseable sollte nur einmal aufgerufen werden. Weitere Informationen finden Sie in der Dokumentation AutoCloseable.

Wenn final Operationen automatisch den Stream schließen würde, könnte schließen zweimal aufgerufen werden. Werfen Sie einen Blick auf das folgende Beispiel:

try (Stream<String> lines = Files.lines(path)) { 
    lines.count(); 
} 

Wie es jetzt definiert ist, die schließen Methode auf Linien genau einmal aufgerufen werden.Unabhängig davon, ob die endgültige Operation normal abgeschlossen wird, oder der Vorgang abgebrochen wird mit in IOException. Wenn der Strom stattdessen implizit in dem Betrieb endgültig geschlossen würde, die schließt Methode wäre einmal aufgerufen werden, wenn ein IOException auftritt, und zweimal, wenn der Vorgang erfolgreich abgeschlossen wird.

Answer 4

ist hier eine Alternative, die eine andere Methode von Files verwendet und vermeidet Filedeskriptoren undichte:

Files.find(Paths.get("JAVA_DOCS_DIR/docs/api/"), 
    100, (path, attr) -> path.toString().endsWith(".html")) 
    .map(file -> runtimizeException(() -> Files.readAllLines(file, StandardCharsets.ISO_8859_1).size()) 
    .forEachOrdered(System.out::println); 

Im Gegensatz zu Ihrer Version, wird es eine int anstelle eines long für die Linie zählen zurückzukehren; aber du hast keine Dateien mit so vielen Zeilen, oder?

Answer 5

Es ist möglich, eine Dienstprogrammmethode zu erstellen, die Streams in der Mitte einer Pipeline zuverlässig schließt.

Dies stellt sicher, dass jede Ressource mit einer try-with-resource-Anweisung geschlossen wird, vermeidet jedoch die Notwendigkeit einer benutzerdefinierten Hilfsmethode und ist viel weniger ausführlich als das Schreiben der try-Anweisung direkt in Lambda.

Mit dieser Methode wird die Pipeline von der Frage wie folgt aussieht:

Files.find(Paths.get("Java_8_API_docs/docs/api"), 100, 
     (path, attr) -> path.toString().endsWith(".html")) 
    .map(file -> applyAndClose(
     () -> Files.lines(file, StandardCharsets.ISO_8859_1), 
     Stream::count)) 
    .forEachOrdered(System.out::println); 

Die Implementierung sieht wie folgt aus:

/** 
* Applies a function to a resource and closes it afterwards. 
* @param sup Supplier of the resource that should be closed 
* @param op operation that should be performed on the resource before it is closed 
* @return The result of calling op.apply on the resource 
*/ 
private static <A extends AutoCloseable, B> B applyAndClose(Callable<A> sup, Function<A, B> op) { 
    try (A res = sup.call()) { 
     return op.apply(res); 
    } catch (RuntimeException exc) { 
     throw exc; 
    } catch (Exception exc) { 
     throw new RuntimeException("Wrapped in applyAndClose", exc); 
    } 
} 

^{(Da Ressourcen, die oft auch Ausnahmen auslösen, wenn sie geschlossen werden müssen, Sie werden zugewiesen, Nicht-Laufzeit-Ausnahmen werden in Laufzeit-Ausnahmen eingepackt, wodurch die Notwendigkeit für eine separate Methode vermieden wird, die dies tut.}