Как правильно закрыть переменное количество потоков?

Вы можете написать составной AutoCloseable для управления динамическим количеством 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; }
    }
}

и вы можете использовать его в своем методе:

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

документация Stream.flatMap гласит:

Каждый отображаемый поток закрывается после помещения его содержимого в этот поток.

Другими словами, для обычного закрытия потоков не требуется никаких дополнительных действий. Однако, поскольку закрыты только обработанные потоки, не следует жадно создавать потоки, не зная, будут ли они впоследствии обработаны потоком:

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

Создавая подпотоки внутри flatMap, гарантируется, что каждый из них будет создан только в том случае, если поток будет его обрабатывать. Таким образом, это решение закроет все подпотоки даже без внешнего Stream внутри оператора try-with-resource.