Java parallel stream - urutan pemanggilan metode parallel () [ditutup]

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

Ketika saya menulis ini, saya berasumsi bahwa utas akan memunculkan hanya panggilan peta karena paralel ditempatkan setelah peta. Tetapi beberapa baris dalam file mendapatkan nomor catatan yang berbeda untuk setiap eksekusi.

Saya membaca dokumentasi stream Java resmi dan beberapa situs web untuk memahami bagaimana stream bekerja di bawah tenda.

Beberapa pertanyaan:

• Java parallel stream bekerja berdasarkan SplitIterator , yang diimplementasikan oleh setiap koleksi seperti ArrayList, LinkedList dll. Ketika kita membangun aliran paralel dari koleksi-koleksi itu, iterator split yang sesuai akan digunakan untuk membagi dan mengulangi koleksi. Ini menjelaskan mengapa paralelisme terjadi pada tingkat sumber input asli (File lines) bukan pada hasil peta (yaitu Rekam pojo). Apakah pemahaman saya benar?

• Dalam kasus saya, inputnya adalah aliran file IO. Iterator split mana yang akan digunakan?

• Tidak masalah di mana kita menempatkan parallel()pipa. Sumber input asli akan selalu dipecah dan operasi perantara yang tersisa akan diterapkan.

  Dalam hal ini, Java seharusnya tidak mengizinkan pengguna untuk menempatkan operasi paralel di mana saja dalam pipa kecuali pada sumber aslinya. Karena, ini memberikan pemahaman yang salah bagi mereka yang tidak tahu bagaimana java stream bekerja secara internal. Saya tahu parallel()operasi akan ditentukan untuk tipe objek Stream dan karenanya, ia bekerja dengan cara ini. Tetapi, lebih baik memberikan beberapa solusi alternatif.

• Dalam cuplikan kode di atas, saya mencoba menambahkan nomor baris ke setiap catatan dalam file input dan karenanya harus dipesan. Namun, saya ingin menerapkannya doSomeOperation()secara paralel karena ini adalah logika yang berat. Satu-satunya cara untuk mencapai adalah menulis iterator split saya sendiri. Apakah ada cara lain?

Ini menjelaskan mengapa paralelisme terjadi pada tingkat sumber input asli (File lines) bukan pada hasil peta (yaitu Rekam pojo).

Seluruh aliran bersifat paralel atau berurutan. Kami tidak memilih subset operasi untuk dijalankan secara berurutan atau paralel.

Ketika operasi terminal dimulai, pipa aliran dijalankan secara berurutan atau paralel tergantung pada orientasi aliran yang digunakannya. [...] Ketika operasi terminal dimulai, pipa aliran dijalankan secara berurutan atau paralel tergantung pada mode aliran yang digunakan. sumber yang sama

Seperti yang Anda sebutkan, stream paralel menggunakan iterator split. Jelas, ini untuk mempartisi data sebelum operasi mulai berjalan.

Dalam kasus saya, inputnya adalah aliran file IO. Iterator split mana yang akan digunakan?

Melihat sumbernya, saya melihatnya menggunakan java.nio.file.FileChannelLinesSpliterator

Tidak masalah di mana kami menempatkan paralel () di dalam pipa. Sumber input asli akan selalu dipecah dan operasi perantara yang tersisa akan diterapkan.

Baik. Anda bahkan dapat menelepon parallel()dan sequential()beberapa kali. Yang dipanggil terakhir akan menang. Saat kami menelepon parallel(), kami mengaturnya untuk aliran yang dikembalikan; dan seperti yang dinyatakan di atas, semua operasi berjalan secara berurutan atau paralel.

Dalam hal ini, Java seharusnya tidak mengizinkan pengguna untuk melakukan operasi paralel di mana saja dalam pipa kecuali pada sumber aslinya ...

Ini menjadi masalah pendapat. Saya pikir Zabuza memberikan alasan yang bagus untuk mendukung pilihan desainer JDK.

Satu-satunya cara untuk mencapai adalah menulis iterator split saya sendiri. Apakah ada cara lain?

Ini tergantung pada operasi Anda

• Jika findFirst()operasi terminal Anda sebenarnya, maka Anda bahkan tidak perlu khawatir tentang eksekusi paralel, karena bagaimanapun juga tidak akan ada banyak panggilan doSomething()( findFirst()adalah hubungan arus pendek). .parallel()sebenarnya dapat menyebabkan lebih dari satu elemen diproses, sementara findFirst()pada aliran sekuensial akan mencegahnya.

• Jika operasi terminal Anda tidak menghasilkan banyak data, maka mungkin Anda bisa membuat Recordobjek menggunakan aliran sekuensial, kemudian memproses hasilnya secara paralel:

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

• Jika pipa Anda akan memuat banyak data dalam memori (yang mungkin menjadi alasan Anda menggunakan Files.lines()), maka mungkin Anda akan memerlukan iterator pemecah kustom. Sebelum saya pergi ke sana, saya akan melihat opsi lain (seperti garis simpanan dengan kolom id untuk memulai - itu hanya pendapat saya).
  Saya juga mencoba memproses catatan dalam kumpulan yang lebih kecil, seperti ini:

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

  Ini dijalankan doSomeOperation()secara paralel tanpa memuat semua data ke dalam memori. Tetapi perhatikan bahwa batchSizeperlu dipikirkan.

Desain Stream asli menyertakan ide untuk mendukung tahapan pipa berikutnya dengan pengaturan eksekusi paralel yang berbeda, tetapi ide ini telah ditinggalkan. API dapat berasal dari saat ini, tetapi di sisi lain, desain API yang memaksa penelepon untuk membuat keputusan tunggal untuk eksekusi paralel atau berurutan akan jauh lebih rumit.

Yang sebenarnya Spliteratordigunakan oleh Files.lines(…)tergantung pada implementasi. Di Java 8 (Oracle atau OpenJDK), Anda selalu mendapatkan yang sama dengan BufferedReader.lines(). Di JDK yang lebih baru, jika Pathmilik filesystem default dan charset adalah salah satu yang didukung untuk fitur ini, Anda mendapatkan Stream dengan Spliteratorimplementasi khusus , the java.nio.file.FileChannelLinesSpliterator. Jika prasyarat tidak terpenuhi, Anda mendapatkan yang sama dengan BufferedReader.lines(), yang masih didasarkan pada yang Iteratorditerapkan di dalam BufferedReaderdan dibungkus melalui Spliterators.spliteratorUnknownSize.

Tugas spesifik Anda paling baik ditangani dengan kebiasaan Spliteratoryang dapat melakukan penomoran baris langsung pada sumbernya, sebelum pemrosesan paralel, untuk memungkinkan pemrosesan paralel berikutnya tanpa batasan.

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

Dan berikut ini adalah demonstrasi sederhana ketika penerapan paralel diterapkan. Output dari mengintip dengan jelas menunjukkan perbedaan antara dua contoh. Catatan: mapPanggilan baru saja dilemparkan untuk menambahkan metode lain sebelum 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();