【UseThreadPoolExecutor1.class】
package com.jxb.thread13;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.ThreadPoolExecutor.DiscardOldestPolicy;
/*
* 自定義線程池 ThreadPoolExecutor
*/
public class UseThreadPoolExecutor1 {
public static void main(String[] args) {
/**
* 在使用有界隊列時，若有新的任務需要執(zhí)行，如果線程池實際線程數(shù)小于corePoolSize，則優(yōu)先創(chuàng)建線程，
* 若大于corePoolSize，則會將任務加入隊列，
* 若隊列已滿，則在總線程數(shù)不大于maximumPoolSize的前提下，創(chuàng)建新的線程，
* 若線程數(shù)大于maximumPoolSize，則執(zhí)行拒絕策略。或其他自定義方式。
*
*/
ThreadPoolExecutor pool = new ThreadPoolExecutor(
1, //coreSize 核心線程
2, //MaxSize 最大線程數(shù)
60, //60
TimeUnit.SECONDS,
new ArrayBlockingQueue<Runnable>(3) //指定一種有界隊列 （有界隊列）3個
//new LinkedBlockingQueue<Runnable>()
//, new MyRejected() //自定義拒絕策略
//, new DiscardOldestPolicy() //丟棄最老的請求(隊列里面的)，嘗試再次提交當前任務。
);
MyTask mt1 = new MyTask(1, "任務1");
MyTask mt2 = new MyTask(2, "任務2");
MyTask mt3 = new MyTask(3, "任務3");
MyTask mt4 = new MyTask(4, "任務4");
MyTask mt5 = new MyTask(5, "任務5");
MyTask mt6 = new MyTask(6, "任務6");
pool.execute(mt1); //當啟動一個時，直接打印
pool.execute(mt2); //當啟動兩個時，放一個到queue里面，每個等五秒打印
pool.execute(mt3); //當啟動三個時，放兩個到queue里面，每個等五秒打印
pool.execute(mt4); //當啟動四個時，放三個到queue里面，每個等五秒打印
pool.execute(mt5); //當啟動五個時，本來應該放四個到queue里面，但是queue的容量為3，所以重新創(chuàng)建一個線程，先打印1 5，其他等五秒再打印
pool.execute(mt6); //當啟動六個時，本來應該放五個到queue里面，但是queue的容量為3，所以再重新創(chuàng)建一個線程，但是最大線程數(shù)為2，再創(chuàng)建就超出了，所以報錯
pool.shutdown(); //等線程任務執(zhí)行完畢后才關閉
}
}
【UseThreadPoolExecutor1.class 的運行結果】
這里運行結果有多種，現(xiàn)在就展現(xiàn)兩種：
1、六個任務同時都啟動，使用默認的拒絕策略
run taskId =1
run taskId =5
Exception in thread "main"
java.util.concurrent.RejectedExecutionException: Task 6 rejected from java.util.concurrent.ThreadPoolExecutor@70dea4e[Running, pool size = 2, active threads = 2, queued tasks = 3, completed tasks = 0]
at java.util.concurrent.ThreadPoolExecutor$AbortPolicy.rejectedExecution(Unknown Source)
at java.util.concurrent.ThreadPoolExecutor.reject(Unknown Source)
at java.util.concurrent.ThreadPoolExecutor.execute(Unknown Source)
at com.jxb.thread13.UseThreadPoolExecutor1.main(UseThreadPoolExecutor1.java:43)
run taskId =2
run taskId =3
run taskId =4
2、六個任務同時都啟動，使用自定義的拒絕策略
自定義處理..
run taskId =5
run taskId =1
當前被拒絕任務為：6
run taskId =2
run taskId =3
run taskId =4
【UseThreadPoolExecutor2.class】
package com.jxb.thread13;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;
/*
* 自定義線程池 ThreadPoolExecutor
*/
public class UseThreadPoolExecutor2 implements Runnable{
private static AtomicInteger count = new AtomicInteger(0);
@Override
public void run() {
try {
int temp = count.incrementAndGet(); //i++
System.out.println("任務" + temp);
Thread.sleep(2000);
} catch (InterruptedException e) {
e.printStackTrace();
}
}
public static void main(String[] args) throws Exception{
//System.out.println(Runtime.getRuntime().availableProcessors());
/**
* 無界隊列(LinkedBlockingQueue)，中共20任務，每次執(zhí)行5個任務(看coreSize的大小)，剩下的15個任務放到queue里面；
* 當前面的五個任務執(zhí)行完了，再五個五個的執(zhí)行。
*/
BlockingQueue<Runnable> queue =
//new LinkedBlockingQueue<Runnable>();
/**
* 當換成有界隊列(ArrayBlockingQueue)時，先執(zhí)行5個，本來應該放15個到queue，但是
* queue的容量是10，所以還剩下5個；再看maxSize的大小。還未超出(20最大線程數(shù)-5核心線程)，將剩下的5個放入線程中執(zhí)行。
* 變成 先執(zhí)行10個，再從queue里面取出10個執(zhí)行.
* 【測試】 相當于這個可以執(zhí)行的總線程數(shù)為：5(核心線程)+10(隊列)+15(最大線程數(shù)-核心線程數(shù))
*/
new ArrayBlockingQueue<Runnable>(10);
ExecutorService executor = new ThreadPoolExecutor(
5, //coreSize
20, //maxSize
120L, //2fenzhong
TimeUnit.SECONDS,
queue);
//循環(huán)20次，提交了20個任務
for(int i = 0 ; i < 20; i++){
executor.execute(new UseThreadPoolExecutor2());
}
Thread.sleep(1000);
System.out.println("queue size:" + queue.size());
Thread.sleep(2000);
executor.shutdown();
}
}
【MyTask.class】
package com.jxb.thread13;
public class MyTask implements Runnable {
private int taskId;
private String taskName;
public MyTask(int taskId, String taskName){
this.taskId = taskId;
this.taskName = taskName;
}
public int getTaskId() {
return taskId;
}
public void setTaskId(int taskId) {
this.taskId = taskId;
}
public String getTaskName() {
return taskName;
}
public void setTaskName(String taskName) {
this.taskName = taskName;
}
@Override
public void run() {
try {
System.out.println("run taskId =" + this.taskId);
Thread.sleep(5*1000);
//System.out.println("end taskId =" + this.taskId);
} catch (InterruptedException e) {
e.printStackTrace();
}
}
public String toString(){
return Integer.toString(this.taskId);
}
}
【MyRejected.class】
package com.jxb.thread13;
import java.net.HttpURLConnection;
import java.util.concurrent.RejectedExecutionHandler;
import java.util.concurrent.ThreadPoolExecutor;
/**
* 自定義拒絕策略可以實現(xiàn)RejectedExecutionHandler接口，并重寫rejectedExecution(參數(shù)1，參數(shù)2)方法
* 參數(shù)1：當前任務對象
* 參數(shù)2：線程池對象
*/
public class MyRejected implements RejectedExecutionHandler{
public MyRejected(){
}
@Override
public void rejectedExecution(Runnable r, ThreadPoolExecutor executor) {
System.out.println("自定義處理..");
System.out.println("當前被拒絕任務為：" + r.toString());
/**
* 拒絕任務的時候，可以使用 HttpURLConnection發(fā)送一個請求，將拒絕的數(shù)據(jù)發(fā)送給請求方客戶端，
* 告訴他這個任務現(xiàn)在處理不了。這種解決方案不利于大量的并發(fā)操作等。
* 例如如果有1000個拒絕，難道要請求1000次嗎？
* 發(fā)送請求還可以使用：Apache HttpClient
* 【https://blog.csdn.net/u013473691/article/details/52297195】
*
* 第二種解決方案：記錄log。然后在不是高峰期的時候再使用定時任務去解析日志等
*/
}
}
