Как отключить пулы исполнителей / планировщиков задач Spring до того, как все остальные компоненты в веб-приложении будут уничтожены?

Question

В веб-приложении Spring у меня есть несколько компонентов DAO и служебного уровня. Один бин сервисного уровня имеет аннотированные методы @Async / @Scheduled. Эти методы зависят от других (autowired) bean-компонентов. Я настроил два пула потоков в XML:

<bean id="taskExecutor" class="org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor">
     <property name="corePoolSize" value="2" />
     <property name="maxPoolSize" value="5" />
     <property name="queueCapacity" value="5" />
     <property name="waitForTasksToCompleteOnShutdown" value="true" />
     <property name="rejectedExecutionHandler">
            <bean class="java.util.concurrent.ThreadPoolExecutor$CallerRunsPolicy"/>
        </property>
    </bean>

<bean id="taskScheduler" class="org.springframework.scheduling.concurrent.ThreadPoolTaskScheduler">
     <property name="poolSize" value="10" />
     <property name="waitForTasksToCompleteOnShutdown" value="true" />
     <property name="rejectedExecutionHandler">
            <bean class="java.util.concurrent.ThreadPoolExecutor$CallerRunsPolicy"/>
        </property>
    </bean>

    <task:annotation-driven executor="taskExecutor" scheduler="taskScheduler"/>

Все работает как положено. Моя проблема в том, что я не могу заставить работать чистое отключение пулов задач. Задачи работают с базой данных и файловой системой. Когда я останавливаю веб-приложение, требуется некоторое время, чтобы оно было остановлено. Это указывает на то, что свойство waitForTasksToCompleteOnShutdown работает. Тем не менее, я получаю IllegalStateExceptions в журнале, указывающий, что некоторые bean-компоненты уже уничтожены, но некоторые потоки рабочих задач все еще выполняются, и они перестают работать, потому что их зависимости уничтожены.

Существует проблема JIRA, которая может иметь отношение: SPR-5387

Мой вопрос таков: есть ли способ сказать Spring, чтобы он инициализировал bean-компоненты executor / scheduler, или есть ли способ сказать Spring, чтобы они сначала уничтожались?

Я понимаю, что разрушение происходит в обратном порядке инициализации. Поэтому последний компонент, инициированный бобом, будет уничтожен первым. Если бины пула потоков будут уничтожены первыми, все выполняющиеся в настоящее время задачи будут завершены и все равно смогут получить доступ к зависимым бинам.

Я также попытался использовать атрибут зависимости от пулов потоков, ссылающийся на мой компонент службы, который имеет аннотации @Async и @Scheduled. Похоже, что они никогда не выполняются, и я не получаю ошибки инициализации контекста. Я предполагаю, что аннотированному служебному компоненту сначала необходимо инициализировать эти пулы потоков, и если я использую «зависит от», я изменяю порядок и делаю их нефункциональными.

Answer 1

Два способа:

1. Иметь бобовое орудие ApplicationListener<ContextClosedEvent>.onApplicationEvent() будет вызван до того, как контекст и все bean-компоненты будут уничтожены.

2. У реализации бина Lifecycle или SmartLifecycle .stop() будет вызван до того, как контекст и все компоненты будут уничтожены.

В любом случае вы можете завершить выполнение задания до того, как произойдет механизм уничтожения бина.

Например:

@Component
public class ContextClosedHandler implements ApplicationListener<ContextClosedEvent> {
    @Autowired ThreadPoolTaskExecutor executor;
    @Autowired ThreadPoolTaskScheduler scheduler;

    @Override
    public void onApplicationEvent(ContextClosedEvent event) {
        scheduler.shutdown();
        executor.shutdown();
    }       
}

(Правка: исправленоподпись метода)

Answer 2

Я добавил ниже код для завершения задач, которые вы можете использовать. Вы можете изменить номера повторов.

package com.xxx.test.schedulers;

import java.util.Map;
import java.util.concurrent.TimeUnit;

import org.apache.log4j.Logger;
import org.springframework.beans.BeansException;
import org.springframework.beans.factory.config.BeanPostProcessor;
import org.springframework.context.ApplicationContext;
import org.springframework.context.ApplicationContextAware;
import org.springframework.context.ApplicationListener;
import org.springframework.context.event.ContextClosedEvent;
import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;
import org.springframework.scheduling.concurrent.ThreadPoolTaskScheduler;
import org.springframework.stereotype.Component;

import com.xxx.core.XProvLogger;

@Component
class ContextClosedHandler implements ApplicationListener<ContextClosedEvent> , ApplicationContextAware,BeanPostProcessor{


private ApplicationContext context;

public Logger logger = XProvLogger.getInstance().x;

public void onApplicationEvent(ContextClosedEvent event) {


    Map<String, ThreadPoolTaskScheduler> schedulers = context.getBeansOfType(ThreadPoolTaskScheduler.class);

    for (ThreadPoolTaskScheduler scheduler : schedulers.values()) {         
        scheduler.getScheduledExecutor().shutdown();
        try {
            scheduler.getScheduledExecutor().awaitTermination(20000, TimeUnit.MILLISECONDS);
            if(scheduler.getScheduledExecutor().isTerminated() || scheduler.getScheduledExecutor().isShutdown())
                logger.info("Scheduler "+scheduler.getThreadNamePrefix() + " has stoped");
            else{
                logger.info("Scheduler "+scheduler.getThreadNamePrefix() + " has not stoped normally and will be shut down immediately");
                scheduler.getScheduledExecutor().shutdownNow();
                logger.info("Scheduler "+scheduler.getThreadNamePrefix() + " has shut down immediately");
            }
        } catch (IllegalStateException e) {
            e.printStackTrace();
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }

    Map<String, ThreadPoolTaskExecutor> executers = context.getBeansOfType(ThreadPoolTaskExecutor.class);

    for (ThreadPoolTaskExecutor executor: executers.values()) {
        int retryCount = 0;
        while(executor.getActiveCount()>0 && ++retryCount<51){
            try {
                logger.info("Executer "+executor.getThreadNamePrefix()+" is still working with active " + executor.getActiveCount()+" work. Retry count is "+retryCount);
                Thread.sleep(1000);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
        if(!(retryCount<51))
            logger.info("Executer "+executor.getThreadNamePrefix()+" is still working.Since Retry count exceeded max value "+retryCount+", will be killed immediately");
        executor.shutdown();
        logger.info("Executer "+executor.getThreadNamePrefix()+" with active " + executor.getActiveCount()+" work has killed");
    }
}


@Override
public void setApplicationContext(ApplicationContext context)
        throws BeansException {
    this.context = context;

}


@Override
public Object postProcessAfterInitialization(Object object, String arg1)
        throws BeansException {
    return object;
}


@Override
public Object postProcessBeforeInitialization(Object object, String arg1)
        throws BeansException {
    if(object instanceof ThreadPoolTaskScheduler)
        ((ThreadPoolTaskScheduler)object).setWaitForTasksToCompleteOnShutdown(true);
    if(object instanceof ThreadPoolTaskExecutor)
        ((ThreadPoolTaskExecutor)object).setWaitForTasksToCompleteOnShutdown(true);
    return object;
}

}

Answer 3

У меня были похожие проблемы с потоками, запускаемыми в Spring bean.Эти потоки не закрывались должным образом после того, как я вызвал executor.shutdownNow () в методе @PreDestroy.Таким образом, решение для меня состояло в том, чтобы позволить потоку finsih с IO уже запускаться и не запускать больше IO, как только вызывается @PreDestroy.А вот и метод @PreDestroy.Для моего приложения ожидание в течение 1 секунды было приемлемым.

@PreDestroy
    public void beandestroy() {
        this.stopThread = true;
        if(executorService != null){
            try {
                // wait 1 second for closing all threads
                executorService.awaitTermination(1, TimeUnit.SECONDS);
            } catch (InterruptedException e) {
                Thread.currentThread().interrupt();
            }
        }
    }

Здесь я объяснил все проблемы, возникающие при попытке закрыть темы. http://programtalk.com/java/executorservice-not-shutting-down/

Answer 4

Если это будет веб-приложение, вы также можете использовать интерфейс ServletContextListener.

public class SLF4JBridgeListener implements ServletContextListener {

   @Autowired 
   ThreadPoolTaskExecutor executor;

   @Autowired 
   ThreadPoolTaskScheduler scheduler;

    @Override
    public void contextInitialized(ServletContextEvent sce) {

    }

    @Override
    public void contextDestroyed(ServletContextEvent sce) {
         scheduler.shutdown();
         executor.shutdown();     

    }

}

Answer 5

Мы можем добавить свойство «AwaitTerminationSeconds» для taskExecutor и taskScheduler, как показано ниже,

<property name="awaitTerminationSeconds" value="${taskExecutor .awaitTerminationSeconds}" />

<property name="awaitTerminationSeconds" value="${taskScheduler .awaitTerminationSeconds}" />

Документация для свойства "waitForTasksToCompleteOnShutdown" сообщает, когда вызывается shutdown

" Закрытие контейнера Spring продолжается, пока выполняются текущие задачи. Если вы хотите, чтобы этот исполнитель блокировал и ждал завершения задач, прежде чем остальная часть контейнера продолжит закрываться - например, чтобы поддерживать другие ресурсы, которые могут понадобиться вашим задачам - установите свойство «awaitTerminationSeconds» вместо или в дополнение к этому свойству."

https://docs.spring.io/spring/docs/current/javadoc-api/org/springframework/scheduling/concurrent/ExecutorConfigurationSupport.html#setWaitForTasksToCompleteOnShutdown-boolean-

Поэтому всегда рекомендуется использовать свойства waitForTasksToCompleteOnShutdown и awaitTerminationSeconds вместе. Значение awaitTerminationSeconds зависит от нашего приложения.