Utilizzo di ArrayBlockingQueue rende il processo più lento

Question

Recentemente ho utilizzato ArrayBlockingQueue per il mio processo multi-thread. Ma sembrava che rallentasse piuttosto che accelerare. Ragazzi potreste aiutarmi? Sto fondamentalmente importazione di un file (circa 300k righe) e l'analisi e la loro memorizzazione nel DB

public class CellPool { 
private static class RejectedHandler implements RejectedExecutionHandler { 
    @Override 
    public void rejectedExecution(Runnable arg0, ThreadPoolExecutor arg1) { 
     System.err.println(Thread.currentThread().getName() + " execution rejected: " + arg0);  
    } 
    } 

    private static class Task implements Runnable { 
    private JSONObject obj; 

    public Task(JSONObject obj) { 
     this.obj = obj; 
    } 

    @Override 
    public void run() { 
     try { 
     Thread.sleep(1); 
     runThis(obj); 
     } catch (InterruptedException e) { 
     e.printStackTrace(); 
     } 
    } 

    public void runThis(JSONObject obj) { 
     //where the rows are parsed and stored in the DB, etc 
    } 
    } 

    public static void executeCellPool(String filename) throws InterruptedException { 
    // fixed pool fixed queue 
    BlockingQueue<Runnable> queue = new ArrayBlockingQueue<Runnable>(300000, true); 
    ThreadPoolExecutor executor = new ThreadPoolExecutor(90, 100, 1, TimeUnit.MINUTES, queue); 

    DataSet ds = CommonDelimitedParser.getDataSet(filename); 
    final String[] colNames = ds.getColumns(); 
    while (ds.next()) { 
     JSONObject obj = new JSONObject(); 
     //some JSON object 
     Task t = new Task(obj); 
     executor.execute(t); 
    } 
    } 

}

Answer 1

Se si vuole persistere record da un file in un database relazionale più velocemente possibile si dovrebbe utilizzare l'inserto batch JDBC piuttosto che inserire i record uno per uno.

Answer 2

tl; dr Le dimensioni grandi della coda possono avere un impatto negativo, così come il numero di thread elevati. Idealmente, vuoi che i tuoi consumatori e produttori lavorino a un ritmo simile.

Il motivo per cui l'aggiunta della coda causa problemi è perché si sta utilizzando una coda molto grande (che non è necessaria) che sta occupando le risorse. Tipicamente, una coda di blocco blocca i produttori quando non vi è spazio nella coda e consumatori quando non ci sono oggetti lasciati nella coda. Creando una dimensione così grande di una dimensione statica, Java sta assegnando quello spazio nella memoria quando quasi sicuramente non lo si sta utilizzando. Sarebbe più efficace forzare il produttore ad attendere lo spazio in coda per chiarire se i consumatori sono consumatori troppo lentamente. Non è necessario memorizzare tutte le righe del file nella coda contemporaneamente.

Le code Executor pool di thread sono discusse in javadoc here.

Code bloccate. Una coda limitata (ad esempio, ArrayBlockingQueue) aiuta a prevenire l'esaurimento delle risorse quando viene utilizzata con un massimo di MaximumPoolSizes, ma può essere più difficile da regolare e controllare. Le dimensioni della coda e le dimensioni massime del pool possono essere scambiate l'una per l'altra: l'utilizzo di code e piccoli pool di grandi dimensioni riduce al minimo l'utilizzo della CPU, le risorse del sistema operativo e il sovraccarico di commutazione del contesto, ma può comportare un throughput artificialmente basso. Se le attività si bloccano frequentemente (ad esempio se sono vincolate all'I/O), un sistema può essere in grado di pianificare il tempo per più thread di quanto non si consenta altrimenti. L'utilizzo di code di piccole dimensioni richiede generalmente dimensioni di pool maggiori, il che rende le CPU più impegnative, ma può verificarsi un sovraccarico di programmazione inaccettabile, che riduce anche il throughput.

L'ampia dimensione del thread di 90, combinata con la dimensione del pool molto grande di 300000, è molto probabile che utilizzi molta memoria e causi un ulteriore sovraccarico di pianificazione dei thread. Li lascerei entrambi considerevolmente. Non so su quale hardware si stia eseguendo, ma dato che sembra che si stia scrivendo un programma intensivo di I/O, proverei a raddoppiare il numero di thread che la CPU può gestire e a giocare con le dimensioni della coda di blocco per vedere ciò che funziona (nota: non ho studiato questo, questo è basato sulla mia esperienza nell'esecuzione di code ed esecutori, felice che altri suggeriscano un conteggio diverso!).

Di nota, tuttavia, è che il metodo execute() genera un errore RejectedExecutionException in caso di errore di aggiunta alla coda se la coda è troppo piccola. Un modo per monitorare la coda sarebbe controllare la sua capacità prima di pianificare un'attività. È possibile farlo chiamando:

executor.getQueue().remainingCapacity() 

Non utilizzare il metodo executor.getQueue() di alterare la coda in qualsiasi modo, ma può essere utilizzato per il monitoraggio.

Un'alternativa è utilizzare una coda illimitata, ad esempio LinkedBlockingQueue senza una capacità definita. In questo modo, non avrai bisogno di gestire le dimensioni delle code.Tuttavia, se i tuoi produttori funzionano molto più velocemente dei tuoi consumatori, avrai ancora una volta il problema di consumare troppa memoria.

Inoltre, kostya ha ragione, un inserimento batch JDBC sarebbe più veloce.