Fluxo paralelo Java - ordem de chamar o método parallel () [fechado]

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

Quando escrevi isso, assumi que os threads seriam gerados apenas na chamada do mapa, pois o paralelo é colocado após o mapa. Mas algumas linhas no arquivo estavam recebendo números de registros diferentes para cada execução.

Li a documentação oficial do fluxo Java e alguns sites para entender como os fluxos funcionam sob o capô.

Algumas questões:

• O fluxo paralelo de Java funciona com base no SplitIterator , que é implementado por todas as coleções como ArrayList, LinkedList etc. Quando construímos um fluxo paralelo a partir dessas coleções, o iterador de divisão correspondente será usado para dividir e iterar a coleção. Isso explica por que o paralelismo aconteceu no nível da fonte de entrada original (linhas de arquivo) e não no resultado do mapa (ou seja, Record pojo). Meu entendimento está correto?

• No meu caso, a entrada é um fluxo de E / S de arquivo. Qual iterador dividido será usado?

• Não importa onde colocamos parallel()no pipeline. A fonte de entrada original será sempre dividida e as operações intermediárias restantes serão aplicadas.

  Nesse caso, o Java não deve permitir que os usuários coloquem operações paralelas em qualquer lugar do pipeline, exceto na fonte original. Porque está dando um entendimento errado para quem não sabe como o java stream funciona internamente. Eu sei que a parallel()operação teria sido definida para o tipo de objeto Stream e, portanto, está funcionando dessa maneira. Mas, é melhor fornecer uma solução alternativa.

• No trecho de código acima, estou tentando adicionar um número de linha a cada registro no arquivo de entrada e, portanto, ele deve ser solicitado. No entanto, quero aplicar doSomeOperation()em paralelo, pois é uma lógica de peso pesado. A única maneira de conseguir é escrever meu próprio iterador dividido personalizado. Existe alguma outra maneira?

Isso explica por que o paralelismo aconteceu no nível da fonte de entrada original (linhas do arquivo) e não no resultado do mapa (ou seja, Record pojo).

O fluxo inteiro é paralelo ou seqüencial. Não selecionamos um subconjunto de operações para executar sequencialmente ou em paralelo.

Quando a operação do terminal é iniciada, o pipeline de fluxo é executado sequencialmente ou em paralelo, dependendo da orientação do fluxo no qual é chamado. [...] Quando a operação do terminal é iniciada, o pipeline de fluxo é executado sequencialmente ou em paralelo, dependendo do modo do fluxo no qual é chamado. mesma fonte

Como você mencionou, os fluxos paralelos usam iteradores divididos. Claramente, isso é para particionar os dados antes que as operações comecem a ser executadas.

No meu caso, a entrada é um fluxo de E / S de arquivo. Qual iterador dividido será usado?

Olhando a fonte, vejo que ela usa java.nio.file.FileChannelLinesSpliterator

Não importa onde colocamos paralelo () no pipeline. A fonte de entrada original será sempre dividida e as operações intermediárias restantes serão aplicadas.

Direita. Você pode até ligar parallel()e sequential()várias vezes. O último invocado vencerá. Quando chamamos parallel(), definimos isso para o fluxo retornado; e, como mencionado acima, todas as operações são executadas sequencialmente ou em paralelo.

Nesse caso, Java não deve permitir que os usuários coloquem operações paralelas em qualquer lugar do pipeline, exceto na fonte original ...

Isso se torna uma questão de opiniões. Eu acho que Zabuza dá um bom motivo para apoiar a escolha dos designers do JDK.

A única maneira de conseguir é escrever meu próprio iterador dividido personalizado. Existe alguma outra maneira?

Isso depende das suas operações

• Se findFirst()é a sua operação real do terminal, você nem precisa se preocupar com a execução paralela, porque de doSomething()qualquer maneira não haverá muitas chamadas ( findFirst()está em curto-circuito). .parallel()de fato, pode fazer com que mais de um elemento seja processado, enquanto que findFirst()em um fluxo seqüencial impediria isso.

• Se a operação do terminal não criar muitos dados, talvez você possa criar seus Recordobjetos usando um fluxo sequencial e processe o resultado em paralelo:

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

• Se o seu pipeline carrega muitos dados na memória (que pode ser o motivo pelo qual você está usando Files.lines()), talvez seja necessário um iterador dividido personalizado. Antes de ir para lá, porém, eu procurava outras opções (como salvar linhas com uma coluna de identificação - para começar - essa é apenas a minha opinião).
  Eu também tentaria processar registros em lotes menores, como este:

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

  Isso é executado doSomeOperation()em paralelo sem carregar todos os dados na memória. Mas observe que batchSizeserá necessário pensar um pouco.

O design original do Stream incluía a ideia de dar suporte aos estágios subsequentes do pipeline com diferentes configurações de execução paralela, mas essa ideia foi abandonada. A API pode resultar desse momento, mas, por outro lado, um design de API que força o chamador a tomar uma única decisão inequívoca para execução paralela ou sequencial seria muito mais complicado.

O real Spliteratorem uso por Files.lines(…)depende da implementação. No Java 8 (Oracle ou OpenJDK), você sempre obtém o mesmo que com BufferedReader.lines(). Nos JDKs mais recentes, se o Pathpertence ao sistema de arquivos padrão e o charset é um dos suportados para esse recurso, você obtém um Stream com uma Spliteratorimplementação dedicada , o java.nio.file.FileChannelLinesSpliterator. Se as pré-condições não forem atendidas, você obtém o mesmo que com BufferedReader.lines(), que ainda é baseado em uma Iteratorimplementação interna BufferedReadere via empacotada Spliterators.spliteratorUnknownSize.

Sua tarefa específica é melhor gerenciada com um costume Spliteratorque pode executar a numeração de linha diretamente na origem, antes do processamento paralelo, para permitir o processamento paralelo subsequente sem restrições.

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

E a seguir é uma demonstração simples de quando a aplicação de paralelo é aplicada. A saída do peek mostra claramente a diferença entre os dois exemplos. Nota: A mapchamada é lançada apenas para adicionar outro método antes de 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();