java - reutilizacion - forma más simple de reutilizar código

¿Cómo se puede reutilizar un threadpool después del cierre?

Question

Tengo un archivo .csv que contiene más de 70 millones de líneas, de las cuales cada línea generará Runnable y luego se ejecutará mediante threadpool. Este Runnable insertará un registro en Mysql.

Lo que es más, quiero registrar una posición del archivo csv para que RandomAccessFile pueda ubicarlo. La posición se escribe en un archivo . Deseo escribir este registro cuando todos los hilos en el hilo terminan. Se invoca ThreadPoolExecutor.shutdown (). Pero cuando lleguen más líneas, necesito un subproceso de nuevo. ¿Cómo puedo reutilizar este grupo de temas actual en lugar de crear uno nuevo?

El código es el siguiente:

public static boolean processPage() throws Exception {
long pos = getPosition();
long start = System.currentTimeMillis();
raf.seek(pos);
if(pos==0)
raf.readLine();
for (int i = 0; i < PAGESIZE; i++) {
String lineStr = raf.readLine();
if (lineStr == null)
return false;
String[] line = lineStr.split(",");
final ExperienceLogDO log = CsvExperienceLog.generateLog(line);
//System.out.println("userId: "+log.getUserId()%512);
pool.execute(new Runnable(){
public void run(){
try {
experienceService.insertExperienceLog(log);
} catch (BaseException e) {
e.printStackTrace();
}
}
});
long end = System.currentTimeMillis();
}
BufferedWriter resultWriter = new BufferedWriter(
new OutputStreamWriter(new FileOutputStream(new File(
RESULT_FILENAME), true)));
resultWriter.write("/n");
resultWriter.write(String.valueOf(raf.getFilePointer()));
resultWriter.close();
long time = System.currentTimeMillis()-start;
System.out.println(time);
return true;
}

Gracias !

Answer 1

Como se indica en la documentation , no puede reutilizar un ExecutorService que se ha cerrado. Lo recomendaría en contra de cualquier solución provisional , ya que (a) es posible que no funcionen como se esperaba en todas las situaciones; y (b) puede lograr lo que quiere utilizando clases estándar.

Usted debe

1. instanciar un nuevo ExecutorService ; o

2. no terminar el ExecutorService .

La primera solución se implementa fácilmente, así que no la detallaré.

Por el segundo, dado que desea ejecutar una acción una vez que todas las tareas enviadas hayan finalizado, puede echar un vistazo a ExecutorCompletionService y usarlo en su lugar. Envuelve un ExecutorService que hará la gestión de subprocesos, pero los ejecutables se envolverán en algo que le dirá al ExecutorCompletionService cuando hayan finalizado, por lo que puede informarte:

ExecutorService executor = ...;
ExecutorCompletionService ecs = new ExecutorCompletionService(executor);
for (int i = 0; i < totalTasks; i++) {
... ecs.submit(...); ...
}
for (int i = 0; i < totalTasks; i++) {
ecs.take();
}

El método take() en la clase ExecutorCompletionService se bloqueará hasta que una tarea haya finalizado (de forma normal o abrupta). Devolverá un Future , por lo que puede verificar los resultados si lo desea.

Espero que esto pueda ayudarlo, ya que no entendí completamente su problema.

Answer 2

Después de cerrar la llamada en un ExecutorService, no se aceptará ninguna nueva tarea . Esto significa que debe crear un nuevo ExecutorService para cada ronda de tareas.

Sin embargo, con Java 8 ForkJoinPool.awaitQuiescence se introdujo. Si puede cambiar de un ExecutorService normal a ForkJoinPool, puede usar este método para esperar hasta que no se ejecuten más tareas en ForkJoinPool sin tener que llamar al apagado. Esto significa que puede llenar un ForkJoinPool con Tareas, esperar hasta que esté vacío (en reposo) y luego comenzar a llenarlo con Tareas, y así sucesivamente.

Answer 3

crear y agrupar todas las tareas y enviarlas al grupo con invokeAll (que solo se devuelve cuando todas las tareas se completan con éxito)