Come chiudere correttamente una quantità variabile di flussi?

Question

Sto creando più flussi a cui devo accedere in parallelo (o eventualmente parallelo). So come provare le risorse quando la quantità di risorse è fissata in fase di compilazione, ma cosa succede se la quantità di risorse è determinata da un parametro?

ho qualcosa di simile a questo:

private static void foo(String path, String... files) throws IOException { 
    @SuppressWarnings("unchecked") 
    Stream<String>[] streams = new Stream[files.length]; 

    try { 
     for (int i = 0; i < files.length; i++) { 
      final String file = files[i]; 
      streams[i] = Files.lines(Paths.get(path, file)) 
       .onClose(() -> System.out.println("Closed " + file)); 
     } 

     // do something with streams 
     Stream.of(streams) 
      .parallel() 
      .flatMap(x -> x) 
      .distinct() 
      .sorted() 
      .limit(10) 
      .forEach(System.out::println); 
    } 
    finally { 
     for (Stream<String> s : streams) { 
      if (s != null) { 
       s.close(); 
      } 
     } 
    } 
} 

Answer 1

si potrebbe scrivere un composito AutoCloseable per la gestione di una quantità dinamica di AutoCloseable:

import java.util.ArrayList; 
import java.util.List; 

public class CompositeAutoclosable<T extends AutoCloseable> implements AutoCloseable { 
    private final List<T> components= new ArrayList<>(); 

    public void addComponent(T component) { components.add(component); } 

    public List<T> getComponents() { return components; } 

    @Override 
    public void close() throws Exception { 
     Exception e = null; 
     for (T component : components) { 
      try { component.close(); } 
      catch (Exception closeException) { 
       if (e == null) { e = closeException; } 
       else { e.addSuppressed(closeException); } 
      } 
     } 
     if (e != null) { throw e; } 
    } 
} 

e si potrebbe usare nel metodo:

private static void foo(String path, String... files) throws Exception { 
    try (CompositeAutoclosable<Stream<String>> streams 
      = new CompositeAutoclosable<Stream<String>>()) { 
     for (int i = 0; i < files.length; i++) { 
      final String file = files[i]; 
      streams.addComponent(Files.lines(Paths.get(path, file)) 
       .onClose(() -> System.out.println("Closed " + file))); 
     } 
     streams.getComponents().stream() 
      .parallel() 
      .flatMap(x -> x) 
      .distinct() 
      .sorted() 
      .limit(10) 
      .forEach(System.out::println); 
    } 
} 

Answer 2

Il documentation of Stream.flatMap dice:

Ogni flusso mappato viene chiuso dopo che il suo contenuto è stato inserito in questo flusso.

In altre parole, per la chiusura ordinaria dei flussi, non è necessaria alcuna azione aggiuntiva. Tuttavia, dal momento che i flussi solo processati sono chiusi, non si deve creare i flussi con entusiasmo senza sapere se sono successivamente elaborati dal torrente:

private static void foo(String path, String... files) throws IOException { 
    Arrays.stream(files).flatMap(file-> { 
       try { return Files.lines(Paths.get(path, file)) 
        .onClose(() -> System.out.println("Closed " + file)); } 
       catch(IOException ex) { throw new UncheckedIOException(ex); } }) 
      .parallel() 
      .distinct() 
      .sorted() 
      .limit(10) 
      .forEachOrdered(System.out::println); 
} 

Con la creazione di sotto-correnti all'interno flatMap, è garantito che ciascuno è solo creato se lo stream sta per elaborarlo. Pertanto, questa soluzione chiuderà tutti i flussi secondari anche senza avere l'esterno Stream all'interno di un'istruzione try-with-resource. Lo svantaggio di questa soluzione è che non eseguirà la pulizia come l'istruzione try-with-resource nel caso eccezionale e persino il flusso esterno in un'istruzione try-with-resource non lo risolverà.

Considero questa eccezione mancante-sicurezza un difetto nell'implementazione Stream che dovrebbe essere corretta.

Provare a risolvere questo problema nel codice per ottenere un comportamento di chiusura sicuro, che si comporta come un'istruzione try-with-resource, ovvero non garantisce solo la chiusura di tutte le risorse, ma anche l'assenza di eccezioni generate da invocazioni close , è complicato. Essa implica ripercorrere ciò Files.lines(…) fa aggiungendo internamente e poi i pezzi mancanti:

static Closeable join(Closeable a, Closeable b) { 
    return a==null? b:()->{ try(Closeable c=a) { b.close(); }}; 
} 
private static void foo(String path, String... files) throws IOException { 
    Closeable[] cl={ null }; 
    try(Closeable c=()->{ if(cl[0]!=null) cl[0].close(); }) { 
     Stream.Builder<Stream<String>> b=Stream.builder(); 
     for(String file: files) { 
      BufferedReader br=Files.newBufferedReader(Paths.get(path, file)); 
      cl[0]=join(cl[0], br); 
      b.add(br.lines()); 
     } 
     b.build() 
     .parallel() 
     .flatMap(Function.identity()) 
     .distinct() 
     .sorted() 
     .limit(10) 
     .forEachOrdered(System.out::println); 
    } 
} 

questo modo si chiude tutte le risorse, indipendentemente dal fatto che e dove eccezioni accaduto, durante la lavorazione flusso, la chiusura o l'inizializzazione (si chiuderà tutto risorsa che sono stati assegnati finora).

Se si desidera verificare il comportamento, è possibile utilizzare il seguente codice:

private static void foo2(String path, String... files) throws IOException { 
    Closeable[] cl={ null }; 
    try(Closeable c=()->{ if(cl[0]!=null) cl[0].close(); }) { 
     Stream.Builder<Stream<String>> b=Stream.builder(); 
     for(String file: files) { 
      if(Math.random()>0.7) throw new RuntimeException("simulated opening failure"); 
      BufferedReader br=Files.newBufferedReader(Paths.get(path, file)); 
      System.out.println("opened "+file); 
      br=new BufferedReader(br) { 
       @Override 
       public void close() throws IOException { 
        System.out.println("closing "+file); 
        super.close(); 
        throw new RuntimeException("simulated closing "+file+" failure"); 
       } 
      }; 
      cl[0]=join(cl[0], br); 
      b.add(br.lines()); 
     } 
     b.build() 
     .parallel() 
     .flatMap(Function.identity()) 
     .distinct() 
     .sorted() 
     .limit(10) 
     .forEachOrdered(System.out::println); 
     throw new RuntimeException("simulated processing failure"); 
    } 
}