I have a .csv file containing over 70 million lines of which each line is to generate a Runnable and then executed by threadpool. This Runnable will insert a record into Mysql.
What's more , I want to record a position of the csv file for the RandomAccessFile to locate. The position is written to a File.I want to write this record when all the threads in threadpool are finished.So ThreadPoolExecutor.shutdown() is invoked. But when more lines come, I need a threadpool again. How can I reuse this current threadpool instead of make a new one.
The code is as follows:
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;
}
Thanks !
After calling shutdown on a ExecutorService, no new Task will be accepted. This means you have to create a new ExecutorService for each round of tasks.
Reduced number of threads in our applicationNow we can reuse the same thread pool to run concurrent tasks.
You can reuse the executor service if you restructure your code somewhat. Basically you collect all your tasks, execute them, and await execution before proceeding. Of course, you could also alternatively just use a new Executor Service for each of your time steps, but at least you have options.
Once a thread in the thread pool completes its task, it's returned to a queue of waiting threads. From this moment it can be reused. This reuse enables applications to avoid the cost of creating a new thread for each task. There is only one thread pool per process.
As stated in the documentation, you cannot reuse an ExecutorService
that has been shut down. I'd recommend against any workarounds, since (a) they may not work as expected in all situations; and (b) you can achieve what you want using standard classes.
You must either
instantiate a new ExecutorService
; or
not terminate the ExecutorService
.
The first solution is easily implemented, so I won't detail it.
For the second, since you want to execute an action once all the submitted tasks have finished, you might take a look at ExecutorCompletionService
and use it instead. It wraps an ExecutorService
which will do the thread management, but the runnables will get wrapped into something that will tell the ExecutorCompletionService
when they have finished, so it can report back to you:
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(); }
The method take()
on the ExecutorCompletionService
class will block until a task has finished (either normally or abruptly). It will return a Future
, so you can check the results if you wish.
I hope this can help you, since I didn't completely understand your problem.
create and group all tasks and submit them to the pool with invokeAll (which only returns when all tasks are successfully completed)
If you love us? You can donate to us via Paypal or buy me a coffee so we can maintain and grow! Thank you!
Donate Us With