进程是资源分配的最小单位,线程是CPU调度的最小单位

般来说,一个进程由程序段、数据段和进程控制块三部分组成



程序计数器
栈内存

线程基本信息

private long tid;// 线程id属性
public long getId() // 获取线程id
private String name;// 线程名称
public final String getName();//获取线程名称
public final void setName(String name);// 设置线程名称
Thread(String threadName);//构造线程名称
private int priority;//保存一个Thread线程实例的优先级
public final int getPriority();//获取线程优先级。
public final void setPriority(int priority);//设置线程优先级
// Java线程的最大优先级值为10,最小值为1,默认值为5
public static final int MIN_PRIORITY = 1;
public static final int NORM_PRIORITY = 5;
public static final int MAX_PRIORITY = 10;
private boolean daemon=false;//该属性保存Thread线程实例的守护状态,默认为false,表示是普通的用户线程,而不是守护线程
public final void setDaemon(boolean on);//将线程实例标记为守护线程或用户线程,如果参数值为true,那么将线程实例标记为守护线程
private int threadStatus;//该属性以整数的形式保存线程的状态
public Thread.State getState();//返回表示当前线程的执行状态,为新建、就绪、运行、阻塞、结束等状态中的一种。
public static enum State {
NEW, //新建
RUNNABLE, //就绪、运行
BLOCKED, //阻塞
WAITING, //等待
TIMED_WAITING, //计时等待
TERMINATED; //结束
}
public void start();//用来启动一个线程,当调用start()方法后,JVM才会开启一个新的线程来执行用户定义的线程代码逻辑,在这个过程中会为相应的线程分配需要的资源
public void run();//作为线程代码逻辑的入口方法。run()方法不是由用户程序来调用的,当调用start()方法启动一个线程之后,只要线程获得了CPU执行时间,便进入run()方法体去执行具体的用户线程代码。
public static Thread currentThread();//该方法是一个非常重要的静态方法,用于获取当前线程的Thread实例对象。

第一种方式:集成Thread
package com.Thread;
/**
* @Author shu
* @Date: 2021/11/30/ 10:59
* @Description 线程的Thread创建方式
**/
public class SimpleThread extends Thread{
@Override
public void run() {
super.run();
System.out.println("执行方法"+currentThread().getName());
}
public static void main(String[] args) {
new SimpleThread().start();
}
}
第二种方式:实现Runnable
package com.Thread;
/**
* @Author shu
* @Date: 2021/11/30/ 13:55
* @Description 简单Runnable实现方式
**/
public class SimpleRunnable implements Runnable{
@Override
public void run() {
for (int i = 0; i <10 ; i++) {
System.out.println("执行方法"+Thread.currentThread().getName()+i);
}
}
public static void main(String[] args) {
SimpleRunnable runnable = new SimpleRunnable();
new Thread(runnable).start();
//Lambda简写
Runnable runs=(()->{
for (int i = 0; i <10 ; i++) {
System.out.println("执行方法"+Thread.currentThread().getName()+i);
}
});
}
}
第三种:Callable和FutureTask创建线程
package com.Thread;
import java.util.concurrent.Callable;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.FutureTask;
/**
* @Author shu
* @Date: 2021/11/30/ 14:00
* @Description 简单的Callable实现
**/
public class SimpleCallable {
public static void main(String[] args) throws ExecutionException, InterruptedException {
FutureTask futureTask=new FutureTask(new Callable() {
@Override
public Object call() throws Exception {
int sum = 0 ;
for(int i = 1 ; i <= 10 ; i++ ){
System.out.println(Thread.currentThread().getName()+" => " + i);
sum+=i;
}
return Thread.currentThread().getName()+"执行的结果是:"+sum;
}
});
futureTask.run();
System.out.println(futureTask.get());
}
}
Future接口
Future接口至少提供了三大功能:(1)能够取消异步执行中的任务。(2)判断异步任务是否执行完成。(3)获取异步任务完成后的执行结果。
package java.util.concurrent;
public interface Future<V> {
boolean cancel(boolean mayInterruptRunning); //取消异步执行
boolean isCancelled();
boolean isDone();//判断异步任务是否执行完成
//获取异步任务完成后的执行结果
V get() throws InterruptedException, ExecutionException;
//设置时限,获取异步任务完成后的执行结果
V get(long timeout, TimeUnit unit) throws InterruptedException,
ExecutionException, TimeoutException;
}
RunnableFuture接口
public interface RunnableFuture<V> extends Runnable, Future<V> {
void run();
}
RunnableFuture继承了Runnable接口,从而保证了其实例可以作为Thread线程实例的target目标;同时,RunnableFuture通过继承Future接口,保证了可以获取未来的异步执行结果。
FutureTask实现类
FutureTask类是Future接口的实现类,提供了对异步任务的操作的具体实现。但是,FutureTask类不仅实现了Future接口,还实现了Runnable接口,或者更加准确地说,FutureTask类实现了RunnableFuture接口。

总结一下
线程池创建线程
实际上创建一个线程实例在时间成本、资源耗费上都很高(稍后会介绍),在高并发的场景中,断然不能频繁进行线程实例的创建与销毁,而是需要对已经创建好的线程实例进行复用,这就涉及线程池的技术。(线程复用)
//创建一个包含三个线程的线程池
private static ExecutorService pool = Executors.newFixedThreadPool(3);
//方法一:执行一个 Runnable类型的target执行目标实例,无返回
void execute(Runnable command);
//方法二:提交一个 Callable类型的target执行目标实例, 返回一个Future异步任务实例
<T> Future<T> submit(Callable<T> task);
//方法三:提交一个 Runnable类型的target执行目标实例, 返回一个Future异步任务实例
Future<?> submit(Runnable task);
时间片
线程调度
线程优先级请看前面内容

新建(New)、就绪(Runnable)、运行(Running)、阻塞(Blocked)和死亡(Dead)这5种状态。
wait方法的线程会进入WAITING状态,只有等到其他线程的通知或被中断后才会返回。sleep方法会导致当前线程休眠。与wait方法不同的是,sleep方法不会释放当前占有的锁,会导致线程进入TIMED-WATING状态,而wait方法会导致当前线程进入WATING状态。 //使目前正在执行的线程休眠millis毫秒
public static void sleep(long millis) throws InterruptException;
//使目前正在执行的线程休眠millis毫秒,nanos纳秒
public static void sleep(long millis,int nanos) throws InterruptException;
yield方法会使当前线程让出(释放)CPU执行时间片,与其他线程一起重新竞争CPU时间片。(运行状态转为就绪状态)interrupt方法用于向线程发行一个终止通知信号,会影响该线程内部的一个中断标识位,这个线程本身并不会因为调用了interrupt方法而改变状态(阻塞、终止等)。join方法用于等待其他线程终止,如果在当前线程中调用一个线程的join方法,则当前线程转为阻塞状态,等到另一个线程结束,当前线程再由阻塞状态转为就绪状态,等待获取CPU的使用权。 //重载版本1:此方法会把当前线程变为TIMED_WAITING,直到被合并线程执行结束
public final void join() throws InterruptedException:
//重载版本2:此方法会把当前线程变为TIMED_WAITING,直到被合并线程执行结束,或者等待被合并线程执行millis的时间
public final synchronized void join(long millis) throws InterruptedException:
//重载版本3:此方法会把当前线程变为TIMED_WAITING,直到被合并线程执行结束,或者等待被合并线程执行millis+nanos的时间
public final synchroinzed void join(long millis, int nanos) throws InterruptedException:
notify方法,用于唤醒在此对象监视器上等待的一个线程,如果所有线程都在此对象上等待,则会选择唤醒其中一个线程,选择是任意的。setDaemon方法用于定义一个守护线程,也叫作服务线程,该线程是后台线程,有一个特性,即为用户线程提供公共服务,在没有用户线程可服务时会自动离开。New状态
通过new Thread(…)已经创建线程,但尚未调用start()启动线程,该线程处于NEW(新建)状态。
RUNNABLE状态
Java把Ready(就绪)和Running(执行)两种状态合并为一种状态:RUNNABLE(可执行)状态(或者可运行状态)。调用了线程的start()实例方法后,线程就处于就绪状态。此线程获取到CPU时间片后,开始执行run()方法中的业务代码,线程处于执行状态。
就绪状态仅仅表示线程具备运行资格,如果没有被操作系统的调度程序挑选中,线程就永远处于就绪状态。当前线程进入就绪状态的条件大致包括以下几种:调用线程的start()方法,此线程就会进入就绪状态。·当前线程的执行时间片用完。·线程睡眠(Sleep)操作结束。·对其他线程合入(Join)操作结束。·等待用户输入结束。·线程争抢到对象锁(Object Monitor)。·当前线程调用了yield()方法出让CPU执行权限。
线程调度程序从就绪状态的线程中选择一个线程,被选中的线程状态将变成执行状态。这也是线程进入执行状态的唯一方式。
BLOCKED状态
(1)线程等待获取锁等待获取一个锁,而该锁被其他线程持有,则该线程进入阻塞状态。当其他线程释放了该锁,并且线程调度器允许该线程持有该锁时,该线程退出阻塞状态。(2)IO阻塞线程发起了一个阻塞式IO操作后,如果不具备IO操作的条件,线程就会进入阻塞状态。IO包括磁盘IO、网络IO等。IO阻塞的一个简单例子:线程等待用户输入内容后继续执行。
WAITING状态
处于WAITING(无限期等待)状态的线程不会被分配CPU时间片,需要被其他线程显式地唤醒,才会进入就绪状态。线程调用以下3种方法会让自己进入无限等待状态:·Object.wait()方法,对应的唤醒方式为:Object.notify()/Object.notifyAll()。·Thread.join()方法,对应的唤醒方式为:被合入的线程执行完毕。·LockSupport.park()方法,对应的唤醒方式为:LockSupport.unpark(Thread)。
TIMED_WAITING状态
处于TIMED_WAITING(限时等待)状态的线程不会被分配CPU时间片,如果指定时间之内没有被唤醒,限时等待的线程会被系统自动唤醒,进入就绪状态。以下3种方法会让线程进入限时等待状态:·Thread.sleep(time)方法,对应的唤醒方式为:sleep睡眠时间结束。·Object.wait(time)方法,对应的唤醒方式为:调用Object.notify()/Object.notifyAll()主动唤醒,或者限时结束。·LockSupport.parkNanos(time)/parkUntil(time)方法,对应的唤醒方式为:线程调用配套的LockSupport.unpark(Thread)方法结束,或者线程停止(park)时限结束。进入BLOCKED状态、WAITING状态、TIMED_WAITING状态的线程都会让出CPU的使用权;另外,等待或者阻塞状态的线程被唤醒后,进入Ready状态,需要重新获取时间片才能接着运行。
TERMINATED状态
线程结束任务之后,将会正常进入TERMINATED(死亡)状态;或者说在线程执行过程中发生了异常(而没有被处理),也会导致线程进入死亡状态。


创建一个单线程化线程池,也就是只有一个线程的线程池,所创建的线程池用唯一的工作线程来执行任务,使用此方法创建的线程池能保证所有任务按照指定顺序(如FIFO)执行。
package ExecutorDemo.newSingleThreadExecutor;
import java.util.concurrent.Executor;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.atomic.AtomicInteger;
/**
* @description:
* @author: shu
* @createDate: 2022/10/30 17:32
* @version: 1.0
*/
public class SingleThreadExecutorDemo {
public static final int SLEEP_FLAG = 500;
/**
* 静态类
*/
static class TargetTask implements Runnable {
// 计数器,原子类
static AtomicInteger TaskNo = new AtomicInteger();
public String taskName;
/**
* 构造器
*/
public TargetTask() {
this.taskName = "task-" + TaskNo.get();
TaskNo.incrementAndGet();
}
/**
* 线程入口
*/
@Override
public void run() {
System.out.println("任务编号-" + taskName + "正在运行");
try {
Thread.sleep(SLEEP_FLAG);
System.out.println("任务编号-" + taskName + "运行结束");
} catch (InterruptedException e) {
throw new RuntimeException(e);
}
}
}
/**
* 主方法
* @param args
* @throws InterruptedException
*/
public static void main(String[] args) throws InterruptedException {
// 线程池
ExecutorService singleThreadExecutor = Executors.newSingleThreadExecutor();
// 批量添加线程
for (int i = 0; i < 5; i++) {
singleThreadExecutor.execute(new TargetTask());
// singleThreadExecutor.submit(new TargetTask());
}
Thread.sleep(1000);
// 线程池销毁
singleThreadExecutor.shutdown();;
}
}

特点
使用场景
单线程化的线程池所适用的场景是:任务按照提交次序,一个任务一个任务地逐个执行的场景。
package ExecutorDemo.newFixedThreadPool;
import ExecutorDemo.newSingleThreadExecutor.SingleThreadExecutorDemo;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.atomic.AtomicInteger;
/**
* @description:
* @author: shu
* @createDate: 2022/10/30 17:55
* @version: 1.0
*/
public class FixedThreadPoolDemo {
public static final int SLEEP_FLAG = 500;
/**
* 静态类
*/
static class TargetTask implements Runnable {
// 计数器,原子类
static AtomicInteger TaskNo = new AtomicInteger();
public String taskName;
/**
* 构造器
*/
public TargetTask() {
this.taskName = "task-" + TaskNo.get();
TaskNo.incrementAndGet();
}
/**
* 线程入口
*/
@Override
public void run() {
System.out.println("任务编号-" + taskName + "正在运行");
try {
Thread.sleep(SLEEP_FLAG);
System.out.println("任务编号-" + taskName + "运行结束");
} catch (InterruptedException e) {
throw new RuntimeException(e);
}
}
}
/**
* 主方法
* @param args
* @throws InterruptedException
*/
public static void main(String[] args) throws InterruptedException {
// 线程池
ExecutorService singleThreadExecutor = Executors.newFixedThreadPool(2);
// 批量添加线程
for (int i = 0; i < 7; i++) {
singleThreadExecutor.execute(new TargetTask());
// singleThreadExecutor.submit(new TargetTask());
}
Thread.sleep(1000);
// 线程池销毁
singleThreadExecutor.shutdown();;
}
}

特点
使用场景
固定数量的线程池的适用场景:需要任务长期执行的场景。但是内部使用无界队列来存放排队任务,当大量任务超过线程池最大容量需要处理时,队列无限增大,使服务器资源迅速耗尽。
package ExecutorDemo.newCachedThreadPool;
import ExecutorDemo.newFixedThreadPool.FixedThreadPoolDemo;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.atomic.AtomicInteger;
/**
* @description:
* @author: shu
* @createDate: 2022/10/30 18:35
* @version: 1.0
*/
public class CachedThreadPoolDemo {
public static final int SLEEP_FLAG = 500;
/**
* 静态类
*/
static class TargetTask implements Runnable {
// 计数器,原子类
static AtomicInteger TaskNo = new AtomicInteger();
public String taskName;
/**
* 构造器
*/
public TargetTask() {
this.taskName = "task-" + TaskNo.get();
TaskNo.incrementAndGet();
}
/**
* 线程入口
*/
@Override
public void run() {
System.out.println("任务编号-" + taskName + "正在运行");
try {
Thread.sleep(SLEEP_FLAG);
System.out.println("任务编号-" + taskName + "运行结束");
} catch (InterruptedException e) {
throw new RuntimeException(e);
}
}
}
/**
* 主方法
* @param args
* @throws InterruptedException
*/
public static void main(String[] args) throws InterruptedException {
// 线程池
ExecutorService singleThreadExecutor = Executors.newCachedThreadPool();
// 批量添加线程
for (int i = 0; i < 7; i++) {
singleThreadExecutor.execute(new TargetTask());
// singleThreadExecutor.submit(new TargetTask());
}
Thread.sleep(1000);
// 线程池销毁
singleThreadExecutor.shutdown();;
}
}

特点
使用场景
可缓存线程池”的适用场景:需要快速处理突发性强、耗时较短的任务场景,如Netty的NIO处理场景、REST API接口的瞬时削峰场景。但是线程池没有最大线程数量限制,如果大量的异步任务执行目标实例同时提交,可能会因创建线程过多而导致资源耗尽。
该方法用于创建一个可调度线程池,即一个提供“延时”和“周期性”任务调度功能的ScheduledExecutorService类型的线程池
package ExecutorDemo.newScheduledThreadPool;
import ExecutorDemo.newFixedThreadPool.FixedThreadPoolDemo;
import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicInteger;
/**
* @description:
* @author: shu
* @createDate: 2022/10/30 19:54
* @version: 1.0
*/
public class ScheduledThreadPoolDemo {
public static final int SLEEP_FLAG = 500;
/**
* 静态类
*/
static class TargetTask implements Runnable {
// 计数器,原子类
static AtomicInteger TaskNo = new AtomicInteger();
public String taskName;
/**
* 构造器
*/
public TargetTask() {
this.taskName = "task-" + TaskNo.get();
TaskNo.incrementAndGet();
}
/**
* 线程入口
*/
@Override
public void run() {
System.out.println("任务编号-" + taskName + "正在运行");
try {
Thread.sleep(SLEEP_FLAG);
System.out.println("任务编号-" + taskName + "运行结束");
} catch (InterruptedException e) {
throw new RuntimeException(e);
}
}
}
/**
* 主方法
* @param args
* @throws InterruptedException
*/
public static void main(String[] args) throws InterruptedException {
// 线程池
ScheduledExecutorService service = Executors.newScheduledThreadPool(2);
// 批量添加线程
for (int i = 0; i < 7; i++) {
ScheduledFuture<?> future = service.scheduleWithFixedDelay(new TargetTask(), 0, 500, TimeUnit.MILLISECONDS);
}
Thread.sleep(1000);
// 线程池销毁
service.shutdown();;
}
}
参数
public ScheduledFuture<?> scheduleAtFixedRate(
Runnable command, //异步任务target执行目标实例
long initialDelay, //首次执行延时
long period, //两次开始执行最小间隔时间
TimeUnit unit //所设置的时间的计时单位,如TimeUnit.SECONDS常量
);
public ScheduledFuture<?> scheduleWithFixedDelay(
Runnable command, //异步任务target执行目标实例
long initialDelay, //首次执行延时
long delay, //前一次执行结束到下一次执行开始的间隔时间(间隔执行延迟时间)
TimeUnit unit //所设置的时间的计时单位,如TimeUnit.SECONDS常量
);
使用场景
可调度线程池的适用场景:周期性地执行任务的场景。Spring Boot中的任务调度器,底层借助了JUC的ScheduleExecutorService“可调度线程池”实现,并且可以通过@Configuration配置类型的Bean。
注意
尽管Executors的工厂方法使用方便,但是在生产场景中被很多企业(尤其是大厂)的开发规范所禁用。
注意
// 使用标准构造器构造一个普通的线程池
public ThreadPoolExecutor(
int corePoolSize, // 核心线程数,即使线程空闲(Idle),也不会回收
int maximumPoolSize, // 线程数的上限
long keepAliveTime, TimeUnit unit, // 线程最大空闲(Idle)时长
BlockingQueue<Runnable> BlockingQueue, // 任务的排队队列
ThreadFactory threadFactory, // 新线程的产生方式
RejectedExecutionHandler handler) // 拒绝策略

参数corePoolSize用于设置核心(Core)线程池数量,参数maximumPoolSize用于设置最大线程数量。
合理设置
参考文章:https://blog.csdn.net/riemann_/article/details/104704197
CPU密集型:corePoolSize = CPU核数 + 1
IO密集型:corePoolSize = CPU核数 * 2
工作规则
BlockingQueue(阻塞队列)的实例用于暂时接收到的异步任务,如果线程池的核心线程都在忙,那么所接收到的目标任务缓存在阻塞队列中。
线程构造器的keepAliveTime(空闲线程存活时间)参数用于设置池内线程最大Idle(空闲)时长(或者说保活时长),如果超过这个时间,默认情况下Idle、非Core线程会被回收。
但若调用了allowCoreThreadTimeOut(boolean)方法,并且传入了参数true,则keepAliveTime参数所设置的Idle超时策略也将被应用于核心线程。
在线程池的任务缓存队列为有界队列(有容量限制的队列)的时候,如果队列满了,提交任务到线程池的时候就会被拒绝。总体来说,任务被拒绝有两种情况:(1)线程池已经被关闭。(2)工作队列已满且maximumPoolSize已满。
AbortPolicy:拒绝策略。·DiscardPolicy:抛弃策略。·DiscardOldestPolicy:抛弃最老任务策略。·CallerRunsPolicy:调用者执行策略。·自定义策略。


ThreadFactory是Java线程工厂接口,这是一个非常简单的接口,具体如下:
package java.util.concurrent;
public interface ThreadFactory {
//唯一的方法:创建一个新线程
Thread newThread(Runnable target);
}
package ThreadFactory;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.ThreadFactory;
import java.util.concurrent.atomic.AtomicInteger;
/**
* @description: 先建线程工厂
* @author: shu
* @createDate: 2022/10/31 16:01
* @version: 1.0
*/
public class ThreadFactoryDemo {
public static final int SLEEP_FLAG = 500;
//一个简单的线程工厂
static public class SimpleThreadFactory implements ThreadFactory
{
static AtomicInteger threadNo = new AtomicInteger(1);
//实现其唯一的创建线程方法
@Override
public Thread newThread(Runnable target)
{
String threadName = "simpleThread-" + threadNo.get();
System.out.println("创建一个线程,名称为:" + threadName);
threadNo.incrementAndGet();
//设置线程名称和异步执行目标
Thread thread = new Thread(target,threadName);
//设置为守护线程
thread.setDaemon(true);
return thread;
}
}
/**
* 静态类
*/
static class TargetTask implements Runnable {
// 计数器,原子类
static AtomicInteger TaskNo = new AtomicInteger();
public String taskName;
/**
* 构造器
*/
public TargetTask() {
this.taskName = "task-" + TaskNo.get();
TaskNo.incrementAndGet();
}
/**
* 线程入口
*/
@Override
public void run() {
System.out.println("任务编号-" + taskName + "正在运行");
try {
Thread.sleep(SLEEP_FLAG);
System.out.println("任务编号-" + taskName + "运行结束");
} catch (InterruptedException e) {
throw new RuntimeException(e);
}
}
}
public static void main(String[] args) throws InterruptedException {
//使用自定义线程工厂快捷创建一个固定大小的线程池
ExecutorService pool =
Executors.newFixedThreadPool(2,new SimpleThreadFactory());
for (int i = 0; i < 5; i++)
{
pool.submit(new TargetTask());
}
//等待10秒
Thread.sleep(1000);
// 线程池销毁
pool.shutdown();
}
}
// 省略其他

注意
Executors为线程池工厂类,用于快捷创建线程池(Thread Pool);ThreadFactory为线程工厂类,用于创建线程(Thread)。

ArrayBlockingQueue在创建时必须设置大小,接收的任务超出corePoolSize数量时,任务被缓存到该阻塞队列中,任务缓存的数量只能为创建时设置的大小,若该阻塞队列已满,则会为新的任务创建线程,直到线程池中的线程总数大于maximumPoolSize。
LinkedBlockingQueue:是一个基于链表实现的阻塞队列,按FIFO排序任务,可以设置容量(有界队列),不设置容量则默认使用Integer.Max_VALUE作为容量(无界队列)。该队列的吞吐量高于ArrayBlockingQueue。
是具有优先级的无界队列
这是一个无界阻塞延迟队列,底层基于PriorityBlockingQueue实现,队列中每个元素都有过期时间,当从队列获取元素(元素出队)时,只有已经过期的元素才会出队,队列头部的元素是过期最快的元素。
(同步队列)是一个不存储元素的阻塞队列,每个插入操作必须等到另一个线程的调用移除操作,否则插入操作一直处于阻塞状态,其吞吐量通常高于LinkedBlockingQueue
ThreadPoolExecutor线程池调度器为每个任务执行前后都提供了钩子方法。
//任务执行之前的钩子方法(前钩子)
protected void beforeExecute(Thread t, Runnable r) { }
//任务执行之后的钩子方法(后钩子)
protected void afterExecute(Runnable r, Throwable t) { }
//线程池终止时的钩子方法(停止钩子)
protected void terminated() { }
package ThreadFactory;
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;
/**
* @description:
* @author: shu
* @createDate: 2022/10/31 16:52
* @version: 1.0
*/
public class ThreadPoolExecutorDemo {
public static final int SLEEP_FLAG = 500;
/**
* 静态类
*/
static class TargetTask implements Runnable {
// 计数器,原子类
static AtomicInteger TaskNo = new AtomicInteger();
public String taskName;
/**
* 构造器
*/
public TargetTask() {
this.taskName = "task-" + TaskNo.get();
TaskNo.incrementAndGet();
}
/**
* 线程入口
*/
@Override
public void run() {
System.out.println("任务编号-" + taskName + "正在运行");
try {
Thread.sleep(SLEEP_FLAG);
System.out.println("任务编号-" + taskName + "运行结束");
} catch (InterruptedException e) {
throw new RuntimeException(e);
}
}
}
public static void main(String[] args) throws InterruptedException {
ExecutorService pool = new ThreadPoolExecutor(2,//coreSize
4, //最大线程数
60,//空闲保活时长
TimeUnit.SECONDS,
new LinkedBlockingQueue<>(2)) //等待队列
{
//继承:调度器终止钩子
@Override
protected void terminated()
{
System.out.println("调度器已经终止!");
}
//继承:执行前钩子
@Override
protected void beforeExecute(Thread t, Runnable target)
{
System.out.println("前钩被执行!");
super.beforeExecute(t, target);
}
//继承:执行后钩子
@Override
protected void afterExecute(Runnable target, Throwable t)
{
System.out.println("钩子函数执行后");
}
};
for (int i = 1; i <= 5; i++)
{
pool.execute(new TargetTask());
}
//等待10秒
Thread.sleep(1000);
// 线程池销毁
pool.shutdown();
}
}
在线程池的任务缓存队列为有界队列(有容量限制的队列)的时候,如果队列满了,提交任务到线程池的时候就会被拒绝。总体来说,任务被拒绝有两种情况:(1)线程池已经被关闭。(2)工作队列已满且maximumPoolSize已满。
·AbortPolicy:拒绝策略。·DiscardPolicy:抛弃策略。·DiscardOldestPolicy:抛弃最老任务策略。·CallerRunsPolicy:调用者执行策略。·自定义策略。
线程的5种状态:
package java.util.concurrent;
// 省略import
public class ThreadPoolExecutor extends AbstractExecutorService {
// runState is stored in the high-order bits
private static final int RUNNING = -1 << COUNT_BITS;
private static final int SHUTDOWN = 0 << COUNT_BITS;
private static final int STOP = 1 << COUNT_BITS;
private static final int TIDYING = 2 << COUNT_BITS;
private static final int TERMINATED = 3 << COUNT_BITS;
// 省略其他
}
关闭线程的方法
shutdown:是JUC提供的一个有序关闭线程池的方法,此方法会等待当前工作队列中的剩余任务全部执行完成之后,才会执行关闭,但是此方法被调用之后线程池的状态转为SHUTDOWN,线程池不会再接收新的任务。
shutdownNow:是JUC提供的一个立即关闭线程池的方法,此方法会打断正在执行的工作线程,并且会清空当前工作队列中的剩余任务,返回的是尚未执行的任务。
awaitTermination:等待线程池完成关闭。在调用线程池的shutdown()与shutdownNow()方法时,当前线程会立即返回,不会一直等待直到线程池完成关闭。如果需要等到线程池关闭完成,可以调用awaitTermination()方法。
源码分析
shutdown
public void shutdown()
{
final ReentrantLock mainLock = this.mainLock;
mainLock.lock();
try
{
// 检查权限
checkShutdownAccess();
// 设置线程池状态
advanceRunState(SHUTDOWN);
// 中断空闲线程
interruptIdleWorkers();
// 钩子函数,主要用于清理一些资源
onShutdown();
} finally
{
mainLock.unlock();
}
tryTerminate();
}
shutdownNow()
public List<Runnable> shutdownNow()
{
List<Runnable> tasks;
final ReentrantLock mainLock = this.mainLock;
mainLock.lock();
try
{
// 检查状态
checkShutdownAccess();
// 将线程池状态变为 STOP
advanceRunState(STOP);
// 中断所有线程,包括工作线程以及空闲线程
interruptWorkers();
// 丢弃工作队列中的剩余任务
tasks = drainQueue();
} finally
{
mainLock.unlock();
}
tryTerminate();
return tasks;
}
awaitTermination()
threadPool.shutdown();
try {
//一直等待,直到线程池完成关闭
while (!threadPool.awaitTermination(60,TimeUnit.SECONDS)){
System.out.println("线程池任务还未执行结束");
}
} catch (InterruptedException e) {
e.printStackTrace();
}
如果线程池完成关闭,awaitTermination()方法将会返回true,否则当等待时间超过指定时间后将会返回false。如果需要调用awaitTermination(),建议不是永久等待,而是设置一定重试次数。
优雅的关闭
package Utils;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.TimeUnit;
/**
* @description:
* @author: shu
* @createDate: 2022/10/31 20:19
* @version: 1.0
*/
public class ThreadUtils {
public static void shutdownThreadPoolGracefully(
ExecutorService threadPool)
{
// 若已经关闭则返回
if (!(threadPool instanceof ExecutorService) ||
threadPool.isTerminated())
{
return;
}
try
{
threadPool.shutdown(); //拒绝接受新任务
} catch (SecurityException e)
{
return;
} catch (NullPointerException e)
{
return;
}
try
{
// 等待60秒,等待线程池中的任务完成执行
if (!threadPool.awaitTermination(60, TimeUnit.SECONDS))
{
// 调用 shutdownNow() 方法取消正在执行的任务
threadPool.shutdownNow();
// 再次等待60秒,如果还未结束,可以再次尝试,或者直接放弃
if (!threadPool.awaitTermination(60, TimeUnit.SECONDS))
{
System.err.println("线程池任务未正常执行结束");
}
}
} catch (InterruptedException ie)
{
// 捕获异常,重新调用 shutdownNow() 方法
threadPool.shutdownNow();
}
// 仍然没有关闭,循环关闭1000次,每次等待10毫秒
if (!threadPool.isTerminated())
{
try
{
for (int i = 0; i < 1000; i++)
{
if (threadPool.awaitTermination(10, TimeUnit.MILLISECONDS))
{
break;
}
threadPool.shutdownNow();
}
} catch (InterruptedException e)
{
System.err.println(e.getMessage());
} catch (Throwable e)
{
System.err.println(e.getMessage());
}
}
}
}