Executor service: How it acts as watchdog for thread-pools?

user_3438473 0
Viewed 424

I want to understand how ExecutorService acts as watchdog for the thread-pools that it creates.

Basically, as I understand, ExecutorService is just an object, it is not a "thread or process" which creates other threads.

Typically,for ThreadPoolExecutor:

threadPoolExecutor = (ThreadPoolExecutor) Executors.newFixedThreadPool(2);
// Now, requesting Executor Service to "execute" each submitted Tasks.
threadPoolExecutor.execute(runnable);

Likewise for ScheduleThreadPoolExecutor

scheduledThreadPool = Executors.newSingleThreadScheduledExecutor(threadFactory);
scheduledThreadPool.scheduleAtFixedRate(runnable, 2000, 3000, TimeUnit.MILLISECONDS);

Essentially, they are just objects, how are they able to , for example, "restart a thread if it dies".

I am not able to understand this, for example, in case of ScheduledThreadPoolExecutor we call method once for periodic action, after that , how that object is able to manage threads.

I did look into the code, I still have doubts as to how a object can manage all this? (creation of thread pools, submitting the Jobs to Queues, restarting threads in thread-pool and so on)

scheduledexecutorservice threadpoolexecutor executorservice multithreading java
1 Answers

You should have better look into the code. Actually, most of the executors has a private nested class that encapsulate your Runnable and managed it inside the thread.

Executors.newSingleThreadScheduledExecutor just create a ScheduledThreadPoolExecutor with one thread. And then when you push a task, it creates a ScheduledFutureTask (also encapsulated in a RunnableScheduledFuture) and creates a thread if necessary.

scheduleAtFixedRate :

public ScheduledFuture<?> scheduleAtFixedRate(Runnable command,
                                              long initialDelay,
                                              long period,
                                              TimeUnit unit) {
    ...
    RunnableScheduledFuture<?> t = decorateTask(command,
                                                new ScheduledFutureTask<Object>(command,
                                                                null,
                                                                triggerTime,
                                                                unit.toNanos(period)));
    delayedExecute(t);
    return t;
}

delayedExecute

private void delayedExecute(Runnable command) {
    if (isShutdown()) { // handling the cancellation 
        reject(command);
        return;
    }

    if (getPoolSize() < getCorePoolSize()) // increase number of thread if necessary
        prestartCoreThread();
    super.getQueue().add(command); // queue the task to be processed a soon as a task is finished
}

After been enqueued, the executor will unqueue one by one (in the case of a single thread executor, a thread pool with one thread), a Thread will eventually be created, also called workers in the code. And it will call the run() method of the previoussly enqueued task :

ScheduledFutureTask.run

public void run() {
    if (isPeriodic())
        runPeriodic();
    else
        ScheduledFutureTask.super.run(); // just call run() of your Runnable
}

Let's imagine that we previously submitted a task using scheduleAtFixedRate, the task will be considered as periodic and the method runPeriodic will be called :

ScheduledFutureTask.runPeriodic

private void runPeriodic() {
    boolean ok = ScheduledFutureTask.super.runAndReset(); // call run() from your Runnable
    boolean down = isShutdown();
    // Reschedule if not cancelled and not shutdown or policy allows
    if (ok && (!down ||
           (getContinueExistingPeriodicTasksAfterShutdownPolicy() && !isTerminating()))) {
        long p = period;
        if (p > 0)
            time += p;
        else
            time = now() - p;
        ScheduledThreadPoolExecutor.super.getQueue().add((Runnable) this);
    }
    // This might have been the final executed delayed
    // task.  Wake up threads to check.
    else if (down)
        interruptIdleWorkers();
}

This gives you an idea about how the magic happen. Most of the 'watchdog's' job happens inside the thread, they manage themselves. The job of the executor is to ensure that it's queue is always empty and dispatch and creates tasks over the threads. The behavior of tasks is then handled by themselves by accessing the Executor directly, thanks of the nested class possibility of Java.

user_50325410
Related