首先介紹一下ThreadPoolExecutor構造方法：

public ThreadPoolExecutor(int corePoolSize,int maximumPoolSize,long keepAliveTime,TimeUnit unit,BlockingQueue<Runnable> workQueue,ThreadFactory threadFactory,RejectedExecutionHandler handler) {if (corePoolSize < 0 ||maximumPoolSize <= 0 ||maximumPoolSize < corePoolSize ||keepAliveTime < 0)throw new IllegalArgumentException();if (workQueue == null || threadFactory == null || handler == null)throw new NullPointerException();this.corePoolSize = corePoolSize;this.maximumPoolSize = maximumPoolSize;this.workQueue = workQueue;this.keepAliveTime = unit.toNanos(keepAliveTime);this.threadFactory = threadFactory;this.handler = handler;}

構造方法中的字段含義如下

• corePoolSize：核心線程數量，當有新任務在execute()方法提交時，會執行以下判斷：
  • 如果運行的線程少于 corePoolSize，則創建新線程來處理任務，即使線程池中的其他線程是空閑的；
  • 如果線程池中的線程數量大于等于 corePoolSize 且小于 maximumPoolSize，則只有當workQueue滿時才創建新的線程去處理任務；
  • 如果設置的corePoolSize 和 maximumPoolSize相同，則創建的線程池的大小是固定的，這時如果有新任務提交，若workQueue未滿，則將請求放入workQueue中，等待有空閑的線程去從workQueue中取任務并處理；
  • 如果運行的線程數量大于等于maximumPoolSize，這時如果workQueue已經滿了，則通過handler所指定的策略來處理任務
  • 所以，任務提交時，判斷的順序為 corePoolSize –> workQueue –> maximumPoolSize。
• maximumPoolSize：最大線程數量
• workQueue：等待隊列，當任務提交時，如果線程池中的線程數量大于等于corePoolSize的時候，把該任務封裝成一個Worker對象放入等待隊列；
• keepAliveTime：線程池維護線程所允許的空閑時間。當線程池中的線程數量大于corePoolSize的時候，如果這時沒有新的任務提交，核心線程外的線程不會立即銷毀，而是會等待，直到等待的時間超過了keepAliveTime；
• threadFactory：它是ThreadFactory類型的變量，用來創建新線程。默認使用Executors.defaultThreadFactory() 來創建線程。使用默認的ThreadFactory來創建線程時，會使新創建的線程具有相同的NORM_PRIORITY優先級并且是非守護線程，同時也設置了線程的名稱。
• handler：它是RejectedExecutionHandler類型的變量，表示線程池的飽和策略。如果阻塞隊列滿了并且沒有空閑的線程，這時如果繼續提交任務，就需要采取一種策略處理該任務。線程池提供了4種策略：
  • AbortPolicy：直接拋出異常，這是默認策略；
  • CallerRunsPolicy：用調用者所在的線程來執行任務；
  • DiscardOldestPolicy：丟棄阻塞隊列中靠最前的任務，并執行當前任務；
  • DiscardPolicy：直接丟棄任務；

簡單來說，在執行execute()方法時如果狀態一直是RUNNING時，的執行過程如下：

如果workerCount < corePoolSize，則創建并啟動一個線程來執行新提交的任務；

如果workerCount >= corePoolSize，且線程池內的阻塞隊列未滿，則將任務添加到該阻塞隊列中；

如果workerCount >= corePoolSize && workerCount < maximumPoolSize，且線程池內的阻塞隊列已滿，則創建并啟動一個線程來執行新提交的任務；

如果workerCount >= maximumPoolSize，并且線程池內的阻塞隊列已滿, 則根據拒絕策略來處理該任務, 默認的處理方式是直接拋異常。

實驗：拒絕策略CallerRunsPolicy

? ? 測試當拒絕策略是CallerRunsPolicy時，用調用者所在的線程來執行任務，是什么現象。

實驗環境

• ?jdk 1.8
• postman模擬并發
• mysql5.7
• intellj?
• springboot?2.1.4.RELEASE

實驗代碼

本實驗代碼，在https://github.com/vincentduan/mavenProject?下的threadManagement目錄下。

實驗步驟

1.??首先配置maven pom.xml文件內容，關鍵內容如下：

<dependencyManagement><dependencies><dependency><groupId>org.springframework.boot</groupId><artifactId>spring-boot-dependencies</artifactId><version>2.1.4.RELEASE</version><scope>import</scope><type>pom</type></dependency></dependencies></dependencyManagement><dependencies><dependency><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-web</artifactId></dependency><dependency><groupId>org.springframework.boot</groupId><artifactId>spring-boot-configuration-processor</artifactId><optional>true</optional></dependency><dependency><groupId>mysql</groupId><artifactId>mysql-connector-java</artifactId></dependency><dependency><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-jdbc</artifactId></dependency><dependency><groupId>com.alibaba</groupId><artifactId>druid</artifactId><version>1.1.6</version></dependency><dependency><groupId>org.projectlombok</groupId><artifactId>lombok</artifactId><version>1.18.6</version></dependency>

2. 配置數據庫連接池：

@SpringBootConfiguration public class DBConfiguration {@Autowiredprivate Environment environment;@Beanpublic DataSource createDataSource() {DruidDataSource druidDataSource = new DruidDataSource();druidDataSource.setUrl(environment.getProperty("spring.datasource.url"));druidDataSource.setUsername(environment.getProperty("spring.datasource.username"));druidDataSource.setPassword(environment.getProperty("spring.datasource.password"));druidDataSource.setDriverClassName(environment.getProperty("spring.datasource.driver-class-name"));return druidDataSource;}}

3. 配置線程池：

@Configuration @EnableAsync @Slf4j public class ExecutorConfig {@AutowiredVisiableThreadPoolTaskExecutor visiableThreadPoolTaskExecutor;@Beanpublic Executor asyncServiceExecutor() {log.info("start asyncServiceExecutor");ThreadPoolTaskExecutor executor = visiableThreadPoolTaskExecutor;//配置核心線程數executor.setCorePoolSize(5);//配置最大線程數executor.setMaxPoolSize(5);//配置隊列大小executor.setQueueCapacity(5);//配置線程池中的線程的名稱前綴executor.setThreadNamePrefix("async-service-");// rejection-policy：當pool已經達到max size的時候，如何處理新任務// CALLER_RUNS：不在新線程中執行任務，而是有調用者所在的線程來執行executor.setRejectedExecutionHandler(new ThreadPoolExecutor.CallerRunsPolicy());//執行初始化executor.initialize();return executor;} }

VisiableThreadPoolTaskExecutor類作為一個bean，并且繼承了ThreadPoolTaskExecutor；

4. 接下來創建一個service，為了模擬并發，我們將方法執行sleep了30秒。如下：

@Service @Slf4j public class UserThreadServiceImpl implements UserThreadService {@AutowiredUserThreadDao userThreadDao;@Override@Async("asyncServiceExecutor")public void serviceTest(String username) {log.info("開啟執行一個Service, 這個Service執行時間為30s, threadId:{}",Thread.currentThread().getId());userThreadDao.add(username, Integer.parseInt(Thread.currentThread().getId() +""), "started");try {Thread.sleep(30000);} catch (InterruptedException e) {e.printStackTrace();}log.info("執行完成一個Service, threadId:{}",Thread.currentThread().getId());userThreadDao.update(username, Integer.parseInt(Thread.currentThread().getId() +""), "ended");}@Overridepublic void update(String username, int threadId, String status) {userThreadDao.update(username, threadId, status);} }

5. 其中的dao很簡單，就是往數據庫插入數據和更新數據。

6. 創建web Controller：

@RestController @RequestMapping("/serviceTest") public class TestController {@AutowiredUserThreadService userThreadService;@Autowiredprivate VisiableThreadPoolTaskExecutor visiableThreadPoolTaskExecutor;@GetMapping("test/{username}")public Object test(@PathVariable("username") String username) {userThreadService.serviceTest(username);JSONObject jsonObject = new JSONObject();ThreadPoolExecutor threadPoolExecutor = visiableThreadPoolTaskExecutor.getThreadPoolExecutor();jsonObject.put("ThreadNamePrefix", visiableThreadPoolTaskExecutor.getThreadNamePrefix());jsonObject.put("TaskCount", threadPoolExecutor.getTaskCount());jsonObject.put("completedTaskCount", threadPoolExecutor.getCompletedTaskCount());jsonObject.put("activeCount", threadPoolExecutor.getActiveCount());jsonObject.put("queueSize", threadPoolExecutor.getQueue().size());return jsonObject;}}

執行測試

因為CorePoolSize數量是5，MaxPoolSize也是5，因此線程池的大小是固定的。而QueueCapacity數量也是5。因此當請求前5次時，返回響應：

當前線程數達到CorePoolSize時，新來的請求就會進入到workQueue中，如下所示：

而QueueCapacity的數量也是5，因此達到QueueCapacity的最大限制后，同時也達到了MaxPoolSize的最大限制，將會根據拒絕策略來處理該任務，這里的策略是CallerRunsPolicy時，用調用者所在的線程來執行任務，表現為當前頁面被阻塞住了，直到當前調用者所在的線程執行完畢。如下所示：

從控制臺的輸出也能看出CallerRunsPolicy策略執行線程是調用者線程：

2019-08-19 21:06:50.255 INFO 1302 --- [async-service-4] c.a.i.s.impl.UserThreadServiceImpl : 開啟執行一個Service, 這個Service執行時間為30s, threadId:51 2019-08-19 21:06:50.271 INFO 1302 --- [async-service-4] cn.ac.iie.dao.UserThreadDao : threadId:51, jdbcTemplate:org.springframework.jdbc.core.JdbcTemplate@5d83694c 2019-08-19 21:06:50.751 INFO 1302 --- [async-service-5] c.a.i.s.impl.UserThreadServiceImpl : 開啟執行一個Service, 這個Service執行時間為30s, threadId:52 2019-08-19 21:06:50.771 INFO 1302 --- [async-service-5] cn.ac.iie.dao.UserThreadDao : threadId:52, jdbcTemplate:org.springframework.jdbc.core.JdbcTemplate@5d83694c 2019-08-19 21:06:55.028 INFO 1302 --- [nio-8080-exec-1] c.a.i.s.impl.UserThreadServiceImpl : 開啟執行一個Service, 這個Service執行時間為30s, threadId:24 2019-08-19 21:06:55.036 INFO 1302 --- [nio-8080-exec-1] cn.ac.iie.dao.UserThreadDao : threadId:24, jdbcTemplate:org.springframework.jdbc.core.JdbcTemplate@5d83694c

其中[nio-8080-exec-1]表示就是調用者的線程。而其他線程都是[async-service-]

總結

以上是生活随笔為你收集整理的ThreadPoolExecutor里面4种拒绝策略--CallerRunsPolicy的全部內容，希望文章能夠幫你解決所遇到的問題。

如果覺得生活随笔網站內容還不錯，歡迎將生活随笔推薦給好友。