이 글에서는 java에 대한 관련 지식을 소개하며, 스레드 풀을 사용하는 이유와 스레드 풀 사용에 대한 관련 내용을 포함하여 스레드 풀의 구현 원리에 대한 내용을 주로 소개합니다. 모두를 돕습니다.
추천 학습: "java 비디오 튜토리얼"
스레드 풀을 사용하는 이유는 일반적으로 다음 두 가지 때문입니다.
자주 스레드를 생성하고 삭제합니다. 시스템 리소스를 소비하므로 스레드 풀을 사용하여 스레드를 재사용할 수 있습니다.
스레드 풀을 사용하면 스레드 관리가 더 쉬워집니다. 스레드 풀은 스레드 수를 동적으로 관리하고, 차단 대기열을 가지며, 정기적으로 작업을 실행하고, 환경 격리 등을 수행할 수 있습니다.
/** * @author 一灯架构 * @apiNote 线程池示例 **/ public class ThreadPoolDemo { public static void main(String[] args) { // 1. 创建线程池 ThreadPoolExecutor threadPoolExecutor = new ThreadPoolExecutor( 3, 3, 0L, TimeUnit.MILLISECONDS, new LinkedBlockingQueue(), Executors.defaultThreadFactory(), new ThreadPoolExecutor.AbortPolicy()); // 2. 往线程池中提交3个任务 for (int i = 0; i { System.out.println(Thread.currentThread().getName() + " 关注公众号:一灯架构"); }); } // 3. 关闭线程池 threadPoolExecutor.shutdown(); } }
출력 결과:
pool-1-thread-2 关注公众号:一灯架构 pool-1-thread-1 关注公众号:一灯架构 pool-1-thread-3 关注公众号:一灯架构
스레드 풀 사용은 매우 간단합니다.
새 ThreadPoolExecutor() 생성자를 호출하고 핵심 매개변수를 지정하고 스레드 풀을 만듭니다. .
execute() 메서드를 호출하여 Runnable 작업을 제출합니다.
사용 후 shutdown() 메서드를 호출하여 스레드 풀을 닫습니다.
스레드 풀 구성 방식에서 핵심 매개변수의 역할을 다시 살펴보겠습니다.
스레드 풀에는 7개의 핵심 매개변수가 있습니다:
매개변수 이름 | 매개변수 의미 |
---|---|
int corePoolSize | 코어 스레드 수 |
int maximumPoolSize | 최대 스레드 수 |
long keepAliveTime | 스레드 생존 시간 |
TimeUnit 단위 | 시간 단위 |
BlockingQueue workQueue | Blocking queue |
ThreadFactory threadFactory | 스레드 생성 Factory |
RejectedExecutionHandler 핸들러 | 거부 전략 |
corePoolSize 核心线程数
当往线程池中提交任务,会创建线程去处理任务,直到线程数达到corePoolSize,才会往阻塞队列中添加任务。默认情况下,空闲的核心线程并不会被回收,除非配置了allowCoreThreadTimeOut=true。
maximumPoolSize 最大线程数
当线程池中的线程数达到corePoolSize,阻塞队列又满了之后,才会继续创建线程,直到达到maximumPoolSize,另外空闲的非核心线程会被回收。
keepAliveTime 线程存活时间
非核心线程的空闲时间达到了keepAliveTime,将会被回收。
TimeUnit 时间单位
线程存活时间的单位,默认是TimeUnit.MILLISECONDS(毫秒),可选择的有:
TimeUnit.NANOSECONDS(纳秒) TimeUnit.MICROSECONDS(微秒) TimeUnit.MILLISECONDS(毫秒) TimeUnit.SECONDS(秒) TimeUnit.MINUTES(分钟) TimeUnit.HOURS(小时) TimeUnit.DAYS(天)
workQueue 阻塞队列
当线程池中的线程数达到corePoolSize,再提交的任务就会放到阻塞队列的等待,默认使用的是LinkedBlockingQueue,可选择的有:
LinkedBlockingQueue(基于链表实现的阻塞队列)
ArrayBlockingQueue(基于数组实现的阻塞队列)
SynchronousQueue(只有一个元素的阻塞队列)
PriorityBlockingQueue(实现了优先级的阻塞队列)
DelayQueue(实现了延迟功能的阻塞队列)
threadFactory 线程创建工厂
用来创建线程的工厂,默认的是Executors.defaultThreadFactory(),可选择的还有Executors.privilegedThreadFactory()实现了线程优先级。当然也可以自定义线程创建工厂,创建线程的时候最好指定线程名称,便于排查问题。
RejectedExecutionHandler 拒绝策略
当线程池中的线程数达到maximumPoolSize,阻塞队列也满了之后,再往线程池中提交任务,就会触发执行拒绝策略,默认的是AbortPolicy(直接终止,抛出异常),可选择的有:
AbortPolicy(直接终止,抛出异常)
DiscardPolicy(默默丢弃,不抛出异常)
DiscardOldestPolicy(丢弃队列中最旧的任务,执行当前任务)
CallerRunsPolicy(返回给调用者执行)
线程池的工作原理,简单理解如下:
当往线程池中提交任务的时候,会先判断线程池中线程数是否核心线程数,如果小于,会创建核心线程并执行任务。
如果线程数大于核心线程数,会判断阻塞队列是否已满,如果没有满,会把任务添加到阻塞队列中等待调度执行。
如果阻塞队列已满,会判断线程数是否小于最大线程数,如果小于,会继续创建最大线程数并执行任务。
如果线程数大于最大线程数,会执行拒绝策略,然后结束。
public class ThreadPoolExecutor extends AbstractExecutorService { // 线程池的控制状态,Integer长度是32位,前3位用来存储线程池状态,后29位用来存储线程数量 private final AtomicInteger ctl = new AtomicInteger(ctlOf(RUNNING, 0)); // 线程个数所占的位数 private static final int COUNT_BITS = Integer.SIZE - 3; // 线程池的最大容量,2^29-1,约5亿个线程 private static final int CAPACITY = (1 workers = new HashSet(); // 等待条件,用来响应中断 private final Condition termination = mainLock.newCondition(); // 是否允许回收核心线程 private volatile boolean allowCoreThreadTimeOut; // 线程数的历史峰值 private int largestPoolSize; /** * 以下是线程池的七大核心参数 */ private volatile int corePoolSize; private volatile int maximumPoolSize; private volatile long keepAliveTime; private final BlockingQueue<runnable> workQueue; private volatile ThreadFactory threadFactory; private volatile RejectedExecutionHandler handler; }</runnable>
线程池的控制状态ctl用来存储线程池状态和线程个数,前3位用来存储线程池状态,后29位用来存储线程数量。
设计者多聪明,用一个变量存储了两块内容。
线程池共有5种状态:
状态名称 | 状态含义 | 状态作用 |
---|---|---|
RUNNING | 运行中 | 线程池创建后默认状态,接收新任务,并处理阻塞队列中的任务。 |
SHUTDOWN | 已关闭 | 调用shutdown方法后处于该状态,不再接收新任务,处理阻塞队列中任务。 |
STOP | 已停止 | 调用shutdownNow方法后处于该状态,不再新任务,并中断所有线程,丢弃阻塞队列中所有任务。 |
TIDYING | 处理中 | 所有任务已完成,所有工作线程都已回收,等待调用terminated方法。 |
TERMINATED | 已终止 | 调用terminated方法后处于该状态,线程池的最终状态。 |
看一下往线程池中提交任务的源码,这是线程池的核心逻辑:
// 往线程池中提交任务 public void execute(Runnable command) { // 1. 判断提交的任务是否为null if (command == null) throw new NullPointerException(); int c = ctl.get(); // 2. 判断线程数是否小于核心线程数 if (workerCountOf(c) <p>execute方法的逻辑也很简单,最终就是调用addWorker方法,把任务添加到worker集合中,再看一下addWorker方法的源码:</p><pre class="brush:php;toolbar:false">// 添加worker private boolean addWorker(Runnable firstTask, boolean core) { retry: for (; ; ) { int c = ctl.get(); int rs = runStateOf(c); // 1. 检查是否允许提交任务 if (rs >= SHUTDOWN && !(rs == SHUTDOWN && firstTask == null && !workQueue.isEmpty())) return false; // 2. 使用死循环保证添加线程成功 for (; ; ) { int wc = workerCountOf(c); // 3. 校验线程数是否超过容量限制 if (wc >= CAPACITY || wc >= (core ? corePoolSize : maximumPoolSize)) return false; // 4. 使用CAS修改线程数 if (compareAndIncrementWorkerCount(c)) break retry; c = ctl.get(); // 5. 如果线程池状态变了,则从头再来 if (runStateOf(c) != rs) continue retry; } } boolean workerStarted = false; boolean workerAdded = false; Worker w = null; try { // 6. 把任务和新线程包装成一个worker w = new Worker(firstTask); final Thread t = w.thread; if (t != null) { // 7. 加锁,控制并发 final ReentrantLock mainLock = this.mainLock; mainLock.lock(); try { // 8. 再次校验线程池状态是否异常 int rs = runStateOf(ctl.get()); if (rs largestPoolSize) largestPoolSize = s; workerAdded = true; } } finally { mainLock.unlock(); } if (workerAdded) { // 12. 启动线程 t.start(); workerStarted = true; } } } finally { if (!workerStarted) addWorkerFailed(w); } return workerStarted; }
方法虽然很长,但是逻辑很清晰。就是把任务和线程包装成worker,添加到worker集合,并启动线程。
再看一下worker类的结构:
private final class Worker extends AbstractQueuedSynchronizer implements Runnable { // 工作线程 final Thread thread; // 任务 Runnable firstTask; // 创建worker,并创建一个新线程(用来执行任务) Worker(Runnable firstTask) { setState(-1); this.firstTask = firstTask; this.thread = getThreadFactory().newThread(this); } }
再看一下run方法的源码:
// 线程执行入口 public void run() { runWorker(this); } // 线程运行核心方法 final void runWorker(Worker w) { Thread wt = Thread.currentThread(); Runnable task = w.firstTask; w.firstTask = null; w.unlock(); boolean completedAbruptly = true; try { // 1. 如果当前worker中任务是null,就从阻塞队列中获取任务 while (task != null || (task = getTask()) != null) { // 加锁,保证thread不被其他线程中断(除非线程池被中断) w.lock(); // 2. 校验线程池状态,是否需要中断当前线程 if ((runStateAtLeast(ctl.get(), STOP) || (Thread.interrupted() && runStateAtLeast(ctl.get(), STOP))) && !wt.isInterrupted()) wt.interrupt(); try { beforeExecute(wt, task); Throwable thrown = null; try { // 3. 执行run方法 task.run(); } catch (RuntimeException x) { thrown = x; throw x; } catch (Error x) { thrown = x; throw x; } catch (Throwable x) { thrown = x; throw new Error(x); } finally { afterExecute(task, thrown); } } finally { task = null; w.completedTasks++; // 解锁 w.unlock(); } } completedAbruptly = false; } finally { // 4. 从worker集合删除当前worker processWorkerExit(w, completedAbruptly); } }
runWorker方法逻辑也很简单,就是不断从阻塞队列中拉取任务并执行。
再看一下从阻塞队列中拉取任务的逻辑:
// 从阻塞队列中拉取任务 private Runnable getTask() { boolean timedOut = false; for (; ; ) { int c = ctl.get(); int rs = runStateOf(c); // 1. 如果线程池已经停了,或者阻塞队列是空,就回收当前线程 if (rs >= SHUTDOWN && (rs >= STOP || workQueue.isEmpty())) { decrementWorkerCount(); return null; } int wc = workerCountOf(c); // 2. 再次判断是否需要回收线程 boolean timed = allowCoreThreadTimeOut || wc > corePoolSize; if ((wc > maximumPoolSize || (timed && timedOut)) && (wc > 1 || workQueue.isEmpty())) { if (compareAndDecrementWorkerCount(c)) return null; continue; } try { // 3. 从阻塞队列中拉取任务 Runnable r = timed ? workQueue.poll(keepAliveTime, TimeUnit.NANOSECONDS) : workQueue.take(); if (r != null) return r; timedOut = true; } catch (InterruptedException retry) { timedOut = false; } } }
推荐学习:《java视频教程》
위 내용은 하나의 기사로 Java 스레드 풀의 구현 원리를 이해하십시오.의 상세 내용입니다. 자세한 내용은 PHP 중국어 웹사이트의 기타 관련 기사를 참조하세요!