常用的3个线程池
//一池5个处理线程 数量创建的时候需要赋值
ExecutorService threadPool = Executors.newFixedThreadPool(5);
//一池1个处理线程
ExecutorService threadPool = Executors.newSingleThreadExecutor();
//一池N个处理线程 带缓存扩容的
ExecutorService threadPool = Executors.newCachedThreadPool();
public static ExecutorService newFixedThreadPool(int nThreads) {
return new ThreadPoolExecutor(nThreads, nThreads,
0L, TimeUnit.MILLISECONDS,
new LinkedBlockingQueue<Runnable>());
}
public static ExecutorService newSingleThreadExecutor() {
return new FinalizableDelegatedExecutorService
(new ThreadPoolExecutor(1, 1,
0L, TimeUnit.MILLISECONDS,
new LinkedBlockingQueue<Runnable>()));
}
public static ExecutorService newCachedThreadPool() {
return new ThreadPoolExecutor(0, Integer.MAX_VALUE,
60L, TimeUnit.SECONDS,
new SynchronousQueue<Runnable>());
}
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.acc = System.getSecurityManager() == null ?
null :
AccessController.getContext();
this.corePoolSize = corePoolSize;
this.maximumPoolSize = maximumPoolSize;
this.workQueue = workQueue;
this.keepAliveTime = unit.toNanos(keepAliveTime);
this.threadFactory = threadFactory;
this.handler = handler;
}
线程池的7个参数详解
1.corePoolSize:线程池中的常驻核心线程数
2.maximumPoolSize：线程池能够容纳同时执行的最大线程数，此值必须大于等于1
3.keepAliveTime：多余的空闲线程的存货时间。
当前线程池数量超过corePoolSize时，当空闲时间达到keepAliveTime值时，
多余空闲线程会被销毁直到只剩下corePoolSize个线程为止。
4.unit：keepAliveTime的单位
5.workQueue：任务队列，被提交但尚未呗执行的任务。（类似银行等待排队办理业务的人）
6.threadFactory：表示生成线程池中工作线程的线程工厂，用于创建线程一般默认的即可
7.handler：拒绝策略，表示当队列满了并且工作线程大于等于线程池的最大线程数（maximumPoolSize）
1.png
2.png
4种拒绝策略
1.AbortPolicy
new ThreadPoolExecutor.AbortPolicy();
默认拒绝策略：线程池达到最大数，队列满了，剩余的请求 全部拒绝 并且抛出RejectedExecutionException异常
如果是比较关键的业务，推荐使用此拒绝策略，这样子在系统不能承载更大的并发量的时候，能够及时的通过异常发现。
2.CallerRunsPolicy
new ThreadPoolExecutor.CallerRunsPolicy();
将任务返还给调用者线程执行
如果任务被拒绝了，则由调用线程（提交任务的线程）直接执行此任务。
3.DiscardOldestPolicy
new ThreadPoolExecutor.DiscardOldestPolicy();
丢弃队列最前面的任务，然后重新提交被拒绝的任务。（比如队列3个等待，目前后面还有5个，那就把队列3个全部丢弃，然后5个中 进去3个到队列，然后剩2个，然后队列3个再丢弃2个，把后面剩的2个继续添加到队列，前提是队列的请求都还在等待中）
此拒绝策略，是一种喜新厌旧的拒绝策略。是否要采用此种拒绝策略，还得根据实际业务是否允许丢弃老任务来认真衡量。
4.DiscardPolicy
new ThreadPoolExecutor.DiscardPolicy();
丢弃任务，但是不抛出异常。如果线程队列已满，则后续提交的任务都会被丢弃，且是静默丢弃。
使用此策略，可能会使我们无法发现系统的异常状态。建议是一些无关紧要的业务采用此策略。例如，本人的博客网站统计阅读量就是采用的这种拒绝策略。
手写线程池
import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicInteger;
/**
* 手写一个线程池demo
*/
public class ThreadPoolDemo {
public static void main(String[] args) {
int corePoolSize = 2;
int maximumPoolSize = 4;
long keepAliveTime = 0L;
TimeUnit unit = TimeUnit.SECONDS;
BlockingQueue<Runnable> workQueue = new ArrayBlockingQueue<>(3);
ThreadFactory threadFactory = new MyThreadFactory();
RejectedExecutionHandler handler = new MyHndler();
ThreadPoolExecutor threadPoolExecutor = new ThreadPoolExecutor(corePoolSize,maximumPoolSize,keepAliveTime,unit,workQueue,threadFactory,handler);
//预启动所有核心线程
threadPoolExecutor.prestartAllCoreThreads();
for (int i = 1; i <=10 ; i++) {
MyTask myTask = new MyTask(String.valueOf(i));
threadPoolExecutor.execute(myTask);
}
}
}
/**
* 自定义线程池工厂
*/
class MyThreadFactory implements ThreadFactory{
private final AtomicInteger atomicInteger = new AtomicInteger(1);
@Override
public Thread newThread(Runnable r) {
//创建线程
Thread thread = new Thread(r,"My-Thread-Pool-"+atomicInteger.getAndIncrement());
System.out.println(thread.getName()+" has bean created");
return thread;
}
}
/**
* 自定义拒绝策略
*/
class MyHndler implements RejectedExecutionHandler{
@Override
public void rejectedExecution(Runnable r, ThreadPoolExecutor executor) {
System.err.println(r.toString()+"\t rejected");
}
}
/**
* 自定义任务
*/
class MyTask implements Runnable{
private String name;
public MyTask(String name) {
this.name = name;
}
@Override
public void run() {
try {
System.out.println(this.toString()+" is running");
TimeUnit.SECONDS.sleep(3);
} catch (InterruptedException e) {
e.printStackTrace();
}
}
public String getName() {
return name;
}
@Override
public String toString() {
return "MyTask [" +
"name='" + name + '\'' +
']';
}
}
结果
My-Thread-Pool-1 has bean created
My-Thread-Pool-2 has bean created
My-Thread-Pool-3 has bean created
MyTask [name='1'] is running
MyTask [name='2'] is running
My-Thread-Pool-4 has bean created
MyTask [name='4'] is running
MyTask [name='7'] is running
MyTask [name='8'] rejected
MyTask [name='9'] rejected
MyTask [name='10'] rejected
MyTask [name='6'] is running
MyTask [name='3'] is running
MyTask [name='5'] is running
合理线程数配置（根据硬件 根据业务）
1.cpu密集型（该任务需要大量的运算，没有阻塞，CPU一直全速运行）
一般公式：CPU核数+1个线程的线程池
2.IO密集型
2.1.由于IO密集型任务线程并不是一直在执行任务，则应配置尽可能多的线程，如CPU核数*2
2.2.如果该任务需要大量的IO，即大量的阻塞（故需要多配置线程数）：
参考公式：cpu核数/（1-阻塞系数） 阻塞系统在0.8-0.9之间 取0.9
比如：8核CPU： 8/(1-0.9)=80个线程数
网友评论