溫馨提示×
您好，登錄后才能下訂單哦！
點擊 登錄注冊 即表示同意《億速云用戶服務條款》
您好，登錄后才能下訂單哦！
這篇文章主要介紹“SpringBoot線程池和Java線程池怎么使用”，在日常操作中，相信很多人在SpringBoot線程池和Java線程池怎么使用問題上存在疑惑，小編查閱了各式資料，整理出簡單好用的操作方法，希望對大家解答”SpringBoot線程池和Java線程池怎么使用”的疑惑有所幫助！接下來，請跟著小編一起來學習吧！
public class AsyncTest {
@Async
public void async(String name) throws InterruptedException {
System.out.println("async" + name + " " + Thread.currentThread().getName());
Thread.sleep(1000);
}
}啟動類上需要添加@EnableAsync注解，否則不會生效。
@SpringBootApplication
//@EnableAsync
public class Test1Application {
public static void main(String[] args) throws InterruptedException {
ConfigurableApplicationContext run = SpringApplication.run(Test1Application.class, args);
AsyncTest bean = run.getBean(AsyncTest.class);
for(int index = 0; index <= 10; ++index){
bean.async(String.valueOf(index));
}
}
}此時可不加 @EnableAsync注解
@SpringBootTest
class Test1ApplicationTests {
@Resource
ThreadPoolTaskExecutor threadPoolTaskExecutor;
@Test
void contextLoads() {
Runnable runnable = () -> {
System.out.println(Thread.currentThread().getName());
};
for(int index = 0; index <= 10; ++index){
threadPoolTaskExecutor.submit(runnable);
}
}
}SpringBoot線程池的常見配置：
spring: task: execution: pool: core-size: 8 max-size: 16 # 默認是 Integer.MAX_VALUE keep-alive: 60s # 當線程池中的線程數量大于 corePoolSize 時，如果某線程空閑時間超過keepAliveTime，線程將被終止 allow-core-thread-timeout: true # 是否允許核心線程超時,默認true queue-capacity: 100 # 線程隊列的大小,默認Integer.MAX_VALUE shutdown: await-termination: false # 線程關閉等待 thread-name-prefix: task- # 線程名稱的前綴
TaskExecutionAutoConfiguration 類中定義了 ThreadPoolTaskExecutor,該類的內部實現也是基于java原生的 ThreadPoolExecutor類。initializeExecutor()方法在其父類中被調用，但是在父類中 RejectedExecutionHandler 被定義為了 private RejectedExecutionHandler rejectedExecutionHandler = new ThreadPoolExecutor.AbortPolicy(); ，并通過initialize()方法將AbortPolicy傳入initializeExecutor()中。
注意在TaskExecutionAutoConfiguration 類中，ThreadPoolTaskExecutor類的bean的名稱為： applicationTaskExecutor 和 taskExecutor。
// TaskExecutionAutoConfiguration#applicationTaskExecutor()
@Lazy
@Bean(name = { APPLICATION_TASK_EXECUTOR_BEAN_NAME,
AsyncAnnotationBeanPostProcessor.DEFAUL
T_TASK_EXECUTOR_BEAN_NAME })
@ConditionalOnMissingBean(Executor.class)
public ThreadPoolTaskExecutor applicationTaskExecutor(TaskExecutorBuilder builder) {
return builder.build();
}// ThreadPoolTaskExecutor#initializeExecutor()
@Override
protected ExecutorService initializeExecutor(
ThreadFactory threadFactory, RejectedExecutionHandler rejectedExecutionHandler) {
BlockingQueue<Runnable> queue = createQueue(this.queueCapacity);
ThreadPoolExecutor executor;
if (this.taskDecorator != null) {
executor = new ThreadPoolExecutor(
this.corePoolSize, this.maxPoolSize, this.keepAliveSeconds, TimeUnit.SECONDS,
queue, threadFactory, rejectedExecutionHandler) {
@Override
public void execute(Runnable command) {
Runnable decorated = taskDecorator.decorate(command);
if (decorated != command) {
decoratedTaskMap.put(decorated, command);
}
super.execute(decorated);
}
};
}
else {
executor = new ThreadPoolExecutor(
this.corePoolSize, this.maxPoolSize, this.keepAliveSeconds, TimeUnit.SECONDS,
queue, threadFactory, rejectedExecutionHandler);
}
if (this.allowCoreThreadTimeOut) {
executor.allowCoreThreadTimeOut(true);
}
this.threadPoolExecutor = executor;
return executor;
}// ExecutorConfigurationSupport#initialize()
public void initialize() {
if (logger.isInfoEnabled()) {
logger.info("Initializing ExecutorService" + (this.beanName != null ? " '" + this.beanName + "'" : ""));
}
if (!this.threadNamePrefixSet && this.beanName != null) {
setThreadNamePrefix(this.beanName + "-");
}
this.executor = initializeExecutor(this.threadFactory, this.rejectedExecutionHandler);
}覆蓋默認的 taskExecutor對象，bean的返回類型可以是ThreadPoolTaskExecutor也可以是Executor。
@Configuration
public class ThreadPoolConfiguration {
@Bean("taskExecutor")
public ThreadPoolTaskExecutor taskExecutor() {
ThreadPoolTaskExecutor taskExecutor = new ThreadPoolTaskExecutor();
//設置線程池參數信息
taskExecutor.setCorePoolSize(10);
taskExecutor.setMaxPoolSize(50);
taskExecutor.setQueueCapacity(200);
taskExecutor.setKeepAliveSeconds(60);
taskExecutor.setThreadNamePrefix("myExecutor--");
taskExecutor.setWaitForTasksToCompleteOnShutdown(true);
taskExecutor.setAwaitTerminationSeconds(60);
//修改拒絕策略為使用當前線程執行
taskExecutor.setRejectedExecutionHandler(new ThreadPoolExecutor.CallerRunsPolicy());
//初始化線程池
taskExecutor.initialize();
return taskExecutor;
}
}如果出現了多個線程池，例如再定義一個線程池 taskExecutor2，則直接執行會報錯。此時需要指定bean的名稱即可。
@Bean("taskExecutor2")
public ThreadPoolTaskExecutor taskExecutor2() {
ThreadPoolTaskExecutor taskExecutor = new ThreadPoolTaskExecutor();
//設置線程池參數信息
taskExecutor.setCorePoolSize(10);
taskExecutor.setMaxPoolSize(50);
taskExecutor.setQueueCapacity(200);
taskExecutor.setKeepAliveSeconds(60);
taskExecutor.setThreadNamePrefix("myExecutor2--");
taskExecutor.setWaitForTasksToCompleteOnShutdown(true);
taskExecutor.setAwaitTerminationSeconds(60);
//修改拒絕策略為使用當前線程執行
taskExecutor.setRejectedExecutionHandler(new ThreadPoolExecutor.CallerRunsPolicy());
//初始化線程池
taskExecutor.initialize();
return taskExecutor;
}引用線程池時，需要將變量名更改為bean的名稱，這樣會按照名稱查找。
@Resource ThreadPoolTaskExecutor taskExecutor2;
對于使用@Async注解的多線程則在注解中指定bean的名字即可。
@Async("taskExecutor2")
public void async(String name) throws InterruptedException {
System.out.println("async" + name + " " + Thread.currentThread().getName());
Thread.sleep(1000);
}線程池的四種拒絕策略
ThreadPoolExecutor 類的構造函數如下：
public ThreadPoolExecutor(int corePoolSize, int maximumPoolSize, long keepAliveTime, TimeUnit unit,
BlockingQueue<Runnable> workQueue) {
this(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue,
Executors.defaultThreadFactory(), defaultHandler);
}不限制最大線程數（maximumPoolSize=Integer.MAX_VALUE）,如果有空閑的線程超過需要，則回收，否則重用已有的線程。
new ThreadPoolExecutor(0, Integer.MAX_VALUE, 60L, TimeUnit.SECONDS, new SynchronousQueue<Runnable>());
定長線程池，超出線程數的任務會在隊列中等待。
return new ThreadPoolExecutor(nThreads, nThreads, 0L, TimeUnit.MILLISECONDS, new LinkedBlockingQueue<Runnable>());
類似于newCachedThreadPool，線程數無上限，但是可以指定corePoolSize。可實現延遲執行、周期執行。
public ScheduledThreadPoolExecutor(int corePoolSize) {
super(corePoolSize, Integer.MAX_VALUE, 0, NANOSECONDS,
new DelayedWorkQueue());
}周期執行：
ScheduledExecutorService scheduledThreadPool = Executors.newScheduledThreadPool(5);
scheduledThreadPool.scheduleAtFixedRate(()->{
System.out.println("rate");
}, 1, 1, TimeUnit.SECONDS);延時執行：
scheduledThreadPool.schedule(()->{
System.out.println("delay 3 seconds");
}, 3, TimeUnit.SECONDS);單線程線程池，可以實現線程的順序執行。
public static ExecutorService newSingleThreadExecutor() {
return new FinalizableDelegatedExecutorService
(new ThreadPoolExecutor(1, 1,
0L, TimeUnit.MILLISECONDS,
new LinkedBlockingQueue<Runnable>()));
}CallerRunsPolicy：線程池讓調用者去執行。
AbortPolicy：如果線程池拒絕了任務，直接報錯。
DiscardPolicy：如果線程池拒絕了任務，直接丟棄。
DiscardOldestPolicy：如果線程池拒絕了任務，直接將線程池中最舊的，未運行的任務丟棄，將新任務入隊。
直接在主線程中執行了run方法。
public static class CallerRunsPolicy implements RejectedExecutionHandler {
public CallerRunsPolicy() { }
public void rejectedExecution(Runnable r, ThreadPoolExecutor e) {
if (!e.isShutdown()) {
r.run();
}
}
}效果類似于：
Runnable thread = ()->{
System.out.println(Thread.currentThread().getName());
try {
Thread.sleep(0);
} catch (InterruptedException e) {
throw new RuntimeException(e);
}
};
thread.run();直接拋出RejectedExecutionException異常，并指示任務的信息，線程池的信息。、
public static class AbortPolicy implements RejectedExecutionHandler {
public AbortPolicy() { }
public void rejectedExecution(Runnable r, ThreadPoolExecutor e) {
throw new RejectedExecutionException("Task " + r.toString() +
" rejected from " +
e.toString());
}
}DiscardPolicy
什么也不做。
public static class DiscardPolicy implements RejectedExecutionHandler {
public DiscardPolicy() { }
public void rejectedExecution(Runnable r, ThreadPoolExecutor e) {
}
}DiscardOldestPolicy
e.getQueue().poll() : 取出隊列最舊的任務。
e.execute(r) : 當前任務入隊。
public static class DiscardOldestPolicy implements RejectedExecutionHandler {
public DiscardOldestPolicy() { }
public void rejectedExecution(Runnable r, ThreadPoolExecutor e) {
if (!e.isShutdown()) {
e.getQueue().poll();
e.execute(r);
}
}
}java的線程池中保存的是 java.util.concurrent.ThreadPoolExecutor.Worker 對象，該對象在 被維護在private final HashSet<Worker> workers = new HashSet<Worker>();。workQueue是保存待執行的任務的隊列，線程池中加入新的任務時，會將任務加入到workQueue隊列中。
private final class Worker
extends AbstractQueuedSynchronizer
implements Runnable
{
/**
* This class will never be serialized, but we provide a
* serialVersionUID to suppress a javac warning.
*/
private static final long serialVersionUID = 6138294804551838833L;
/** Thread this worker is running in. Null if factory fails. */
final Thread thread;
/** Initial task to run. Possibly null. */
Runnable firstTask;
/** Per-thread task counter */
volatile long completedTasks;
/**
* Creates with given first task and thread from ThreadFactory.
* @param firstTask the first task (null if none)
*/
Worker(Runnable firstTask) {
setState(-1); // inhibit interrupts until runWorker
this.firstTask = firstTask;
this.thread = getThreadFactory().newThread(this);
}
/** Delegates main run loop to outer runWorker */
public void run() {
runWorker(this);
}
// Lock methods
//
// The value 0 represents the unlocked state.
// The value 1 represents the locked state.
protected boolean isHeldExclusively() {
return getState() != 0;
}
protected boolean tryAcquire(int unused) {
if (compareAndSetState(0, 1)) {
setExclusiveOwnerThread(Thread.currentThread());
return true;
}
return false;
}
protected boolean tryRelease(int unused) {
setExclusiveOwnerThread(null);
setState(0);
return true;
}
public void lock() { acquire(1); }
public boolean tryLock() { return tryAcquire(1); }
public void unlock() { release(1); }
public boolean isLocked() { return isHeldExclusively(); }
void interruptIfStarted() {
Thread t;
if (getState() >= 0 && (t = thread) != null && !t.isInterrupted()) {
try {
t.interrupt();
} catch (SecurityException ignore) {
}
}
}
}work對象的執行依賴于 runWorker()，與我們平時寫的線程不同，該線程處在一個循環中，并不斷地從隊列中獲取新的任務執行。因此線程池中的線程才可以復用，而不是像我們平常使用的線程一樣執行完畢就結束。
final void runWorker(Worker w) {
Thread wt = Thread.currentThread();
Runnable task = w.firstTask;
w.firstTask = null;
w.unlock(); // allow interrupts
boolean completedAbruptly = true;
try {
while (task != null || (task = getTask()) != null) {
w.lock();
// If pool is stopping, ensure thread is interrupted;
// if not, ensure thread is not interrupted. This
// requires a recheck in second case to deal with
// shutdownNow race while clearing interrupt
if ((runStateAtLeast(ctl.get(), STOP) ||
(Thread.interrupted() &&
runStateAtLeast(ctl.get(), STOP))) &&
!wt.isInterrupted())
wt.interrupt();
try {
beforeExecute(wt, task);
Throwable thrown = null;
try {
task.run();
} catch (RuntimeException x) {
thrown = x; throw x;
} catch (Error x) {
thrown = x; throw x;
} catch (Throwable x) {
thrown = x; throw new Error(x);
} finally {
afterExecute(task, thrown);
}
} finally {
task = null;
w.completedTasks++;
w.unlock();
}
}
completedAbruptly = false;
} finally {
processWorkerExit(w, completedAbruptly);
}
}到此，關于“SpringBoot線程池和Java線程池怎么使用”的學習就結束了，希望能夠解決大家的疑惑。理論與實踐的搭配能更好的幫助大家學習，快去試試吧！若想繼續學習更多相關知識，請繼續關注億速云網站，小編會繼續努力為大家帶來更多實用的文章！
免責聲明：本站發布的內容（圖片、視頻和文字）以原創、轉載和分享為主，文章觀點不代表本網站立場，如果涉及侵權請聯系站長郵箱：is@yisu.com進行舉報，并提供相關證據，一經查實，將立刻刪除涉嫌侵權內容。
