Równoległy strumień Java - kolejność wywoływania metody parallel () [zamknięte]

AtomicInteger recordNumber = new AtomicInteger();
Files.lines(inputFile.toPath(), StandardCharsets.UTF_8)
     .map(record -> new Record(recordNumber.incrementAndGet(), record)) 
     .parallel()           
     .filter(record -> doSomeOperation())
     .findFirst()

Kiedy to napisałem, założyłem, że wątki zostaną odrodzone tylko wywołanie mapy, ponieważ równolegle jest umieszczane po mapie. Ale niektóre wiersze w pliku otrzymywały różne numery rekordów dla każdego wykonania.

Przeczytałem oficjalną dokumentację dotyczącą strumieni Java i kilka stron internetowych, aby zrozumieć, jak działają strumienie pod maską.

Kilka pytań:

• Równoległy strumień Java działa w oparciu o SplitIterator , który jest implementowany przez każdą kolekcję, taką jak ArrayList, LinkedList itp. Kiedy konstruujemy równoległy strumień z tych kolekcji, odpowiedni podzielony iterator zostanie użyty do podzielenia i iteracji kolekcji. To wyjaśnia, dlaczego równoległość występowała na poziomie oryginalnego źródła wejściowego (wierszy pliku), a nie na podstawie wyniku mapy (tj. Zapis pojo). Czy moje rozumowanie jest prawidłowe?

• W moim przypadku wejściem jest plikowy strumień IO. Który iterator podzielony zostanie użyty?

• Nie ma znaczenia, gdzie umieszczamy parallel()w rurociągu. Oryginalne źródło wejściowe będzie zawsze podzielone i zostaną zastosowane pozostałe operacje pośrednie.

  W takim przypadku Java nie powinna umożliwiać użytkownikom wykonywania operacji równoległych w dowolnym miejscu potoku, z wyjątkiem oryginalnego źródła. Ponieważ daje złe zrozumienie dla tych, którzy nie wiedzą, jak działa strumień Java. Wiem, że parallel()operacja byłaby zdefiniowana dla typu obiektu Stream, więc działa w ten sposób. Ale lepiej jest podać alternatywne rozwiązanie.

• W powyższym fragmencie kodu próbuję dodać numer wiersza do każdego rekordu w pliku wejściowym, więc należy go zamówić. Chcę jednak zastosować doSomeOperation()równolegle, ponieważ jest to logika ciężka. Jednym ze sposobów na osiągnięcie tego jest napisanie własnego dostosowanego podzielonego iteratora. Czy jest jakiś inny sposób?

To wyjaśnia, dlaczego równoległość występowała na poziomie oryginalnego źródła wejściowego (wierszy pliku), a nie na podstawie wyniku mapy (tj. Zapis pojo).

Cały strumień ma charakter równoległy lub sekwencyjny. Nie wybieramy podzbioru operacji do uruchomienia sekwencyjnego lub równoległego.

Po zainicjowaniu operacji terminalowej potok strumienia jest wykonywany sekwencyjnie lub równolegle w zależności od orientacji strumienia, na którym jest wywoływany. [...] Gdy inicjowana jest operacja terminalowa, potok strumienia jest wykonywany sekwencyjnie lub równolegle, w zależności od trybu strumienia, na który jest wywoływany. to samo źródło

Jak wspomniałeś, równoległe strumienie używają podzielonych iteratorów. Oczywiście jest to dzielenie danych na partycje przed rozpoczęciem operacji.

W moim przypadku wejściem jest plikowy strumień IO. Który iterator podzielony zostanie użyty?

Patrząc na źródło, widzę, że używa java.nio.file.FileChannelLinesSpliterator

Nie ma znaczenia, gdzie umieszczamy równolegle () w potoku. Oryginalne źródło wejściowe będzie zawsze podzielone i zostaną zastosowane pozostałe operacje pośrednie.

Dobrze. Możesz nawet dzwonić parallel()i sequential()wiele razy. Ten przywołany jako ostatni wygra. Kiedy dzwonimy parallel(), ustawiamy to dla zwracanego strumienia; i jak stwierdzono powyżej, wszystkie operacje są uruchamiane sekwencyjnie lub równolegle.

W takim przypadku Java nie powinna pozwalać użytkownikom na umieszczanie operacji równoległych w dowolnym miejscu potoku, z wyjątkiem oryginalnego źródła ...

To staje się kwestią opinii. Myślę, że Zabuza daje dobry powód, aby poprzeć wybór projektantów JDK.

Jednym ze sposobów na osiągnięcie tego jest napisanie własnego dostosowanego podzielonego iteratora. Czy jest jakiś inny sposób?

To zależy od twoich operacji

• Jeśli findFirst()jest to Twoja prawdziwa operacja terminalowa, nie musisz nawet martwić się o równoległe wykonywanie, ponieważ i tak nie będzie wielu połączeń doSomething()( findFirst()jest to zwarcie). .parallel()w rzeczywistości może powodować przetwarzanie więcej niż jednego elementu, podczas gdy findFirst()w strumieniu sekwencyjnym zapobiegałoby to.

• Jeśli twoja terminalowa operacja nie tworzy dużej ilości danych, być może możesz stworzyć swoje Recordobiekty za pomocą sekwencyjnego strumienia, a następnie przetworzyć wynik równolegle:

  List<Record> smallData = Files.lines(inputFile.toPath(), 
                                       StandardCharsets.UTF_8)
    .map(record -> new Record(recordNumber.incrementAndGet(), record)) 
    .collect(Collectors.toList())
    .parallelStream()     
    .filter(record -> doSomeOperation())
    .collect(Collectors.toList());

• Jeśli Twój potok załadowałby dużo danych do pamięci (co może być powodem, którego używasz Files.lines()), być może będziesz potrzebować niestandardowego iteratora podziału. Zanim jednak tam przejdę, przyjrzę się innym opcjom (na przykład zapisywanie wierszy z kolumną identyfikacyjną - to tylko moja opinia).
  Próbowałbym również przetwarzać rekordy w mniejszych partiach, takich jak to:

  AtomicInteger recordNumber = new AtomicInteger();
  final int batchSize = 10;
  
  try(BufferedReader reader = Files.newBufferedReader(inputFile.toPath(), 
          StandardCharsets.UTF_8);) {
      Supplier<List<Record>> batchSupplier = () -> {
          List<Record> batch = new ArrayList<>();
          for (int i = 0; i < batchSize; i++) {
              String nextLine;
              try {
                  nextLine = reader.readLine();
              } catch (IOException e) {
                  //hanlde exception
                  throw new RuntimeException(e);
              }
  
              if(null == nextLine) 
                  return batch;
              batch.add(new Record(recordNumber.getAndIncrement(), nextLine));
          }
          System.out.println("next batch");
  
          return batch;
      };
  
      Stream.generate(batchSupplier)
          .takeWhile(list -> list.size() >= batchSize)
          .map(list -> list.parallelStream()
                           .filter(record -> doSomeOperation())
                           .collect(Collectors.toList()))
          .flatMap(List::stream)
          .forEach(System.out::println);
  }

  Wykonuje się to doSomeOperation()równolegle bez ładowania wszystkich danych do pamięci. Pamiętaj jednak, że batchSizetrzeba to przemyśleć.

Oryginalny projekt Stream zawierał pomysł na obsługę kolejnych etapów potoku z różnymi ustawieniami wykonywania równoległego, ale pomysł ten został porzucony. Interfejs API może wynikać z tego czasu, ale z drugiej strony projekt interfejsu API, który zmusza osobę wywołującą do podjęcia pojedynczej jednoznacznej decyzji o równoległym lub sekwencyjnym wykonaniu, byłby znacznie bardziej skomplikowany.

Rzeczywisty Spliteratorużywany przez Files.lines(…)zależy od implementacji. W Javie 8 (Oracle lub OpenJDK) zawsze otrzymujesz to samo, co z BufferedReader.lines(). W ostatnich pakietów JDK, jeśli Pathnależy do domyślnego systemu plików i charset jest jednym z obsługiwanych do tej funkcji, można uzyskać strumień z dedykowaną Spliteratorrealizacji, java.nio.file.FileChannelLinesSpliterator. Jeśli warunki wstępne nie są spełnione, otrzymujesz to samo, co z BufferedReader.lines(), który wciąż jest oparty na Iteratorzaimplementowanym wewnątrz BufferedReaderi jest zawinięty przez Spliterators.spliteratorUnknownSize.

Twoje konkretne zadanie najlepiej realizować za pomocą niestandardowego, Spliteratorktóry może wykonać numerację linii bezpośrednio u źródła, przed przetwarzaniem równoległym, aby umożliwić późniejsze przetwarzanie równoległe bez ograniczeń.

public static Stream<Record> records(Path p) throws IOException {
    LineNoSpliterator sp = new LineNoSpliterator(p);
    return StreamSupport.stream(sp, false).onClose(sp);
}

private static class LineNoSpliterator implements Spliterator<Record>, Runnable {
    int chunkSize = 100;
    SeekableByteChannel channel;
    LineNumberReader reader;

    LineNoSpliterator(Path path) throws IOException {
        channel = Files.newByteChannel(path, StandardOpenOption.READ);
        reader=new LineNumberReader(Channels.newReader(channel,StandardCharsets.UTF_8));
    }

    @Override
    public void run() {
        try(Closeable c1 = reader; Closeable c2 = channel) {}
        catch(IOException ex) { throw new UncheckedIOException(ex); }
        finally { reader = null; channel = null; }
    }

    @Override
    public boolean tryAdvance(Consumer<? super Record> action) {
        try {
            String line = reader.readLine();
            if(line == null) return false;
            action.accept(new Record(reader.getLineNumber(), line));
            return true;
        } catch (IOException ex) {
            throw new UncheckedIOException(ex);
        }
    }

    @Override
    public Spliterator<Record> trySplit() {
        Record[] chunks = new Record[chunkSize];
        int read;
        for(read = 0; read < chunks.length; read++) {
            int pos = read;
            if(!tryAdvance(r -> chunks[pos] = r)) break;
        }
        return Spliterators.spliterator(chunks, 0, read, characteristics());
    }

    @Override
    public long estimateSize() {
        try {
            return (channel.size() - channel.position()) / 60;
        } catch (IOException ex) {
            return 0;
        }
    }

    @Override
    public int characteristics() {
        return ORDERED | NONNULL | DISTINCT;
    }
}

Poniżej przedstawiono prostą demonstrację zastosowania równoległego. Dane wyjściowe z podglądu wyraźnie pokazują różnicę między tymi dwoma przykładami. Uwaga: mappołączenie jest po prostu rzucane, aby dodać inną metodę przed parallel.

IntStream.rangeClosed (1,20).peek(a->System.out.print(a+" "))
        .map(a->a + 200).sum();
System.out.println();
IntStream.rangeClosed(1,20).peek(a->System.out.print(a+" "))
        .map(a->a + 200).parallel().sum();