Служба исполнителя: как она действует как сторожевой таймер для пулов потоков?

Question

Я хочу понять, как ExecutorService действует как сторожевой таймер для пулов потоков, которые он создает.

В основном, насколько я понимаю, ExecutorService - это просто объект, это не поток или process ", который создает другие потоки.

Обычно для ThreadPoolExecutor:

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

Аналогично для ScheduleThreadPoolExecutor

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

По сути, это просто объекты , как они могут, например, «перезапустить поток, если он умирает».

Я не могу этого понять, например, в случае ScheduledThreadPoolExecutor мы вызываем метод один раз для действия periodi c, после этого, как этот объект может управлять потоками.

Я изучил код, я все еще сомневаюсь, как объект может всем этим управлять? (создание пулов потоков, отправка заданий в очереди, перезапуск потоков в пуле потоков и т. д.)

Answer 1

Тебе стоит лучше изучить код. На самом деле, у большинства исполнителей есть частный вложенный класс, который инкапсулирует ваш Runnable и управляет им внутри потока.

Executors.newSingleThreadScheduledExecutor просто создает ScheduledThreadPoolExecutor с одной резьбой. А затем, когда вы выполняете sh задачу, она создает ScheduledFutureTask (также инкапсулируется в RunnableScheduledFuture ) и создает поток , если необходимо.

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
}

После постановки в очередь исполнитель удалит из очереди один за другим (в в случае однопоточного исполнителя, пула потоков с одним потоком), в конечном итоге будет создан Thread , также называемый в коде рабочими. И он вызовет метод run () задачи, ранее поставленной в очередь:

ScheduledFutureTask.run

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

Представьте, что мы ранее отправили задачу, используя scheduleAtFixedRate , задача будет считаться periodi c и будет вызван метод runPeriodi c:

ScheduledFutureTask.runPeriodi c

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();
}

Это дает вам представление о том, как происходят маги c. Большая часть работы «сторожевого пса» происходит внутри потока, они управляют им сами. Задача исполнителя - убедиться, что его очередь всегда пуста, а также диспетчеризация и создание задач по потокам. Затем поведение задач обрабатывается самостоятельно путем прямого доступа к Executor, благодаря возможности вложенного класса Java.