Created separate bean for poolexecutor.

Signed-off-by: Kai Zimmermann <kai.zimmermann@bosch-si.com>
This commit is contained in:
Kai Zimmermann
2016-06-08 11:58:59 +02:00
parent d9046bb9cc
commit 4f23fb1377
2 changed files with 18 additions and 67 deletions

View File

@@ -1,59 +0,0 @@
/**
* Copyright (c) 2015 Bosch Software Innovations GmbH and others.
*
* All rights reserved. This program and the accompanying materials
* are made available under the terms of the Eclipse Public License v1.0
* which accompanies this distribution, and is available at
* http://www.eclipse.org/legal/epl-v10.html
*/
package org.eclipse.hawkbit.autoconfigure.scheduling;
import java.util.List;
import java.util.concurrent.Executor;
import java.util.concurrent.ThreadPoolExecutor;
import org.springframework.context.annotation.Bean;
import org.springframework.security.concurrent.DelegatingSecurityContextExecutor;
import org.springframework.security.core.context.SecurityContext;
import org.springframework.security.core.context.SecurityContextHolder;
/**
* Extension for {@link DelegatingSecurityContextExecutor} to allow proper
* shutdown at {@link Bean} destruction time.
*
*/
public class CloseableDelegatingSecurityContextExecutor extends DelegatingSecurityContextExecutor {
private final ThreadPoolExecutor executor;
/**
* Creates a new {@link CloseableDelegatingSecurityContextExecutor} that
* uses the current {@link SecurityContext} from the
* {@link SecurityContextHolder} at the time the task is submitted.
*
* @param delegate
* the {@link Executor} to delegate to. Cannot be null.
*/
public CloseableDelegatingSecurityContextExecutor(final ThreadPoolExecutor delegate) {
super(delegate);
executor = delegate;
}
/**
* Initiates an orderly shutdown in which previously submitted tasks are
* executed, but no new tasks will be accepted.
*/
public void shutdown() {
executor.shutdown();
}
/**
* Initiates an immediate shutdown.
*
* @return a list of the tasks that were awaiting execution
*/
public List<Runnable> shutdownNow() {
return executor.shutdownNow();
}
}

View File

@@ -25,6 +25,7 @@ import org.springframework.core.task.TaskExecutor;
import org.springframework.scheduling.TaskScheduler; import org.springframework.scheduling.TaskScheduler;
import org.springframework.scheduling.concurrent.ConcurrentTaskExecutor; import org.springframework.scheduling.concurrent.ConcurrentTaskExecutor;
import org.springframework.scheduling.concurrent.ThreadPoolTaskScheduler; import org.springframework.scheduling.concurrent.ThreadPoolTaskScheduler;
import org.springframework.security.concurrent.DelegatingSecurityContextExecutor;
import com.google.common.util.concurrent.ThreadFactoryBuilder; import com.google.common.util.concurrent.ThreadFactoryBuilder;
@@ -42,12 +43,21 @@ public class ExecutorAutoConfiguration {
private AsyncConfigurerThreadpoolProperties asyncConfigurerProperties; private AsyncConfigurerThreadpoolProperties asyncConfigurerProperties;
/** /**
* @return ExecutorService for general purpose multi threaded operations. * @return ExecutorService with security context availability in thread
* Tries an orderly shutdown when destroyed. * execution..
*/ */
@Bean(destroyMethod = "shutdown") @Bean
@ConditionalOnMissingBean @ConditionalOnMissingBean
public Executor asyncExecutor() { public Executor asyncExecutor() {
return new DelegatingSecurityContextExecutor(threadPoolExecutor());
}
/**
* @return central ThreadPoolExecutor for general purpose multi threaded
* operations. Tries an orderly shutdown when destroyed.
*/
@Bean(destroyMethod = "shutdown")
public ThreadPoolExecutor threadPoolExecutor() {
final BlockingQueue<Runnable> blockingQueue = new ArrayBlockingQueue<>( final BlockingQueue<Runnable> blockingQueue = new ArrayBlockingQueue<>(
asyncConfigurerProperties.getQueuesize()); asyncConfigurerProperties.getQueuesize());
final ThreadPoolExecutor threadPoolExecutor = new ThreadPoolExecutor(asyncConfigurerProperties.getCorethreads(), final ThreadPoolExecutor threadPoolExecutor = new ThreadPoolExecutor(asyncConfigurerProperties.getCorethreads(),
@@ -57,21 +67,21 @@ public class ExecutorAutoConfiguration {
threadPoolExecutor.setRejectedExecutionHandler((r, executor) -> LOGGER.warn( threadPoolExecutor.setRejectedExecutionHandler((r, executor) -> LOGGER.warn(
"Reject runnable for centralExecutorService, reached limit of queue size {}", "Reject runnable for centralExecutorService, reached limit of queue size {}",
executor.getQueue().size())); executor.getQueue().size()));
return new CloseableDelegatingSecurityContextExecutor(threadPoolExecutor);
return threadPoolExecutor;
} }
/** /**
* @return the executor for UI background processes. Run immediate shutdown * @return the executor for UI background processes.
* when destroyed.
*/ */
@Bean(name = "uiExecutor", destroyMethod = "shutdownNow") @Bean(name = "uiExecutor")
@ConditionalOnMissingBean(name = "uiExecutor") @ConditionalOnMissingBean(name = "uiExecutor")
public Executor uiExecutor() { public Executor uiExecutor() {
final BlockingQueue<Runnable> blockingQueue = new ArrayBlockingQueue<>(20); final BlockingQueue<Runnable> blockingQueue = new ArrayBlockingQueue<>(20);
final ThreadPoolExecutor threadPoolExecutor = new ThreadPoolExecutor(1, 20, 10000, TimeUnit.MILLISECONDS, final ThreadPoolExecutor threadPoolExecutor = new ThreadPoolExecutor(1, 20, 10000, TimeUnit.MILLISECONDS,
blockingQueue, new ThreadFactoryBuilder().setNameFormat("ui-executor-pool-%d").build()); blockingQueue, new ThreadFactoryBuilder().setNameFormat("ui-executor-pool-%d").build());
threadPoolExecutor.setRejectedExecutionHandler(new ThreadPoolExecutor.CallerRunsPolicy()); threadPoolExecutor.setRejectedExecutionHandler(new ThreadPoolExecutor.CallerRunsPolicy());
return new CloseableDelegatingSecurityContextExecutor(threadPoolExecutor); return new DelegatingSecurityContextExecutor(threadPoolExecutor);
} }
/** /**