Java 类org.springframework.core.task.TaskExecutor 实例源码

@Bean("mvcTaskexecutor")
  public TaskExecutor getAsyncExecutor() {
        ConcurrentTaskExecutor executor = new ConcurrentTaskExecutor(
                 Executors.newFixedThreadPool(100));

        executor.setTaskDecorator(new TaskDecorator() {
           @Override
           public Runnable decorate (Runnable runnable) {
               return () -> {

                   long t = System.currentTimeMillis();
                   runnable.run();
                   System.out.printf("Thread %s has a processing time:  %s%n", Thread.currentThread().getName(),       
                                     (System.currentTimeMillis() - t));
               };
           }
       });
        return executor;
}

/**
 * Manages kafka consumers running in a background processing thread for websocket consumers.
 * @param webKafkaConsumerFactory Factory for creating new Consumers
 * @param messagingTemplate messaging template instance for passing websocket messages.
 * @param backgroundConsumerExecutor The executor to run our manager in.
 * @param appProperties defined app properties.
 * @return manager instance for web socket consumers.
 */
@Bean
public WebSocketConsumersManager getWebSocketConsumersManager(
    final WebKafkaConsumerFactory webKafkaConsumerFactory,
    final SimpMessagingTemplate messagingTemplate,
    final TaskExecutor backgroundConsumerExecutor,
    final AppProperties appProperties) {

    // Create manager
    final WebSocketConsumersManager manager = new WebSocketConsumersManager(
        webKafkaConsumerFactory,
        messagingTemplate,
        appProperties.getMaxConcurrentWebSocketConsumers()
    );

    // Submit to executor service
    backgroundConsumerExecutor.execute(manager);

    return manager;
}

private TaskExecutor createDefaultTaskExecutor() {
    // create thread-pool for starting contexts
    ThreadGroup threadGroup =
            new ThreadGroup("eclipse-gemini-blueprint-extender[" + ObjectUtils.getIdentityHexString(this) + "]-threads");
    threadGroup.setDaemon(false);

    ThreadPoolTaskExecutor taskExecutor = new ThreadPoolTaskExecutor();
    taskExecutor.setMaxPoolSize(Runtime.getRuntime().availableProcessors());
    taskExecutor.setThreadGroup(threadGroup);
    taskExecutor.setThreadNamePrefix("EclipseGeminiBlueprintExtenderThread-");
    taskExecutor.initialize();

    isTaskExecutorManagedInternally = true;

    return taskExecutor;
}

@Override
public void afterPropertiesSet() throws Exception {
    BeanWrapper bw = new BeanWrapperImpl(ThreadPoolTaskExecutor.class);
    determinePoolSizeRange(bw);
    if (this.queueCapacity != null) {
        bw.setPropertyValue("queueCapacity", this.queueCapacity);
    }
    if (this.keepAliveSeconds != null) {
        bw.setPropertyValue("keepAliveSeconds", this.keepAliveSeconds);
    }
    if (this.rejectedExecutionHandler != null) {
        bw.setPropertyValue("rejectedExecutionHandler", this.rejectedExecutionHandler);
    }
    if (this.beanName != null) {
        bw.setPropertyValue("threadNamePrefix", this.beanName + "-");
    }
    this.target = (TaskExecutor) bw.getWrappedInstance();
    if (this.target instanceof InitializingBean) {
        ((InitializingBean) this.target).afterPropertiesSet();
    }
}

public static <T> DeferredResult<T> requestAsync(final TaskExecutor executor,
    final Callable<T> action) {
    final DeferredResult<T> result = new DeferredResult<>();

    final Runnable beforeCallable = () -> {
        try {
            T t = action.call();

            if (result.isSetOrExpired()) {
                log.error("async request expired");
                return;
            }

            result.setResult(t);
        } catch (final Exception ex) {
            result.setErrorResult(ex);
        }
    };

    executor.execute(beforeCallable);
    return result;
}

@Override
public void afterPropertiesSet() throws Exception {
    BeanWrapper bw = new BeanWrapperImpl(ThreadPoolTaskExecutor.class);
    determinePoolSizeRange(bw);
    if (this.queueCapacity != null) {
        bw.setPropertyValue("queueCapacity", this.queueCapacity);
    }
    if (this.keepAliveSeconds != null) {
        bw.setPropertyValue("keepAliveSeconds", this.keepAliveSeconds);
    }
    if (this.rejectedExecutionHandler != null) {
        bw.setPropertyValue("rejectedExecutionHandler", this.rejectedExecutionHandler);
    }
    if (this.beanName != null) {
        bw.setPropertyValue("threadNamePrefix", this.beanName + "-");
    }
    this.target = (TaskExecutor) bw.getWrappedInstance();
    if (this.target instanceof InitializingBean) {
        ((InitializingBean) this.target).afterPropertiesSet();
    }
}

@Override
public void afterPropertiesSet() throws Exception {
    BeanWrapper bw = new BeanWrapperImpl(ThreadPoolTaskExecutor.class);
    determinePoolSizeRange(bw);
    if (this.queueCapacity != null) {
        bw.setPropertyValue("queueCapacity", this.queueCapacity);
    }
    if (this.keepAliveSeconds != null) {
        bw.setPropertyValue("keepAliveSeconds", this.keepAliveSeconds);
    }
    if (this.rejectedExecutionHandler != null) {
        bw.setPropertyValue("rejectedExecutionHandler", this.rejectedExecutionHandler);
    }
    if (this.beanName != null) {
        bw.setPropertyValue("threadNamePrefix", this.beanName + "-");
    }
    this.target = (TaskExecutor) bw.getWrappedInstance();
    if (this.target instanceof InitializingBean) {
        ((InitializingBean) this.target).afterPropertiesSet();
    }
}

@Bean(name = "teamProcessorThreadPool")
public TaskExecutor getTeamProcessorThreadPool(Environment applicationProperties) {
  ThreadPoolTaskExecutor taskExecutor = new ThreadPoolTaskExecutor();

  taskExecutor.setAllowCoreThreadTimeOut(false);

  taskExecutor
      .setCorePoolSize(
          Integer.valueOf(applicationProperties.getProperty("teams.threads.size.core")));
  taskExecutor.setMaxPoolSize(
      Integer.valueOf(applicationProperties.getProperty("teams.threads.size.maxpool")));
  taskExecutor.setQueueCapacity(
      Integer.valueOf(applicationProperties.getProperty("teams.threads.queue.capacity")));
  taskExecutor.setRejectedExecutionHandler(new ThreadPoolExecutor.CallerRunsPolicy());
  taskExecutor.setKeepAliveSeconds(5);
  taskExecutor.setWaitForTasksToCompleteOnShutdown(true);
  taskExecutor.setThreadNamePrefix("teamTP-");

  return taskExecutor;
}

@Bean(name = "leagueProcessorThreadPool")
public TaskExecutor getLeagueProcessorThreadPool(Environment applicationProperties) {
  ThreadPoolTaskExecutor taskExecutor = new ThreadPoolTaskExecutor();

  taskExecutor.setAllowCoreThreadTimeOut(false);

  taskExecutor
      .setCorePoolSize(
          Integer.valueOf(applicationProperties.getProperty("leagues.threads.size.core")));
  taskExecutor.setMaxPoolSize(
      Integer.valueOf(applicationProperties.getProperty("leagues.threads.size.maxpool")));
  taskExecutor.setQueueCapacity(
      Integer.valueOf(applicationProperties.getProperty("leagues.threads.queue.capacity")));
  taskExecutor.setRejectedExecutionHandler(new ThreadPoolExecutor.CallerRunsPolicy());
  taskExecutor.setKeepAliveSeconds(5);
  taskExecutor.setWaitForTasksToCompleteOnShutdown(true);
  taskExecutor.setThreadNamePrefix("leagueTP-");

  return taskExecutor;
}

@Bean(name = "personProcessorThreadPool")
public TaskExecutor getPersonProcessorThreadPool(Environment applicationProperties) {
  ThreadPoolTaskExecutor taskExecutor = new ThreadPoolTaskExecutor();

  taskExecutor.setAllowCoreThreadTimeOut(false);

  taskExecutor
      .setCorePoolSize(
          Integer.valueOf(applicationProperties.getProperty("persons.threads.size.core")));
  taskExecutor.setMaxPoolSize(
      Integer.valueOf(applicationProperties.getProperty("persons.threads.size.maxpool")));
  taskExecutor.setQueueCapacity(
      Integer.valueOf(applicationProperties.getProperty("persons.threads.queue.capacity")));
  taskExecutor.setRejectedExecutionHandler(new ThreadPoolExecutor.CallerRunsPolicy());
  taskExecutor.setKeepAliveSeconds(5);
  taskExecutor.setWaitForTasksToCompleteOnShutdown(true);
  taskExecutor.setThreadNamePrefix("personTP-");

  return taskExecutor;
}

public ReverseProxyFilter(
        CharonProperties charon,
        RetryOperations retryOperations,
        RetryOperations defaultRetryOperations,
        RequestDataExtractor extractor,
        MappingsProvider mappingsProvider,
        TaskExecutor taskExecutor,
        RequestForwarder requestForwarder,
        TraceInterceptor traceInterceptor
) {
    this.charon = charon;
    this.retryOperations = retryOperations;
    this.defaultRetryOperations = defaultRetryOperations;
    this.extractor = extractor;
    this.mappingsProvider = mappingsProvider;
    this.taskExecutor = taskExecutor;
    this.requestForwarder = requestForwarder;
    this.traceInterceptor = traceInterceptor;
}

@Autowired
public MFEventBus(final TaskExecutor taskExecutor)
{
    super();

    eventBus = new AsyncEventBus(taskExecutor, new SubscriberExceptionHandler()
    {

        @Override
        public void handleException(final Throwable exception, final SubscriberExceptionContext context)
        {
            onEventBusException(exception, context);

        }
    });

    eventBus.register(this);
}

@Bean(name = CAMUNDA_TASK_EXECUTOR_QUALIFIER)
@ConditionalOnMissingBean(name = CAMUNDA_TASK_EXECUTOR_QUALIFIER)
@ConditionalOnProperty(prefix = "camunda.bpm.job-execution", name = "enabled", havingValue = "true", matchIfMissing = true)
public static TaskExecutor camundaTaskExecutor(CamundaBpmProperties properties) {
  int corePoolSize = properties.getJobExecution().getCorePoolSize();
  int maxPoolSize = properties.getJobExecution().getMaxPoolSize();

  final ThreadPoolTaskExecutor threadPoolTaskExecutor = new ThreadPoolTaskExecutor();

  threadPoolTaskExecutor.setCorePoolSize(corePoolSize);
  threadPoolTaskExecutor.setMaxPoolSize(maxPoolSize);


  LOG.configureJobExecutorPool(corePoolSize, maxPoolSize);
  return threadPoolTaskExecutor;
}

/**
 * Ensure that the Activiti's thread pool uses the correct configuration value.
 * 
 * This assertion is limited in that the configuration values must be set before Spring application context is initialized, which we cannot control easily
 * in unit test.
 */
@Test
public void testActivitiThreadPoolUsesConfiguredValues()
{
    AsyncExecutor asyncExecutor = processEngineConfiguration.getAsyncExecutor();
    SpringAsyncExecutor springAsyncExecutor = (SpringAsyncExecutor) asyncExecutor;
    TaskExecutor taskExecutor = springAsyncExecutor.getTaskExecutor();
    ThreadPoolTaskExecutor threadPoolTaskExecutor = (ThreadPoolTaskExecutor) taskExecutor;

    Integer corePoolSize = threadPoolTaskExecutor.getCorePoolSize();
    Integer maxPoolSize = threadPoolTaskExecutor.getMaxPoolSize();
    Integer keepAliveSeconds = threadPoolTaskExecutor.getKeepAliveSeconds();
    // No real easy way of getting the queue capacity from the already constructed thread pool
    Integer remainingCapacity = ((LinkedBlockingQueue<?>) threadPoolTaskExecutor.getThreadPoolExecutor().getQueue()).remainingCapacity();

    assertEquals(configurationHelper.getProperty(ConfigurationValue.ACTIVITI_THREAD_POOL_CORE_POOL_SIZE, Integer.class), corePoolSize);
    assertEquals(configurationHelper.getProperty(ConfigurationValue.ACTIVITI_THREAD_POOL_MAX_POOL_SIZE, Integer.class), maxPoolSize);
    assertEquals(configurationHelper.getProperty(ConfigurationValue.ACTIVITI_THREAD_POOL_KEEP_ALIVE_SECS, Integer.class), keepAliveSeconds);
    assertEquals(configurationHelper.getProperty(ConfigurationValue.ACTIVITI_THREAD_POOL_QUEUE_CAPACITY, Integer.class), remainingCapacity);
}

@Autowired
public LCServiceInstanceService(
        final CopyProvider copyProvider,
        final DataProvider dataProvider,
        @Value("#{environment.SOURCE_INSTANCE_ID}") final String sourceInstanceId,
        final BrokerActionRepository brokerRepo,
        final LCServiceInstanceManager instanceManager,
        final TaskExecutor executor,
        final DataProviderService dataProviderService) {
    this.copyProvider = copyProvider;
    this.dataProvider = dataProvider;
    this.sourceInstanceId = sourceInstanceId;
    this.brokerRepo = brokerRepo;
    this.instanceManager = instanceManager;
    this.executor = executor;
    this.dataProviderService = dataProviderService;
}

@Bean
    public TaskExecutor taskExecutor() {
        //https://jira.spring.io/browse/BATCH-2269
        final SimpleAsyncTaskExecutor simpleAsyncTaskExecutor = new SimpleAsyncTaskExecutor() {
            @Override
            protected void doExecute(Runnable task) {
                //gets the jobExecution of the configuration thread
                final JobExecution jobExecution = JobSynchronizationManager.getContext().getJobExecution();
                super.doExecute(() -> {
                    JobSynchronizationManager.register(jobExecution);
                    try {
                        task.run();
                    } finally {
//                        JobSynchronizationManager.release();
                        JobSynchronizationManager.close();
                    }
                });
            }
        };
        simpleAsyncTaskExecutor.setConcurrencyLimit(20);
        return simpleAsyncTaskExecutor;
    }

private JobExecution launchImmediately(TaskExecutor taskExecutor, Job job, JobParameters jobParameters)
{
    try
    {
        SimpleJobLauncher launcher = new SimpleJobLauncher();
        launcher.setJobRepository(jobRepository);
        launcher.setTaskExecutor(taskExecutor);
        return launcher.run(job, jobParameters);
    }
    catch (JobExecutionAlreadyRunningException | JobRestartException | JobInstanceAlreadyCompleteException
            | JobParametersInvalidException e)
    {
        logger.error("Unexpected exception", e);
        throw YonaException.unexpected(e);
    }
}

@Test
public void testGetAsyncExecutor() throws Exception {
  AppConfig config = new AppConfig();
  TaskExecutor asyncExecutor = config.getAsyncExecutor();

  AtomicBoolean value = new AtomicBoolean(false);

  asyncExecutor.execute(() -> {
    try {
      Thread.sleep(250);
    } catch (InterruptedException e) {
    }

    value.set(true);
  });

  Assert.assertFalse(value.get());
}

/**
*  Resumes the execution of this task. The task must be suspended and the method it runs
       *  must be suspendable or exceptions will be thrown.
 *
 *  Note that "resuming" in this case menas only to post a request for resumption by setting the status to ACTIVE.
       *  It is up to the method's compute implementation to check for this status and resume when it is detected.
       *
       *  <p>This method may also wait for the task to resume, but this may lead to the calling thread being
       *  permanently locked if the compute method never finishes.
       *
       *  Note that when second thread calls resume() on the same object, it will have to wait for the first thread to complete the call to this method.
       *
       * @param t executor of this task
       * @throws InvalidTaskStateException thrown when the task status doesn't allow resuming or the method
       *              isn't suspendable
       */
public void resume(TaskExecutor t) throws InvalidTaskStateException {
    synchronized (this) {

        if (!(method instanceof SuspendableMethod) || !status.isResumable()) {
            throw new InvalidTaskStateException("error.taskNotSuspendable");
        }
        status = TaskStatus.ACTIVE;
        info.setResumeTime(new Date());
        fireTaskResumed();
        notifyAll();
        if (t == null) {
            (new Thread(this)).start();
        } else {
            t.execute(this);
        }
        debug("Task resumed");
    }
}

public void afterPropertiesSet() throws Exception {
    Class<?> executorClass = (shouldUseBackport() ?
            ClassUtils.forName("org.springframework.scheduling.backportconcurrent.ThreadPoolTaskExecutor", getClass().getClassLoader()) :
            ThreadPoolTaskExecutor.class);
    BeanWrapper bw = new BeanWrapperImpl(executorClass);
    determinePoolSizeRange(bw);
    if (this.queueCapacity != null) {
        bw.setPropertyValue("queueCapacity", this.queueCapacity);
    }
    if (this.keepAliveSeconds != null) {
        bw.setPropertyValue("keepAliveSeconds", this.keepAliveSeconds);
    }
    if (this.rejectedExecutionHandler != null) {
        bw.setPropertyValue("rejectedExecutionHandler", this.rejectedExecutionHandler);
    }
    if (this.beanName != null) {
        bw.setPropertyValue("threadNamePrefix", this.beanName + "-");
    }
    this.target = (TaskExecutor) bw.getWrappedInstance();
    if (this.target instanceof InitializingBean) {
        ((InitializingBean) this.target).afterPropertiesSet();
    }
}

public void copyActivityFeed( final EntityRef connectingEntity, final EntityRef connectedEntityRef )
        throws Exception {
    if (logger.isTraceEnabled()) {
        logger.trace("Copying activities to feed...");
    }
    TaskExecutor taskExecutor = ( TaskExecutor ) getApplicationContext().getBean( "taskExecutor" );
    taskExecutor.execute( new Runnable() {
        @Override
        public void run() {
            try {
                em.copyRelationships( connectedEntityRef, "activities", connectingEntity, "feed" );
            }
            catch ( Exception e ) {
                logger.error( "Error while copying activities into feed", e );
            }
        }
    } );
}

/**
 * This thread runs the WebSocketConsumerManager, which manages any consumers for web sockets.
 * It only needs a single thread, because the manager starts up its own managed thread pool.
 * @return new ThreadPool Task executor.
 */
@Bean
public TaskExecutor backgroundConsumerExecutor() {
    final ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor();
    // Only a single thread in the pool
    executor.setCorePoolSize(1);
    executor.setMaxPoolSize(1);
    executor.setThreadNamePrefix("Web Socket Consumer Manager");
    executor.initialize();

    return executor;
}

private TaskExecutor createDefaultShutdownTaskExecutor() {
    ThreadPoolTaskExecutor taskExecutor = new ThreadPoolTaskExecutor();
    taskExecutor.setThreadNamePrefix("Gemini Blueprint context shutdown thread ");
    taskExecutor.setDaemon(true);
    taskExecutor.setMaxPoolSize(1);
    taskExecutor.initialize();

    isShutdownTaskExecutorManagedInternally = true;

    return taskExecutor;
}

public static boolean execute(Runnable task, long waitTime, TaskExecutor taskExecutor) {
    Assert.notNull(task);

    Counter counter = new Counter("counter for task: " + task);
    Runnable wrapper = new MonitoredRunnable(task, counter);

    boolean internallyManaged = false;

    if (taskExecutor == null) {
        taskExecutor = new SimpleTaskExecutor();
        internallyManaged = true;
    }

    counter.increment();

    taskExecutor.execute(wrapper);

    if (counter.waitForZero(waitTime)) {
        log.error(task + " did not finish in " + waitTime
                + "ms; consider taking a snapshot and then shutdown the VM in case the thread still hangs");

        //log.error("Current Thread dump***\n" + ThreadDump.dumpThreads());

        if (internallyManaged) {
            try {
                ((DisposableBean) taskExecutor).destroy();
            } catch (Exception e) {
                log.error("Exception thrown while destroying internally managed thread executor", e);
            }
        }
        return true;
    }

    return false;
}

@Bean
public TaskExecutor taskExecutor() {
    ThreadPoolTaskExecutor result = new ThreadPoolTaskExecutor();
    result.setMaxPoolSize(1);
    result.setCorePoolSize(1);
    return result;
}

/**
 * Create the task executor which will be used for multi-threading
 *
 * @return TaskExecutor
 */
@Bean
public TaskExecutor taskExecutor() {
    SimpleAsyncTaskExecutor asyncTaskExecutor = new SimpleAsyncTaskExecutor("spring_batch");
    asyncTaskExecutor.setConcurrencyLimit(SimpleAsyncTaskExecutor.NO_CONCURRENCY);
    return asyncTaskExecutor;
}

@Bean
public TaskExecutor taskExecutor() {
    final ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor();
    executor.setCorePoolSize(5);
    executor.setMaxPoolSize(10);
    executor.setQueueCapacity(25);
    return executor;
}

@Bean(name = "webServerTaskExecutor")
public TaskExecutor getWebServerTaskExecutor(@Qualifier("pollingThreadFactory") final ThreadFactory threadFactory,
                                             @Value("${webserver.thread-task-executor.pool.size}") final int corePoolSize,
                                             @Value("${webserver.thread-task-executor.pool.max-size}") final int maxPoolSize,
                                             @Value("${webserver.thread-task-executor.pool.queue-capacity}") final int queueCapacity,
                                             @Value("${webserver.thread-task-executor.pool.keep-alive-sec}") final int keepAliveSeconds) {
    final ThreadPoolTaskExecutor threadPoolTaskExecutor = new ThreadPoolTaskExecutor();
    threadPoolTaskExecutor.setCorePoolSize(corePoolSize);
    threadPoolTaskExecutor.setMaxPoolSize(maxPoolSize);
    threadPoolTaskExecutor.setQueueCapacity(queueCapacity);
    threadPoolTaskExecutor.setKeepAliveSeconds(keepAliveSeconds);
    threadPoolTaskExecutor.setRejectedExecutionHandler(new ThreadPoolExecutor.CallerRunsPolicy());
    threadPoolTaskExecutor.setThreadFactory(threadFactory);
    return threadPoolTaskExecutor;
}

@Bean(name = "jvmTaskExecutor")
public TaskExecutor getJvmTaskExecutor(@Qualifier("pollingThreadFactory") final ThreadFactory threadFactory,
                                       @Value("${jvm.thread-task-executor.pool.size}") final int corePoolSize,
                                       @Value("${jvm.thread-task-executor.pool.max-size}") final int maxPoolSize,
                                       @Value("${jvm.thread-task-executor.pool.queue-capacity}") final int queueCapacity,
                                       @Value("${jvm.thread-task-executor.pool.keep-alive-sec}") final int keepAliveSeconds) {
    final ThreadPoolTaskExecutor threadPoolTaskExecutor = new ThreadPoolTaskExecutor();
    threadPoolTaskExecutor.setCorePoolSize(corePoolSize);
    threadPoolTaskExecutor.setMaxPoolSize(maxPoolSize);
    threadPoolTaskExecutor.setQueueCapacity(queueCapacity);
    threadPoolTaskExecutor.setKeepAliveSeconds(keepAliveSeconds);
    threadPoolTaskExecutor.setRejectedExecutionHandler(new ThreadPoolExecutor.CallerRunsPolicy());
    threadPoolTaskExecutor.setThreadFactory(threadFactory);
    return threadPoolTaskExecutor;
}

@Bean
  @ConditionalOnProperty(prefix = "camunda.bpm.job-execution", name = "enabled", havingValue = "true", matchIfMissing = true)  
  public static JobExecutor jobExecutor(@Qualifier(JobConfiguration.CAMUNDA_TASK_EXECUTOR_QUALIFIER) final TaskExecutor taskExecutor) {
    final SpringJobExecutor springJobExecutor = new SpringJobExecutor();
    springJobExecutor.setTaskExecutor(taskExecutor);
    springJobExecutor.setRejectedJobsHandler(new CallerRunsRejectedJobsHandler());

    springJobExecutor.setWaitTimeInMillis(10);
//    springJobExecutor.setWaitIncreaseFactor(1.0f);
    springJobExecutor.setMaxWait(20);

    return springJobExecutor;
  }

@Bean
public TaskExecutor kafkaMsgExecutor() {
  ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor();
  int corePoolSize = StringUtils.isBlank(environment.getProperty(EXECUTOR_SET_CORE_POOL_SIZE)) ? 5 : Integer.parseInt(environment.getProperty(EXECUTOR_SET_CORE_POOL_SIZE));
  int maxPoolSize = StringUtils.isBlank(environment.getProperty(EXECUTOR_SET_MAX_POOL_SIZE)) ? 10 : Integer.parseInt(environment.getProperty(EXECUTOR_SET_MAX_POOL_SIZE));
  int queueCapacity = StringUtils.isBlank(environment.getProperty(EXECUTOR_SET_QUEUE_CAPACITY)) ? 25 : Integer.parseInt(environment.getProperty(EXECUTOR_SET_QUEUE_CAPACITY));
  executor.setCorePoolSize(corePoolSize);
  executor.setMaxPoolSize(maxPoolSize);
  executor.setQueueCapacity(queueCapacity);
  return executor;
}

@Autowired
public AsyncCartController(final TaskExecutor executor, final AddItem addItem, final FindAll findAll, final FindOne findOne) {
    this.executor = executor;
    this.addItem = addItem;
    this.findAll = findAll;
    this.findOne = findOne;
}

@Bean
public TaskExecutor asyncTaskExecutor() {
    final ThreadPoolTaskExecutor pool = new ThreadPoolTaskExecutor();
    pool.setCorePoolSize(threadPoolConfig.getSize());
    pool.setThreadGroupName(threadPoolConfig.getName());
    return pool;
}

/**
 * A multicast (async) event publisher to replace the synchronous one used by Spring via the ApplicationContext.
 *
 * @param taskExecutor The task executor to use
 * @return The application event multicaster to use
 */
@Bean
public ApplicationEventMulticaster asyncEventMulticaster(final TaskExecutor taskExecutor) {
    final SimpleApplicationEventMulticaster applicationEventMulticaster = new SimpleApplicationEventMulticaster();
    applicationEventMulticaster.setTaskExecutor(taskExecutor);
    return applicationEventMulticaster;
}

private static JobLauncher createJobLauncher(JobRepository jobRepository,
                                             TaskExecutor taskExecutor) throws Exception {
    SimpleJobLauncher jobLauncher = new SimpleJobLauncher();
    jobLauncher.setJobRepository(jobRepository);
    jobLauncher.setTaskExecutor(taskExecutor);
    jobLauncher.afterPropertiesSet();
    return jobLauncher;
}

@Test
public void sendWithExecutor() throws Exception {
    BeforeHandleInterceptor interceptor = new BeforeHandleInterceptor();
    TaskExecutor executor = mock(TaskExecutor.class);
    ExecutorSubscribableChannel testChannel = new ExecutorSubscribableChannel(executor);
    testChannel.addInterceptor(interceptor);
    testChannel.subscribe(this.handler);
    testChannel.send(this.message);
    verify(executor).execute(this.runnableCaptor.capture());
    verify(this.handler, never()).handleMessage(this.message);
    this.runnableCaptor.getValue().run();
    verify(this.handler).handleMessage(this.message);
    assertEquals(1, interceptor.getCounter().get());
    assertTrue(interceptor.wasAfterHandledInvoked());
}

@Autowired
public NFLeagueLoader(TaskExecutor leagueProcessorThreadPool,
                      SessionFactory sessionFactory,
                      SportsBallRepository repo,
                      Environment env, LeagueWorker leagueWorker) {
  this.sessionFactory = sessionFactory;
  this.repo = repo;
  this.leagueWorker = leagueWorker;
  this.numThreads = Integer.valueOf(env.getProperty("leagues.loading.threads", "1"));
  this.poolTaskExecutor = leagueProcessorThreadPool;
}

@Autowired
public NFPersonLoader(TaskExecutor personProcessorThreadPool,
                      SessionFactory sessionFactory,
                      SportsBallRepository repo,
                      Environment env, PersonWorker personWorker) {
  this.sessionFactory = sessionFactory;
  this.repo = repo;
  this.personWorker = personWorker;
  this.numThreads = Integer.valueOf(env.getProperty("persons.loading.threads", "1"));
  this.poolTaskExecutor = personProcessorThreadPool;
}

@Autowired
public NFTeamLoader(TaskExecutor teamProcessorThreadPool,
                    SessionFactory sessionFactory,
                    SportsBallRepository repo,
                    Environment env, TeamWorker submitableWorker) {
  this.sessionFactory = sessionFactory;
  this.repo = repo;
  this.submitableWorker = submitableWorker;
  this.numThreads = Integer.valueOf(env.getProperty("teams.loading.threads", "1"));
  this.poolTaskExecutor = teamProcessorThreadPool;
}

@Bean
public TaskExecutor taskExecutor() {
    ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor();
    executor.setCorePoolSize(7);
    executor.setMaxPoolSize(42);
    executor.setQueueCapacity(11);
    executor.setThreadNamePrefix("AsyncExecutor-");
    executor.initialize();
    return executor;
}