Lo streaming parallelo da un HashSet non viene eseguito in parallelo

Question

Ho una raccolta di elementi che desidero elaborare in parallelo. Quando utilizzo uno List, il parallelismo funziona. Tuttavia, quando utilizzo uno Set, non viene eseguito in parallelo.

Ho scritto un esempio di codice che mostra il problema:

public static void main(String[] args) { 
    ParallelTest test = new ParallelTest(); 

    List<Integer> list = Arrays.asList(1,2); 
    Set<Integer> set = new HashSet<>(list); 

    ForkJoinPool forkJoinPool = new ForkJoinPool(4); 

    System.out.println("set print"); 
    try { 
     forkJoinPool.submit(() -> 
      set.parallelStream().forEach(test::print) 
     ).get(); 
    } catch (Exception e) { 
     return; 
    } 

    System.out.println("\n\nlist print"); 
    try { 
     forkJoinPool.submit(() -> 
      list.parallelStream().forEach(test::print) 
     ).get(); 
    } catch (Exception e) { 
     return; 
    } 
} 

private void print(int i){ 
    System.out.println("start: " + i); 
    try { 
     TimeUnit.SECONDS.sleep(1); 
    } catch (InterruptedException e) { 
    } 
    System.out.println("end: " + i); 
} 

Questo è l'output che ottengo su Windows 7

set print 
start: 1 
end: 1 
start: 2 
end: 2 

list print 
start: 2 
start: 1 
end: 1 
end: 2 

Possiamo vedere che il primo elemento dal Set doveva terminare prima che il secondo elemento venga elaborato. Per List, il secondo elemento inizia prima che il primo elemento termini.

Puoi dirmi che cosa causa questo problema e come evitarlo usando una collezione Set?

Answer 1

Posso riprodurre il comportamento che si vede, dove il parallelismo non corrisponde al parallelismo del parallelismo del pool di fork-join che hai specificato. Dopo aver impostato il parallelismo del pool fork-join su 10 e aumentato il numero di elementi nella raccolta a 50, vedo il parallelismo del flusso basato su elenco che sale solo a 6, mentre il parallelismo del flusso basato su set non viene mai sopra 2.

noti, tuttavia, che questa tecnica di presentare un compito di una piscina fork-join per eseguire il flusso parallelo in quella piscina è un "trucco" attuazione ed è non garantita lavorare. In effetti, i thread o il pool di thread utilizzati per l'esecuzione di flussi paralleli sono non specificato. Per impostazione predefinita, viene utilizzato il pool comune di fork-join, ma in ambienti diversi potrebbero essere utilizzati diversi pool di thread. (Considerare un contenitore all'interno di un server applicazioni.)

Nella classe java.util.stream.AbstractTask, il campo LEAF_TARGET determina la quantità di divisione eseguita, che a sua volta determina la quantità di parallelismo che è possibile ottenere. Il valore di questo campo è basato su ForkJoinPool.getCommonPoolParallelism(), che ovviamente utilizza il parallelismo del pool comune, non quello che è il pool che esegue le attività.

Probabilmente si tratta di un bug (vedere problema OpenJDK JDK-8190974), tuttavia, l'intera area non è specificata comunque. Tuttavia, quest'area del sistema ha sicuramente bisogno di sviluppo, ad esempio in termini di politica di suddivisione, quantità di parallelismo disponibile, gestione delle attività di blocco, tra le altre questioni. Una futura versione di JDK potrebbe risolvere alcuni di questi problemi.

Nel frattempo, è possibile controllare il parallelismo del pool comune di fork-join attraverso l'uso delle proprietà di sistema. Se si aggiunge questa riga al vostro programma,

System.setProperty("java.util.concurrent.ForkJoinPool.common.parallelism", "10"); 

e si esegue i flussi in piscina comune (o se li si invia alla propria piscina che ha un livello sufficientemente elevato di set parallelismo) si potrà osservare che molti più attività vengono eseguite in parallelo.

È anche possibile impostare questa proprietà sulla riga di comando utilizzando l'opzione -D.

Ancora, questo non è un comportamento garantito e potrebbe cambiare in futuro. Ma questa tecnica probabilmente funzionerà per implementazioni JDK 8 per il prossimo futuro.