java - procesamiento - Cierre de arroyos en medio de tuberías.

stream java 8 ejemplo

Question

Cuando ejecuto este código que abre una gran cantidad de archivos durante una canalización de flujo:

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);
}

Me sale una excepción:

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

El problema es que Stream.count no cierra el flujo cuando Stream.count atravesarlo. Pero no veo por qué no debería, dado que se trata de una operación de terminal. Lo mismo se aplica a otras operaciones de terminal, tales como reduce y forEach . flatMap por otro lado, cierra los arroyos en los que se compone.

La documentación me dice que use una declaración try-with-resouces para cerrar secuencias si es necesario. En mi caso podría reemplazar la línea de count con algo como esto:

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

Pero eso es ruidoso y feo y podría ser un verdadero inconveniente en algunos casos con tuberías grandes y complejas.

Así que mis preguntas son las siguientes:

1. ¿Por qué no se diseñaron los flujos para que las operaciones de la terminal cierren los flujos en los que están trabajando? Eso los haría trabajar mejor con las corrientes de IO.
2. ¿Cuál es la mejor solución para cerrar flujos de IO en tuberías?

runtimizeException es un método que envuelve la excepción comprobada en RuntimeException s.

Answer 1

Aquí hay una alternativa que utiliza otro método de Files y evitará la pérdida de descriptores de archivos:

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);

A diferencia de su versión, devolverá un int lugar de un long para el recuento de líneas; Pero no tienes archivos con tantas líneas, ¿verdad?

Answer 2

El cierre de la interfaz AutoCloseable solo se debe llamar una vez. Consulte la documentación de AutoCloseable para más información.

Si las operaciones finales cerrarían el flujo automáticamente, el cierre podría invocarse dos veces. Echa un vistazo al siguiente ejemplo:

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

Tal como se define en este momento, el método de cierre de líneas se invocará exactamente una vez. Independientemente de si la operación final se completa normalmente o la operación se cancela con IOException . Si, en cambio, la secuencia se cerrara implícitamente en la operación final , el método de cierre se llamará una vez, si se produce una excepción IOException , y dos veces si la operación se completa con éxito.

Answer 3

Es posible crear un método de utilidad que cierre de forma confiable los flujos en medio de una tubería.

Esto asegura que cada recurso se cierre con una declaración try-with-resource pero evita la necesidad de un método de utilidad personalizado, y es mucho menos detallado que escribir la declaración try directamente en la lambda.

Con este método, la tubería de la pregunta se ve así:

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);

La implementación se ve así:

/**
 * 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);
    }
}

^{(Dado que los recursos que deben cerrarse a menudo también generan excepciones cuando se asignan, las excepciones que no son de tiempo de ejecución están envueltas en excepciones de tiempo de ejecución, evitando la necesidad de un método separado que haga eso).}

Answer 4

Hay dos problemas aquí: el manejo de las excepciones marcadas, como IOException , y el cierre oportuno de los recursos.

Ninguna de las interfaces funcionales predefinidas declara excepciones marcadas, lo que significa que deben manejarse dentro de la lambda, o envolverse en una excepción no verificada y volver a ser eliminada. Parece que tu función runtimizeException hace eso. Probablemente también tuvo que declarar su propia interfaz funcional para ello. Como probablemente has descubierto, esto es un dolor.

En el cierre de recursos como archivos, se investigó si las secuencias se cierran automáticamente cuando se llega al final de la secuencia. Esto sería conveniente, pero no se trata de cerrar cuando se lanza una excepción. No hay un mecanismo mágico de hacer lo correcto para esto en las corrientes.

Nos quedamos con las técnicas estándar de Java para lidiar con el cierre de recursos, a saber, la construcción try-with-resources introducida en Java 7. TWR realmente quiere que los recursos se cierren al mismo nivel en la pila de llamadas cuando se abrieron. Se aplica el principio de "quien lo abre debe cerrarlo". TWR también se ocupa del manejo de excepciones, lo que generalmente hace que sea conveniente tratar el manejo de excepciones y el cierre de recursos en el mismo lugar.

En este ejemplo, el flujo es algo inusual, ya que asigna un Stream<Path> a un Stream<Stream<String>> . Estas corrientes anidadas son las que no están cerradas, lo que resulta en la eventual excepción cuando el sistema se queda sin descriptores de archivos abiertos. Lo que dificulta esto es que los archivos se abren mediante una operación de transmisión y luego se pasan hacia abajo; esto hace que sea imposible usar TWR.

Un enfoque alternativo para estructurar esta tubería es el siguiente.

La llamada a Files.lines es la que abre el archivo, por lo que este debe ser el recurso en la declaración TWR. El procesamiento de este archivo es donde (algunas) IOExceptions se producen, por lo que podemos hacer el ajuste de excepción en la misma declaración TWR. Esto sugiere tener una función simple que asigna la ruta a un recuento de líneas, mientras se maneja el cierre de recursos y el ajuste de excepciones:

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);
    }
}

Una vez que tenga esta función auxiliar, la línea principal se ve así:

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

Answer 5

Necesitará llamar a close() en esta operación de transmisión, lo que hará que se llame a todos los controladores de cierre subyacentes.

Mejor aún, sería envolver su declaración completa en un bloque try-with-resources , ya que entonces llamará automáticamente al controlador de cierre.

Esto puede no ser posible en su situación, esto significa que tendrá que manejarlo usted mismo en alguna operación. Es posible que sus métodos actuales no sean adecuados para transmisiones en absoluto.

Parece que realmente necesitas hacerlo en tu segunda operación map() .