Design of thread pool reuse with Executors.newFixedThreadPool in Java

Question

In my application I have code which should run every hour on a new items and execute some expensive operation with multithreading:

int i = 1;

//noinspection InfiniteLoopStatement
while (true) {

    try {
        Logger.printIterationNum("Text analysis", "beg", i);

        Document query = new Document("fetchStatus", new Document("$lte", fetchStatusParam));
        ArrayList<Document> unfetchedEvents = dbC_Events.find(query).projection(
                fields(include("_id"), include("link"), include("title"), include("summary"))
        ).into(new ArrayList<>());

        // get full text and images of the events
        this.getFullTextImg(dbC_Events, unfetchedEvents);

        Logger.printIterationNum("Text analysis", "end", i++);

        // make a pause between executions
        Pauser.rest(Pauser.Units.SECONDS, PAUSE_TIME_OUT);
    } catch (Exception e) {
        i = 1;
        e.printStackTrace();
    }
}

private void getFullTextImg(MongoCollection<Document> dbC_Events, ArrayList<Document> unfetchedEvents) {

    int cpuCoresNum = SYS_MAX_NUM_CPU_CORES;
    int itemsPerCore = unfetchedEvents.size() / cpuCoresNum;

    // prevent infinite thread generation if num of unfetched events is less than number of CPU cores
    if (itemsPerCore == 0) {
        itemsPerCore = unfetchedEvents.size();
    }

    // update status of queried items
    DBAgent.updateItemsStatus(dbC_Events, unfetchedEvents, FetchStatus.IN_PROCESS_FETCH.getID());

    int itNum = 1;
    int upperIndex = 0;

    ExecutorService service = Executors.newFixedThreadPool(cpuCoresNum);

    for (int i = 0; i < unfetchedEvents.size(); i += itemsPerCore) {

        int indexFrom = i;
        upperIndex += itemsPerCore;
        int toIndex = upperIndex;

        if (((unfetchedEvents.size() - toIndex) <= itemsPerCore) && (itNum == cpuCoresNum)) {
            toIndex = unfetchedEvents.size();
            i = unfetchedEvents.size();
        }

        int indexTo = toIndex;
        int treadID = itNum++;

        service.execute(() -> {
            try {
                Thread.currentThread().setName("im_evFullTextImg_#" + String.format("%03d", treadID));
                this.fetcher(dbC_Events, new ArrayList<>(unfetchedEvents.subList(indexFrom, indexTo)));
            } catch (Exception e) {
                e.printStackTrace();
            }
        });
    }

    // wait until all threads will be finished
    service.shutdown();
    try {
        service.awaitTermination(Long.MAX_VALUE, TimeUnit.MINUTES);
    } catch (InterruptedException e) {
        e.printStackTrace();
    }
}

My questions:

1. Is it a good idea to initialize the Executors.newFixedThreadPool(cpuCoresNum) each time I execute the getFullTextImg() method, or is it better to initialize it once and pass it as a parameter each time I call getFullTextImg()?

2. If I initialize Executors.newFixedThreadPool(cpuCoresNum) out of getFullTextImg() and pass it as a paramete, will it cause some issues since I use service.shutdown();, which prevents accepting new tasks to the thread pool?

3. If service.shutdown(); is a problematic to reuse the thread pool, how can I redesign it in order to still wait until all threads in a poll will be finished, and only after to continue the code flow?

4. Can the Executors.newFixedThreadPool(cpuCoresNum) reuse lead to memory leak issues? Is there some important thing I have to do (e.g. clean the pool) with the thread pool in order to reuse it?

Answer 1

Answers:

1. It is better to reuse the ExecutorService. Destroying threads and then recreating this is expensive. (The only scenario where destroying and recreating a pool might be advisable is when the pool is going to be unused for a long time, and physical memory usage is a problem. But even then, some flavors of executor are able of shrinking the thread pool size when there is no demand.)

2. Well yes, that would be a problem. But that really means that you should not call shutdown on an executor if you still may need to submit new tasks in it.

3. You can't reuse an executor after calling shutdown on it.

4. There should be no memory leaks ... apart from the fact that the idle threads themselves are using memory (for thread stacks), and will continue to use it until the executor is shut down. For a fixed sized / bounded sized pool, that memory usage by threads is bounded. Most people would not call a bounded memory usage a "leak".

   In fact, not reusing an executor is likely to lead to a (real) memory leaks= if you somehow neglect to call shutdown on the executor.

Answer 2

For #3:

Specifically for this point, to make sure all the tasks in the pool have been closed and then proceed in executing next lines of the code, after the shutdown() command, just use this service.isTerminated() Java function. It returns true when all the tasks in the pool have been closed, then the loop will block execution through a busy wait until everything is closed.

Here is an example:

service.shutdown();
while (!service.isTerminated());
System.out.println("Finished all threads");