什么是线程池
线程池其实就是一种多线程处理形式,处理过程中可以将任务添加到队列中,然后在创建线程后自动启动这些任务。这里的线程就是Thread
,这里的任务就是实现了Runnable
或Callable
接口的实例对象;
为什么使用线程池
使用线程池最大的原因就是可以根据系统的需求
和硬件环境
灵活的控制线程的数量,且可以对所有线程进行统一的管理和控制
,从而提高系统的运行效率
,降低系统运行压力
;当然了,使用线程池的原因不仅仅只有这些,我们可以从线程池自身的优点上来进一步了解线程池的好处;
线程池有哪些优势(重点)
线程池应用场景介绍
应用场景介绍
总之
只要有并发的地方
、任务数量大或小
、每个任务执行时间长或短
的都可以使用线程池;只不过在使用线程池的时候,注意一下设置合理的线程池大小即可;
//构造方法:
public ThreadPoolExecutor(int corePoolSize, //核心线程数量
int maximumPoolSize, // 最大线程数
long keepAliveTime, // 最大空闲时间
TimeUnit unit, // 时间单位
BlockingQueue<Runnable> workQueue, // 任务队列
ThreadFactory threadFactory, // 线程工厂
RejectedExecutionHandler handler) // 饱和处理机制
核心线程数量
:当任务提交到线程池,线程池中的线程数量没有达到核心线程最大数量,就会新建一个线程;
最大线程数
:当达到核心线程最大数量时,且任务对列已经达到最大长度时,允许创建线程,创建的线程数量不大于最大线程的数量;
最大空闲时间
:存活时间,当线程没有执行任务时,允许线程空闲,当线程达到最大空闲时间时就会进行线程回收;
时间单位
:空闲的时间单位可以秒、分等
任务对列
:当线程池中的线程数达到核心线程最大数量时,新的任务被提交时,不会立即创建线程;需要先加入到队列中,只有对列中已经达到最大长度时,才会新建线程;同时新建的线程也需要满足不大于最大线程数;
线程工厂
:允许自定义创建线程
饱和处理机制
:当核心线程数、最大线程、任务队列中都已经达到最大长度时,新的任务需要采取相应的拒绝机制
new Thread
的调用了,从而允许应用程序使用特殊的线程子类、属性等等。 * <p>
* The simplest implementation of this interface is just:
* <pre> {@code
* class SimpleThreadFactory implements ThreadFactory {
* public Thread newThread(Runnable r) {
* return new Thread(r);
* }
* }}</pre>
a客户(任务)去银行(线程池)办理业务,但银行刚开始营业,窗口服务还未就位(相当于线程池中初始数量为0)
于是经理(线程池管理者)就安排1号工作人员(创建1号线程执行任务)接待a客户(创建线程);
在a客户业务还没办完时,b客户(任务)又来了,于是经理(线程池管理者)就安排2号工作人员(创建2号线程执行任务)接待b客户(又创建了一个新的线程);
假设该银行总共就2个窗口(核心线程数量是2);
紧接着在a,b客户都没有结束的情况下c客户来了,于是经理(线程池管理者)就安排c客户先生到了银行大厅的座位上(空位相当于是任务对列)等候,并告知他;如果1,2号工作人员空出,c就可以前去办理业务;
此时d客户又到了银行,(工作人员都在忙,大厅座位也满了)于是经理赶紧安排临时工(新创建的线程)大堂站着,手持Pad设备给d客户办理业务;
假如前面的业务都没有结束的时候e客户又来了,此时正式工作人员都上了,临时工也上了,座位也满了(临时工加正式员工的总数量就是最大线程数),于是经理只能按<<超出银行最大接待能力处理办法>>(饱和处理机制)拒接接待e客户;
最后,进来办业务的人少了,大厅的临时工空闲时间也超过了1个小时(最大空闲时间),经理就会让这部分空闲的员工人下班.(销毁线程)
但是为了保证银行正常工作(有一个allowCoreThreadTimeout变量控制是否允许销毁核心线程,默认false),即使正式工闲着,也不得提前下班,所以1、2号工作人员继续待着(池内保持核心线程数量);
线程池工作流程总结示意图
java内置线程池(非延迟)
ExecutorService
接口是java内置的线程池接口,java内置线程池的基本使用常用方法如下:
void shutdown()
启动一次顺序关闭,执行以前提交的任务,但不接受新任务。
List<Runnable> shutdownNow()
停止所有正在执行的任务,暂停处理正在等待的任务,并返回等待执行的任务
<T> Future<T> submit(Callable<T> task)
执行带返回值的任务,返回一个Future对象。
Future<?> submit(Runnable task)
执行Runnable任务,并返回一个表示该任务的Future。
<T> Future<T> submit(Runnable task, T result)
执行Runnable任务,返回一个表示该任务的Future。
获取ExecutorService可以利用JDK的Executors类中的静态方法,常用获取方式如下:
static ExecutorService newCachedThreadPool()
static ExecutorService newCachedThreadPool(ThreadFactory threadFactory)
static ExecutorService newFixedThreadPool(int nThreads)
static ExecutorService newFixedThreadPool(int nThreads, ThreadFactory threadFactory)
static ExecutorService newSingleThreadExecutor()
static ExecutorService newSingleThreadExecutor(ThreadFactory threadFactory)
newCachedThreadPool
: 默认使用时间60s;线程数量不做限制,优先执行任务对硬件有一定的要求
newFixedThreadPool
: 创建固定的线程池数,当线程满时,任务会进行缓存;
newSingleThreadExecutor
: 无界队列 不限制缓存的任务数量;对安全有要求,单线程
public class MyTest01 {
public static void main(String[] args) {
//1、使用工厂类获取线程池对象
//创建一个可重用固定线程数的线程池
ExecutorService es = Executors.newCachedThreadPool(
new ThreadFactory() {
int n = 1;
@Override
public Thread newThread(Runnable r) {
return new Thread(r, "线程" + n++);
}
}
);
//2、提交任务
for (int i = 1; i <= 50; i++) {
es.submit(new MyRunnable03(i));
}
}
}
class MyRunnable03 implements Runnable {
private int id;
public MyRunnable03(int id) {
this.id = id;
}
@Override
public void run() {
//获取线程的名称,打印一句话
String name = Thread.currentThread().getName();
try {
Thread.sleep(200);
} catch (InterruptedException e) {
e.printStackTrace();
}
System.out.println(name + "执行了任务.....");
}
}
public class MyTest01 {
public static void main(String[] args) {
//1、使用工厂类获取线程池对象
//创建一个可重用固定线程数的线程池
ExecutorService es = Executors.newFixedThreadPool(3,
new ThreadFactory() {
int n = 1;
@Override
public Thread newThread(Runnable r) {
return new Thread(r, "线程" + n++);
}
}
);
//2、提交任务
for (int i = 1; i <= 50; i++) {
es.submit(new MyRunnable03(i));
}
}
}
/**
* @description:
* @author: cjw
* @date: 2024/1/15 21:20
* @param:
* @return:
* 任务类:包含一个任务编号,在任务中,打印出是哪个一个线程
**/
class MyRunnable03 implements Runnable {
private int id;
public MyRunnable03(int id) {
this.id = id;
}
@Override
public void run() {
//获取线程的名称,打印一句话
String name = Thread.currentThread().getName();
try {
Thread.sleep(200);
} catch (InterruptedException e) {
e.printStackTrace();
}
System.out.println(name + " 执行了任务.....");
}
}
public class MyTest01 {
public static void main(String[] args) {
//1、使用工厂类获取线程池对象
//创建一个使用单个worker线程的Executor,且线程池中的所有线程都使用ThreadFactory创建
ExecutorService es = Executors.newSingleThreadExecutor(
new ThreadFactory() {
int n = 1;
@Override
public Thread newThread(Runnable r) {
return new Thread(r, "线程" + n++);
}
}
);
//2、提交任务
for (int i = 1; i <= 50; i++) {
es.submit(new MyRunnable03(i));
}
}
}
/**
* @description:
* @author: cjw
* @date: 2024/1/15 21:20
* @param:
* @return:
* 任务类:包含一个任务编号,在任务中,打印出是哪个一个线程
**/
class MyRunnable03 implements Runnable {
private int id;
public MyRunnable03(int id) {
this.id = id;
}
@Override
public void run() {
//获取线程的名称,打印一句话
String name = Thread.currentThread().getName();
try {
Thread.sleep(200);
} catch (InterruptedException e) {
e.printStackTrace();
}
System.out.println(name + " 执行了任务.....");
}
}
shutdown()
: 关闭线程池,仅仅是不再接受的任务,以前的任务还会继续执行shutdownNow()
:立刻关系线程池,如果线程池中还有缓存的任务,没有执行,则取消执行,并返回这些任务。public class MyTest01 {
public static void main(String[] args) {
//1、使用工厂类获取线程池对象
//创建一个使用单个worker线程的Executor,且线程池中的所有线程都使用ThreadFactory创建
ExecutorService es = Executors.newSingleThreadExecutor(
new ThreadFactory() {
int n = 1;
@Override
public Thread newThread(Runnable r) {
return new Thread(r, "线程" + n++);
}
}
);
//2、提交任务
for (int i = 1; i <= 50; i++) {
es.submit(new MyRunnable03(i));
}
//A:关闭线程池,仅仅是不再接受新的任务,以前的任务还会继续执行
//es.shutdown();
//es.submit(new MyRunnable03(23));
//B:立刻关闭线程池,如果线程池中还有缓存的任务,
// 没有执行,则取消执行,并返回这些任务
List<Runnable> runnables = es.shutdownNow();
System.out.println(es.isShutdown());
System.out.println(runnables);
es.submit(new MyRunnable03(23));
}
}
/**
* @description:
* @author: cjw
* @date: 2024/1/15 21:20
* @param:
* @return:
* 任务类:包含一个任务编号,在任务中,打印出是哪个一个线程
**/
class MyRunnable03 implements Runnable {
private int id;
public MyRunnable03(int id) {
this.id = id;
}
@Override
public void run() {
//获取线程的名称,打印一句话
String name = Thread.currentThread().getName();
try {
Thread.sleep(200);
} catch (InterruptedException e) {
e.printStackTrace();
}
System.out.println(name + " 执行了任务.....");
}
@Override
public String toString() {
return "MyRunnable03{" +
"id=" + id +
'}';
}
}
内置线程池(延迟)–ScheduledExecutorService
ScheduledExecutorService是ExecutorService的子接口,具备了延迟运行或定期执行任务的能力,常用获取方式如下:
public static ScheduledExecutorService newScheduledThreadPool(int corePoolSize)
public static ScheduledExecutorService newScheduledThreadPool(int corePoolSize, ThreadFactory threadFactory)
public static ScheduledExecutorService newSingleThreadScheduledExecutor()
public static ScheduledExecutorService newSingleThreadScheduledExecutor(ThreadFactory threadFactory)
ScheduledExecutorService常用方法如下:
public ScheduledFuture<?> schedule(Runnable command, long delay, TimeUnit unit)
public <V> ScheduledFuture<V> schedule(Callable<V> callable, long delay, TimeUnit unit)
public ScheduledFuture<?> scheduleAtFixedRate(Runnable command, long initialDelay, long period, TimeUnit unit);
public ScheduledFuture<?> scheduleWithFixedDelay(Runnable command, long initialDelay, long delay, TimeUnit unit)
schedule示例
public class ScheduleExecutorServiceDemo01 {
public static void main(String[] args) {
//1、获取一个具备延迟执行任务的线程池对象
ScheduledExecutorService es = Executors.newScheduledThreadPool(3);
//2、创建多个任务对象,提交任务,每个任务延迟2秒执行
for (int i = 1; i <= 10; i++) {
es.schedule(new MyRunnable(i), 2, TimeUnit.MICROSECONDS);
}
//over输出之后2秒执行任务
System.out.println("over");
}
}
class MyRunnable implements Runnable {
private int id;
public MyRunnable(int id) {
this.id = id;
}
@Override
public void run() {
String name = Thread.currentThread().getName();
System.out.println(name + "执行了任务:" + id);
}
}
public class ScheduleExecutorServiceDemo02 {
public static void main(String[] args) {
//1、获取一个具备延迟执行任务的线程池对象
ScheduledExecutorService es = Executors.newScheduledThreadPool(3, new ThreadFactory() {
int n = 1;
@Override
public Thread newThread(Runnable r) {
return new Thread(r, "自定义线程名:"+n++);
}
});
//2、创建多个任务对象,提交任务,每个任务延迟2秒执行
// for (int i = 1; i <= 10; i++) {
// es.schedule(new MyRunnable2(1), 2, TimeUnit.MICROSECONDS);
es.scheduleAtFixedRate(new MyRunnable2(1), 2, 5, TimeUnit.SECONDS);
// }
//over输出之后2秒执行任务
System.out.println("over");
}
}
class MyRunnable2 implements Runnable {
private int id;
public MyRunnable2(int id) {
this.id = id;
}
@Override
public void run() {
String name = Thread.currentThread().getName();
try {
Thread.sleep(1500);
} catch (InterruptedException e) {
e.printStackTrace();
}
System.out.println(name + "执行了任务:" + id);
}
}
public class ScheduleExecutorServiceDemo03 {
public static void main(String[] args) {
ScheduledExecutorService es = Executors.newScheduledThreadPool(3, new ThreadFactory() {
int n = 1;
@Override
public Thread newThread(Runnable r) {
return new Thread(r,"自定义线程:" + n++);
}
});
es.scheduleWithFixedDelay(new MyRunnable3(1), 2, 5, TimeUnit.SECONDS);
System.out.println("over");
}
}
class MyRunnable3 implements Runnable {
private int id;
public MyRunnable3(int id) {
this.id = id;
}
@Override
public void run() {
String name = Thread.currentThread().getName();
try {
Thread.sleep(1500);
} catch (InterruptedException e) {
e.printStackTrace();
}
System.out.println(name + "执行任务:" + id);
}
}
核心线程数(corePoolSize)
核心线程数的设计需要根据任务的处理时间和每秒产生的任务数量来确定,例如:执行一个任务需要0.1秒,系统百分之80的时间每秒都会产生100个任务,那么要想在1秒内处理完这100个任务,就需要10个线程,此时我们就可以设计核心线程数为10;当然实际情况不可能这么平均,所以我们一般按照8082原则设计即可,既按照百分之80的情况设计核心线程数,剩下的百分之20可以利用最大线程数处理。
任务队列长度(workQueue)
任务队列长度一般设计为:核心线程数/单个任务执行时间*2即可;例如上面的场景中,核心线程数设计为10,单个任务执行时间为0.1秒,则队列长度可以设计为200;
最大线程数(maximumPoolSize)
最大线程数的设计除了需要参照核心线程数的条件外,还需要参照系统每秒产生的最大任务数决定;例如:上述环境中,如果系统每秒最大产生的任务是1000个,那么,最大线程数=(最大任务数-任务队列长度)*单个任务执行时间;既:最大线程数=(1000-200)* 0.1 =80个
最大空闲时间
这个参数的设计完全参考系统运行环境和硬件压力设定,没有固定的参考值,用户可以根据经验和系统产生任务的时间间隔合理设置一个值即可;
实现步骤
/**
* Created by IntelliJ IDEA.
*
* @Author : cjw
* @create 2024/1/10 22:17
* 自定义线程池练习
* 任务编号:每个任务执行时间0.2s
*/
public class MyTask implements Runnable {
private int id;
//由于run重写Runnable中的方法,因此id这个属性初始化可以利用构造方法完成
@Override
public void run() {
String name = Thread.currentThread().getName();
System.out.println("线程 = " + name + "即将执行任务:" + id);
try {
Thread.sleep(200);
} catch (InterruptedException e) {
e.printStackTrace();
}
System.out.println("线程 = " + name + "完成了任务:" + id);
}
public MyTask(int id) {
this.id = id;
}
public int getId() {
return id;
}
public void setId(int id) {
this.id = id;
}
@Override
public String toString() {
return "MyTask{" +
"id=" + id +
'}';
}
}
package com.example.springboot05.threadPool;
/**
* Created by IntelliJ IDEA.
*
* @Author : cjw
* @create 2024/1/10 22:58
*
* 测试类
*
* 创建线程池对象
* 提交多个任务
*
*/
public class MyTest {
public static void main(String[] args) {
//创建线程池对象
MyThreadPool pool = new MyThreadPool(2, 4, 20);
//提交多个任务
for (int i = 0; i < 100; i++) {
//创建任务对象,并提交给线程池
MyTask myTask = new MyTask(i);
pool.submit(myTask);
}
}
}
package com.example.springboot05.threadPool;
import java.util.Collections;
import java.util.LinkedList;
import java.util.List;
import java.util.stream.Collectors;
/**
* Created by IntelliJ IDEA.
*
* @Author : cjw
* @create 2024/1/10 22:45
* 包含提交任务,执行任务的能力
*
* 成员变量
* 任务对列 集合 需要控制线程安全问题
* 当前线程数量
* 核心线程数
* 最大线程数
* 任务对列长度
*
* 成员方法
*
* 提交任务:将任务添加到集合中,需要判断是否超出了任务总长度
* 执行任务:判断当前线程的数量,决定创建核心线程数还是非核心线程数量
*
*/
public class MyThreadPool {
private List<Runnable> task = Collections.synchronizedList(new LinkedList<>());
private int num; //当前线程数量
private int corePoolSize; //核心线程数量
private int maxSize;//最大线程数量
private int workSize;//任务对列的长度
public MyThreadPool(int corePoolSize, int maxSize, int workSize) {
this.corePoolSize = corePoolSize;
this.maxSize = maxSize;
this.workSize = workSize;
}
//提交任务
public void submit(Runnable r) {
//判断当前集合中任务的数量是否超出了最大任务数量
if (task.size() >= (workSize + corePoolSize)) { //别拒绝只有超过核心线程数和任务对列中的数量
System.out.println("任务:" + r + "被丢弃啦");
} else {
task.add(r);
//执行任务
execTask(r);
}
}
private void execTask(Runnable r) {
//判断当前线程数量是否超出了核心数量
if (num <= corePoolSize) {
new MyWorker("核心线程"+num, task).start();
num++;
} else if (num > (corePoolSize + workSize) && (num - (corePoolSize + workSize)) <= maxSize) {
new MyWorker("非核心线程"+num, task).start();
num++;
} else if (num > corePoolSize) {
System.out.println("任务:" + r + "被缓存了。。。。");
num++;
}
}
}
package com.example.springboot05.threadPool;
import java.util.List;
/**
* Created by IntelliJ IDEA.
*
* @Author : cjw
* @create 2024/1/10 22:39
*
*/
public class MyWorker extends Thread{
private String name;//保存线程的名字
private List<Runnable> tasks;
public void run() {
String name = Thread.currentThread().getName();
//判断集合中是否有任务,只要有,就一直执行任务
while (tasks!= null && tasks.size()>0) {
Runnable r = tasks.remove(0);
r.run();
}
}
public MyWorker(String name, List<Runnable> tasks) {
super(name);
this.tasks = tasks;
}
}
开发中,有时需要利用线程进行一些计算,然后获取这些计算的结果,而java中的Future接口就是专门用于描述异步计算结果的,可以通过Future对象获取线程计算的结果;Future的常用方法如下:
boolean cancel(boolean mayInterruptlfRunning)
V get()
V get(long timeout,TimeUnit unit)
boolean isCancelled()
booelean isDone()
public class FutureDemo {
public static void main(String[] args) throws ExecutionException, InterruptedException {
//1、获取线程池对象
ExecutorService es = Executors.newCachedThreadPool();
//2、创建Callable类型的任务对象
Future<Integer> f = es.submit(new MyCall(1, 1));
//3、判断任务是否已经完成
boolean done = f.isDone();
System.out.println("第一次判断任务是否完成" + done);
boolean cancelled = f.isCancelled();
System.out.println("第一次判断任务是否取消" + cancelled);
Integer v = f.get();
System.out.println("任务执行的结果是:" + v);
boolean done2 = f.isDone();
System.out.println("第二次判断任务是否完成" + done2);
boolean cancelled2 = f.isCancelled();
System.out.println("第二次判断任务是否取消" + cancelled2);
}
}
class MyCall implements Callable<Integer> {
private int a;
private int b;
//通过构造方法传递两个参数
public MyCall(int a, int b) {
this.a = a;
this.b = b;
}
@Override
public Integer call() throws Exception {
return a+b;
}
}
/**
* 任务类:
* 包含了商品数量,客户名称,送商品的行为
*/
public class MyTask1 implements Runnable{
private static int things = 10; //商品
private String name; //任务名称
public MyTask1(String name) {
this.name = name;
}
@Override
public void run() {
String threadName = Thread.currentThread().getName();
System.out.println(name + "正在使用" + threadName + "参与秒杀...");
synchronized (MyTask1.class) {
try {
Thread.sleep(1000);
} catch (InterruptedException e) {
e.printStackTrace();
}
if (things > 0) {
things--;
System.out.println("任务:"+ this.name + "使用" + threadName + "秒杀商品成功");
} else {
System.out.println("任务:"+ this.name + "使用" + threadName + "秒杀商品失败");
}
}
}
}
/**
* Created by IntelliJ IDEA.
* @Author : cjw
* @create 2024/1/14 23:09
*/
public class MyTest {
public static void main(String[] args) {
//创建一个线程池对象
ThreadPoolExecutor pool = new ThreadPoolExecutor(3, 5,
1, TimeUnit.MINUTES,
new LinkedBlockingQueue<Runnable>(15),
new ThreadPoolExecutor.DiscardPolicy()
);
//循环创建任务对象
for (int i = 1; i <= 200; i++) {
pool.submit(new MyTask1(i+""));
}
//关闭线程池
pool.shutdown();
}
}
/**
* 任务类:
* 从账号取钱
*/
public class MyTask2 implements Runnable{
private static int money = 1000; //商品
private String name; //任务名称
public MyTask2(String name) {
this.name = name;
}
@Override
public void run() {
String threadName = Thread.currentThread().getName();
System.out.println(name + "正在使用" + threadName + "进行取款...");
synchronized (MyTask2.class) {
try {
Thread.sleep(1000);
} catch (InterruptedException e) {
e.printStackTrace();
}
if (money > 0 && money > 800) {
money -= 800;
System.out.println("任务:"+ this.name + "使用" + threadName + "取款成功");
} else {
System.out.println("账户余额不足" + "----余额为:" + money);
}
}
}
}
/**
* @Author : cjw
* @create 2024/1/14 23:09
*/
public class MyTest2 {
public static void main(String[] args) {
//创建一个线程池对象
ThreadPoolExecutor pool = new ThreadPoolExecutor(3, 5,
1, TimeUnit.MINUTES,
new LinkedBlockingQueue<Runnable>(5),
new ThreadFactory() {
int n = 1;
@Override
public Thread newThread(Runnable r) {
return new Thread(r, "AIM" + n++);
}
}
, new ThreadPoolExecutor.DiscardPolicy()
);
//循环创建任务对象
for (int i = 1; i <= 2; i++) {
pool.submit(new MyTask2(i+""));
}
//关闭线程池
pool.shutdown();
}
}
Executors
工厂类的静态方法,创建线程对象;Runnable
或Callable
实现类的实例对象;ExecutorService
的submit
方法或ScheduledExecutorService
的schedule
方法提交并执行线程任务异步执行结果(Future)
shutdown()
方法,关闭线程池