【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) {
/**
* 在使用有界队列时，若有新的任务需要执行，如果线程池实际线程数小于corePoolSize，则优先创建线程，
* 若大于corePoolSize，则会将任务加入队列，
* 若队列已满，则在总线程数不大于maximumPoolSize的前提下，创建新的线程，
* 若线程数大于maximumPoolSize，则执行拒绝策略。或其他自定义方式。
*
*/
ThreadPoolExecutor pool = new ThreadPoolExecutor(
1, //coreSize 核心线程
2, //MaxSize 最大线程数
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，所以重新创建一个线程，先打印1 5，其他等五秒再打印
pool.execute(mt6); //当启动六个时，本来应该放五个到queue里面，但是queue的容量为3，所以再重新创建一个线程，但是最大线程数为2，再创建就超出了，所以报错
pool.shutdown(); //等线程任务执行完毕后才关闭
}
}
【UseThreadPoolExecutor1.class 的运行结果】
这里运行结果有多种，现在就展现两种：
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任务，每次执行5个任务(看coreSize的大小)，剩下的15个任务放到queue里面；
* 当前面的五个任务执行完了，再五个五个的执行。
*/
BlockingQueue<Runnable> queue =
//new LinkedBlockingQueue<Runnable>();
/**
* 当换成有界队列(ArrayBlockingQueue)时，先执行5个，本来应该放15个到queue，但是
* queue的容量是10，所以还剩下5个；再看maxSize的大小。还未超出(20最大线程数-5核心线程)，将剩下的5个放入线程中执行。
* 变成 先执行10个，再从queue里面取出10个执行.
* 【测试】 相当于这个可以执行的总线程数为：5(核心线程)+10(队列)+15(最大线程数-核心线程数)
*/
new ArrayBlockingQueue<Runnable>(10);
ExecutorService executor = new ThreadPoolExecutor(
5, //coreSize
20, //maxSize
120L, //2fenzhong
TimeUnit.SECONDS,
queue);
//循环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;
/**
* 自定义拒绝策略可以实现RejectedExecutionHandler接口，并重写rejectedExecution(参数1，参数2)方法
* 参数1：当前任务对象
* 参数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发送一个请求，将拒绝的数据发送给请求方客户端，
* 告诉他这个任务现在处理不了。这种解决方案不利于大量的并发操作等。
* 例如如果有1000个拒绝，难道要请求1000次吗？
* 发送请求还可以使用：Apache HttpClient
* 【https://blog.csdn.net/u013473691/article/details/52297195】
*
* 第二种解决方案：记录log。然后在不是高峰期的时候再使用定时任务去解析日志等
*/
}
}
网友评论