Convertire `BufferedReader` a` `Flusso <String> in modo parallelo

Question

C'è un modo per ricevere un Stream<String> stream su un BufferedReader reader tale che ogni stringa in stream rappresenta una linea di reader con l'ulteriore condizione che stream viene realizzato direttamente (prima reader leggi tutto)? Voglio elaborare i dati di stream in parallelo per ottenerli da reader per risparmiare tempo.

Modifica: desidero elaborare i dati parallelamente alla lettura. Non voglio elaborare linee diverse parallele. Dovrebbero essere elaborati in ordine.

Facciamo un esempio su come voglio risparmiare tempo. Diciamo che il nostro reader presenterà 100 linee per noi. Occorrono 2 ms per leggere una riga e 1 ms per elaborarla. Se prima leggo tutte le righe e poi le elaboro, mi occorreranno 300 ms. Quello che voglio fare è: non appena viene letta una riga, voglio elaborarla e leggere in parallelo la riga successiva. Il tempo totale sarà quindi di 201 ms.

Quello che non mi piace di BufferedReader.lines(): Per quanto ho capito la lettura inizia quando voglio elaborare le linee. Supponiamo di avere già il mio reader, ma devo fare precomputazioni prima di poter elaborare la prima linea. Diciamo che costano 30 ms. Nell'esempio sopra il tempo totale sarebbe quindi di 231 ms o 301 ms utilizzando reader.lines() (puoi dirmi quale di quelle volte è corretta?). Ma sarebbe possibile portare a termine il lavoro in 201 ms, dal momento che le precomputazioni possono essere eseguite parallelamente alla lettura delle prime 15 righe.

Answer 1

È possibile utilizzare reader.lines().parallel(). In questo modo il tuo input verrà suddiviso in blocchi e ulteriori operazioni di streaming verranno eseguite su blocchi in parallelo. Se ulteriori operazioni richiedono molto tempo, è possibile ottenere un miglioramento delle prestazioni.

Nel tuo caso l'euristica di default non funzionerà come vuoi e immagino che non ci sia una soluzione pronta che ti permetta di utilizzare lotti a linea singola. Puoi scrivere uno splitterator personalizzato che si dividerà dopo ogni riga. Esaminare l'implementazione java.util.Spliterators.AbstractSpliterator. Probabilmente la soluzione più semplice sarebbe scrivere qualcosa di simile, ma limitare le dimensioni del batch a un elemento in trySplit e leggere la riga singola nel metodo tryAdvance.

Answer 2

Per fare ciò che si desidera, si dispone in genere di un thread che legge le righe e li aggiunge a una coda di blocco e un secondo thread che ottiene le righe da questa coda di blocco e le elabora.

Answer 3

Stai guardando nel posto sbagliato. Stai pensando che un flusso di linee leggerà le linee dal file ma non è così che funziona. Non si può dire al sistema sottostante di leggere una riga poiché nessuno sa cosa sia una linea prima di leggere.

A BufferedReader ha il suo nome a causa del suo buffer di caratteri. Questo buffer ha una capacità predefinita di 8192. Ogni volta che viene richiesta una nuova riga, il buffer verrà analizzato per una sequenza di nuova riga e la parte verrà restituita. Quando il buffer non contiene abbastanza caratteri per trovare una riga corrente, verrà riempito l'intero buffer.

Ora, riempiendo il buffer può portare a richieste per leggere byte dal sottostante InputStream per riempire il buffer del decodificatore carattere. Quanti byte saranno richiesti e quanti byte saranno effettivamente letti dipende dalla dimensione del buffer del decodificatore di caratteri, da quanti byte della mappa di codifica effettiva a un carattere e se il sottostante InputStream ha il proprio buffer e quanto è grande .

L'operazione costosa effettiva è la lettura dei byte dal flusso sottostante e non vi è alcuna associazione banale dalle richieste di lettura di riga a queste operazioni di lettura. Richiedere la prima riga può causare la lettura, diciamo un chunk di 16 KiB dal file sottostante, e le successive cento richieste potrebbero essere servite dal buffer pieno e non causare alcun I/O. E nulla di ciò che fai con l'API Stream può cambiare nulla a riguardo. L'unica cosa che si dovrebbe parallelizzare è la ricerca di nuovi caratteri di linea nel buffer che è troppo banale per beneficiare dell'esecuzione parallela.

È possibile ridurre le dimensioni del buffer di tutte le parti coinvolte per ottenere approssimativamente la lettura parallela di una linea durante l'elaborazione della riga precedente, tuttavia, l'esecuzione parallela non compenserà mai il degrado delle prestazioni causato dalle dimensioni del buffer piccolo.