- 浏览: 979840 次
文章分类
- 全部博客 (428)
- Hadoop (2)
- HBase (1)
- ELK (1)
- ActiveMQ (13)
- Kafka (5)
- Redis (14)
- Dubbo (1)
- Memcached (5)
- Netty (56)
- Mina (34)
- NIO (51)
- JUC (53)
- Spring (13)
- Mybatis (17)
- MySQL (21)
- JDBC (12)
- C3P0 (5)
- Tomcat (13)
- SLF4J-log4j (9)
- P6Spy (4)
- Quartz (12)
- Zabbix (7)
- JAVA (9)
- Linux (15)
- HTML (9)
- Lucene (0)
- JS (2)
- WebService (1)
- Maven (4)
- Oracle&MSSQL (14)
- iText (11)
- Development Tools (8)
- UTILS (4)
- LIFE (8)
最新评论
-
Donald_Draper:
Donald_Draper 写道刘落落cici 写道能给我发一 ...
DatagramChannelImpl 解析三(多播) -
Donald_Draper:
刘落落cici 写道能给我发一份这个类的源码吗Datagram ...
DatagramChannelImpl 解析三(多播) -
lyfyouyun:
请问楼主,执行消息发送的时候,报错:Transport sch ...
ActiveMQ连接工厂、连接详解 -
ezlhq:
关于 PollArrayWrapper 状态含义猜测:参考 S ...
WindowsSelectorImpl解析一(FdMap,PollArrayWrapper) -
flyfeifei66:
打算使用xmemcache作为memcache的客户端,由于x ...
Memcached分布式客户端(Xmemcached)
Executor接口的定义:http://donald-draper.iteye.com/blog/2365625
ExecutorService接口定义:http://donald-draper.iteye.com/blog/2365738
Future接口定义:http://donald-draper.iteye.com/blog/2365798
FutureTask解析:http://donald-draper.iteye.com/blog/2365980
CompletionService接口定义:http://donald-draper.iteye.com/blog/2366239
ExecutorCompletionService解析:http://donald-draper.iteye.com/blog/2366254
AbstractExecutorService解析:http://donald-draper.iteye.com/blog/2366348
ScheduledExecutorService接口定义:http://donald-draper.iteye.com/blog/2366436
ThreadPoolExecutor解析一(核心线程池数量、线程池状态等) :
http://donald-draper.iteye.com/blog/2366934
上一篇文章我们讲到线程池ThreadPoolExecutor的java doc使用说明和变量的定义,先回顾一下:
ThreadPoolExecutor的变量主要有核心线程池数量corePoolSize和最大线程池数量maximumPoolSize,即在当前任务线程数大于核心线程数量时,是否(allowCoreThreadTimeOut)允许空闲任务线程等,保活keepAliveTime时间,等待新任务的到来。一个线程工厂ThreadFactory用于创建任务线程,一个拒绝任务处理器RejectedExecutionHandler,默认的拒绝任务策略为AbortPolicy,抛出运行时异常,当然还有直接丢弃策略DiscardPolicy,丢弃旧的任务DiscardOldestPolicy,还有调用者执行任务策略CallerRunsPolicy。上面的变量为volatile,以便线程池执行操作时,可以使用最新的变量。
一个阻塞的任务队列final BlockingQueue<Runnable> workQueue,阻塞队列可以为Linked,Array,Delay,SynchronousQueue等阻塞类型,具体可以根据场景选择。默认为LinkedBlockingQueue队列,一般判断队列是否为空,用isEmpty方法,LinkedBlockingQueue一般用于任务相互之间独立,没有交叉,可独立执行。如果用SynchronousQueue,则可用poll方法判断,同步队列一般用于任务之间有依赖的关系的场景,一个任务执行依赖于另一个任务的结果。DelayQueue队列用于定时任务。ArrayBlockingQueue队列一般可以用于
资源有限情况,可以避免资源被耗尽。一个AtomicInteger的ctl用于包装线程状态runState和任务线程数workerCount;低29位保存任务线程数,高两位用于存储线程池状态,线程池状态已用有四种RUNNING,SHUTDOWN ,STOP,TIDYING ,TERMINATED。
RUNNING:接受新的任务,处理队列任务;
SHUTDOWN:不在接受新的任务,处理队列任务;
STOP:不在接受新任务,不处理队列任务,中断正在执行的任务线程;
TIDYING:所有的任务已经结束,任务线程为0,线程转换到TIDYING;
TERMINATED:线程池已结束,即terminated()方法执行完。
线程的状态是可以数字化比较的。
一个任务线程集final HashSet<Worker> workers,largestPoolSize记录线程池的最大任务线程数,completedTaskCount为完成任务计数器,在任务线程结束时,更新。一个可重入锁mainLock,用于保护非线程安全的变量如workers,largestPoolSize,completedTaskCount。
一个等待线程池结束条件termination,用于控制超时等待线程池关闭。
先来看ThreadPoolExecutor的几种构造方法
有了上面的构造ThreadPoolExecutor,我们来看一下默认的线程工厂和拒绝策略。
先看拒绝策略:
//ThreadPoolExecutor
AbortPolicy其实为ThreadPoolExecutor一个静态内部类。
我们既然看到拒绝策略把其他几个也一起看一下:
直接丢弃策略DiscardPolicy:
丢弃旧的任务DiscardOldestPolicy:
调用者执行任务CallerRunsPolicy:
上面所讲的四种丢弃策略都是ThreadPoolExecutor的内部类,分别为默认的线程池拒绝任务处理策略AbortPolicy,直接抛出RejectedExecutionException;直接丢弃策略DiscardPolicy,丢弃旧的任务DiscardOldestPolicy,调用者执行任务CallerRunsPolicy。
DiscardOldestPolicy和CallerRunsPolicy都是在线程池没关闭时,策略才生效,否则关闭直接丢弃任务。
再看默认的线程工厂
//Executors
//Executors
DefaultThreadFactory为Executors的内部类,
//Thread
再来看执行任务的工作线程Worker:
来看Worker的run方法:
//ThreadPoolExecutor
runWorker有几点要关注:
1. //执行前工作
2.//执行后工作
3.
//处理任务完成后的工作
分别来看每一点:
1. //执行前工作
beforeExecute(wt, task);
2.//执行后工作
3.
从现在开始我们把以前的任务线程一律改为工作线程,下面的工作线程及我们前面所说的任务线程。
//处理任务完成后的工作
本想以后再看,现在看了吧
我们再来看看tryTerminate
来看尝试结束线程池的这一点,
//如果工作线程不为空,则中断空闲工作线程
从上可以看出尝试结束线程池所做的工作为,首先检查线程池运行状态如果为运行中,关闭但任务队列不为空,或线程池工作线程为0,任务队列为空,则直接返回;否则查看工作线程是否为0,不为0,则根据onlyOne参数确定中断多少空闲线程,如果onlyOne为true,中断一个,否则中断所有空闲线程。
总结:
默认的线程池拒绝任务处理策略AbortPolicy,直接抛出RejectedExecutionException;直接丢弃策略DiscardPolicy,丢弃旧的任务DiscardOldestPolicy,调用者执行任务CallerRunsPolicy。DiscardOldestPolicy和CallerRunsPolicy都是在线程池没关闭时,策略才生效,否则关闭直接丢弃任务。
拒绝策略都为ThreadPoolExecutor的内部类。
默认的线程工厂DefaultThreadFactory为Executors的内部类, 用于创建线程,工厂创建分组相同的线程,交由执行器执行。如果有java.lang.SecurityManager,则用System#getSecurityManager线程组,否则用调用者的线程组。创建的新线程为非守护模式,优先级在 MIN_PRIORITY和MAX_PRIORITY之间,默认为NORM_PRIORITY。可以通过Thread#getName获取线程name,默认为pool-N-thread-M,N为线程池编号,M为线程编号。
Worker包装了任务线程,主要是为了维护中断控制状态和其他次要状态记录,及任务的执行。Worker同时继承了AQS,在任务线程执行前lock,任务执行完unlock。加锁的目的主要是保护任务线程的执行,线程池唤醒一个任务线程等待任务,而不是中断当前正在执行任务的线程去执行任务。Worker使用了一个 非重入互质锁,而不是ReentrantLock,这样做的目的是以防在任务执行的过程,线程池控制方法的改变,对任务线程执行的影响,比如setCorePoolSize方法。另外为了防止任务线程在实际执行前被中断,我们初始化锁状态为-1,在runWorker方法中,我们会清除它。runWorker执行任务时,首先释放锁,此时锁打开,允许中断,如果线程池正在stop,确保线程池已中断,否则
做执行前工作,执行任务,做执行后工作,如果任务被中断,则工作线程数量减1;
如果任务完成,则更新完成任务数量,从工作任务集中移除工作线程,尝试结束线程池。
尝试结束线程池,首先检查线程池运行状态如果为运行中,关闭但任务队列不为空,
或线程池工作线程为0,任务队列为空,则直接返回;否则查看工作线程是否为0,不为0,则根据onlyOne参数确定中断多少空闲线程,如果onlyOne为true,中断一个,否则中断所有空闲线程。
ThreadPoolExecutor解析三(线程池执行提交任务):
http://donald-draper.iteye.com/blog/2367199
//RejectedExecutionHandler
//ThreadFactory
//Thread 线程未捕捉异常处理器
ExecutorService接口定义:http://donald-draper.iteye.com/blog/2365738
Future接口定义:http://donald-draper.iteye.com/blog/2365798
FutureTask解析:http://donald-draper.iteye.com/blog/2365980
CompletionService接口定义:http://donald-draper.iteye.com/blog/2366239
ExecutorCompletionService解析:http://donald-draper.iteye.com/blog/2366254
AbstractExecutorService解析:http://donald-draper.iteye.com/blog/2366348
ScheduledExecutorService接口定义:http://donald-draper.iteye.com/blog/2366436
ThreadPoolExecutor解析一(核心线程池数量、线程池状态等) :
http://donald-draper.iteye.com/blog/2366934
上一篇文章我们讲到线程池ThreadPoolExecutor的java doc使用说明和变量的定义,先回顾一下:
ThreadPoolExecutor的变量主要有核心线程池数量corePoolSize和最大线程池数量maximumPoolSize,即在当前任务线程数大于核心线程数量时,是否(allowCoreThreadTimeOut)允许空闲任务线程等,保活keepAliveTime时间,等待新任务的到来。一个线程工厂ThreadFactory用于创建任务线程,一个拒绝任务处理器RejectedExecutionHandler,默认的拒绝任务策略为AbortPolicy,抛出运行时异常,当然还有直接丢弃策略DiscardPolicy,丢弃旧的任务DiscardOldestPolicy,还有调用者执行任务策略CallerRunsPolicy。上面的变量为volatile,以便线程池执行操作时,可以使用最新的变量。
一个阻塞的任务队列final BlockingQueue<Runnable> workQueue,阻塞队列可以为Linked,Array,Delay,SynchronousQueue等阻塞类型,具体可以根据场景选择。默认为LinkedBlockingQueue队列,一般判断队列是否为空,用isEmpty方法,LinkedBlockingQueue一般用于任务相互之间独立,没有交叉,可独立执行。如果用SynchronousQueue,则可用poll方法判断,同步队列一般用于任务之间有依赖的关系的场景,一个任务执行依赖于另一个任务的结果。DelayQueue队列用于定时任务。ArrayBlockingQueue队列一般可以用于
资源有限情况,可以避免资源被耗尽。一个AtomicInteger的ctl用于包装线程状态runState和任务线程数workerCount;低29位保存任务线程数,高两位用于存储线程池状态,线程池状态已用有四种RUNNING,SHUTDOWN ,STOP,TIDYING ,TERMINATED。
RUNNING:接受新的任务,处理队列任务;
SHUTDOWN:不在接受新的任务,处理队列任务;
STOP:不在接受新任务,不处理队列任务,中断正在执行的任务线程;
TIDYING:所有的任务已经结束,任务线程为0,线程转换到TIDYING;
TERMINATED:线程池已结束,即terminated()方法执行完。
线程的状态是可以数字化比较的。
一个任务线程集final HashSet<Worker> workers,largestPoolSize记录线程池的最大任务线程数,completedTaskCount为完成任务计数器,在任务线程结束时,更新。一个可重入锁mainLock,用于保护非线程安全的变量如workers,largestPoolSize,completedTaskCount。
一个等待线程池结束条件termination,用于控制超时等待线程池关闭。
先来看ThreadPoolExecutor的几种构造方法
// Public constructors and methods /** * Creates a new {@code ThreadPoolExecutor} with the given initial * parameters and default thread factory and rejected execution handler. * It may be more convenient to use one of the {@link Executors} factory * methods instead of this general purpose constructor. * 根据核心线程池数量,最大线程池数量,空闲任务保活时间,任务队列构造ThreadPoolExecutor, 线程池工厂默认为Executors.defaultThreadFactory(),拒绝策略为默认的AbortPolicy。 * @param corePoolSize the number of threads to keep in the pool, even * if they are idle, unless {@code allowCoreThreadTimeOut} is set * @param maximumPoolSize the maximum number of threads to allow in the * pool * @param keepAliveTime when the number of threads is greater than * the core, this is the maximum time that excess idle threads * will wait for new tasks before terminating. * @param unit the time unit for the {@code keepAliveTime} argument * @param workQueue the queue to use for holding tasks before they are * executed. This queue will hold only the {@code Runnable} * tasks submitted by the {@code execute} method. * @throws IllegalArgumentException if one of the following holds:<br> * {@code corePoolSize < 0}<br> * {@code keepAliveTime < 0}<br> * {@code maximumPoolSize <= 0}<br> * {@code maximumPoolSize < corePoolSize} * @throws NullPointerException if {@code workQueue} is null */ public ThreadPoolExecutor(int corePoolSize, int maximumPoolSize, long keepAliveTime, TimeUnit unit, BlockingQueue<Runnable> workQueue) { this(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue, Executors.defaultThreadFactory(), defaultHandler); } /** * Creates a new {@code ThreadPoolExecutor} with the given initial * parameters and default rejected execution handler. * 根据核心线程池数量,最大线程池数量,空闲任务保活时间,任务队列,线程池工厂构造ThreadPoolExecutor, 拒绝策略为默认的AbortPolicy。 * @param corePoolSize the number of threads to keep in the pool, even * if they are idle, unless {@code allowCoreThreadTimeOut} is set * @param maximumPoolSize the maximum number of threads to allow in the * pool * @param keepAliveTime when the number of threads is greater than * the core, this is the maximum time that excess idle threads * will wait for new tasks before terminating. * @param unit the time unit for the {@code keepAliveTime} argument * @param workQueue the queue to use for holding tasks before they are * executed. This queue will hold only the {@code Runnable} * tasks submitted by the {@code execute} method. * @param threadFactory the factory to use when the executor * creates a new thread * @throws IllegalArgumentException if one of the following holds:<br> * {@code corePoolSize < 0}<br> * {@code keepAliveTime < 0}<br> * {@code maximumPoolSize <= 0}<br> * {@code maximumPoolSize < corePoolSize} * @throws NullPointerException if {@code workQueue} * or {@code threadFactory} is null */ public ThreadPoolExecutor(int corePoolSize, int maximumPoolSize, long keepAliveTime, TimeUnit unit, BlockingQueue<Runnable> workQueue, ThreadFactory threadFactory) { this(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue, threadFactory, defaultHandler); } /** * Creates a new {@code ThreadPoolExecutor} with the given initial * parameters and default thread factory. * 根据核心线程池数量,最大线程池数量,空闲任务保活时间,任务队列,拒绝策略构造ThreadPoolExecutor, 线程池工厂默认为Executors.defaultThreadFactory()。 * @param corePoolSize the number of threads to keep in the pool, even * if they are idle, unless {@code allowCoreThreadTimeOut} is set * @param maximumPoolSize the maximum number of threads to allow in the * pool * @param keepAliveTime when the number of threads is greater than * the core, this is the maximum time that excess idle threads * will wait for new tasks before terminating. * @param unit the time unit for the {@code keepAliveTime} argument * @param workQueue the queue to use for holding tasks before they are * executed. This queue will hold only the {@code Runnable} * tasks submitted by the {@code execute} method. * @param handler the handler to use when execution is blocked * because the thread bounds and queue capacities are reached * @throws IllegalArgumentException if one of the following holds:<br> * {@code corePoolSize < 0}<br> * {@code keepAliveTime < 0}<br> * {@code maximumPoolSize <= 0}<br> * {@code maximumPoolSize < corePoolSize} * @throws NullPointerException if {@code workQueue} * or {@code handler} is null */ public ThreadPoolExecutor(int corePoolSize, int maximumPoolSize, long keepAliveTime, TimeUnit unit, BlockingQueue<Runnable> workQueue, RejectedExecutionHandler handler) { this(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue, Executors.defaultThreadFactory(), handler); } /** * Creates a new {@code ThreadPoolExecutor} with the given initial * parameters. * 根据核心线程池数量,最大线程池数量,空闲任务保活时间,任务队列,线程池工厂,拒绝策略构造ThreadPoolExecutor。 * @param corePoolSize the number of threads to keep in the pool, even * if they are idle, unless {@code allowCoreThreadTimeOut} is set * @param maximumPoolSize the maximum number of threads to allow in the * pool * @param keepAliveTime when the number of threads is greater than * the core, this is the maximum time that excess idle threads * will wait for new tasks before terminating. * @param unit the time unit for the {@code keepAliveTime} argument * @param workQueue the queue to use for holding tasks before they are * executed. This queue will hold only the {@code Runnable} * tasks submitted by the {@code execute} method. * @param threadFactory the factory to use when the executor * creates a new thread * @param handler the handler to use when execution is blocked * because the thread bounds and queue capacities are reached * @throws IllegalArgumentException if one of the following holds:<br> * {@code corePoolSize < 0}<br> * {@code keepAliveTime < 0}<br> * {@code maximumPoolSize <= 0}<br> * {@code maximumPoolSize < corePoolSize} * @throws NullPointerException if {@code workQueue} * or {@code threadFactory} or {@code handler} is null */ 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.corePoolSize = corePoolSize; this.maximumPoolSize = maximumPoolSize; this.workQueue = workQueue; this.keepAliveTime = unit.toNanos(keepAliveTime); this.threadFactory = threadFactory; this.handler = handler; }
有了上面的构造ThreadPoolExecutor,我们来看一下默认的线程工厂和拒绝策略。
先看拒绝策略:
//ThreadPoolExecutor
/** * A handler for rejected tasks that throws a 默认的线程池拒绝任务处理策略 * {@code RejectedExecutionException}. */ public static class AbortPolicy implements RejectedExecutionHandler { /** * Creates an {@code AbortPolicy}. */ public AbortPolicy() { } /** * Always throws RejectedExecutionException. * 拒绝策略为直接抛出RejectedExecutionException * @param r the runnable task requested to be executed * @param e the executor attempting to execute this task * @throws RejectedExecutionException always. */ public void rejectedExecution(Runnable r, ThreadPoolExecutor e) { throw new RejectedExecutionException("Task " + r.toString() + " rejected from " + e.toString()); } }
AbortPolicy其实为ThreadPoolExecutor一个静态内部类。
我们既然看到拒绝策略把其他几个也一起看一下:
直接丢弃策略DiscardPolicy:
/** * A handler for rejected tasks that silently discards the * rejected task. */ public static class DiscardPolicy implements RejectedExecutionHandler { /** * Creates a {@code DiscardPolicy}. */ public DiscardPolicy() { } /** * Does nothing, which has the effect of discarding task r. * 不做任何工作直接丢弃任务 * @param r the runnable task requested to be executed * @param e the executor attempting to execute this task */ public void rejectedExecution(Runnable r, ThreadPoolExecutor e) { } }
丢弃旧的任务DiscardOldestPolicy:
/** * A handler for rejected tasks that discards the oldest unhandled * request and then retries {@code execute}, unless the executor * is shut down, in which case the task is discarded. */ public static class DiscardOldestPolicy implements RejectedExecutionHandler { /** * Creates a {@code DiscardOldestPolicy} for the given executor. */ public DiscardOldestPolicy() { } /** * Obtains and ignores the next task that the executor * would otherwise execute, if one is immediately available, * and then retries execution of task r, unless the executor * is shut down, in which case task r is instead discarded. * 获取并忽略执行器下一个将会执行的任务,执行新提交的任务,如果有任务线程立刻可用, 则重新尝试执行任务r,如果线程池已关闭,则丢弃任务。 * @param r the runnable task requested to be executed * @param e the executor attempting to execute this task */ public void rejectedExecution(Runnable r, ThreadPoolExecutor e) { if (!e.isShutdown()) { //如果线程中没关闭,任务队列头任务出队列,并不执行,执行的是刚提交的任务 r e.getQueue().poll(); e.execute(r); } } }
调用者执行任务CallerRunsPolicy:
/* Predefined RejectedExecutionHandlers */ /** * A handler for rejected tasks that runs the rejected task * directly in the calling thread of the {@code execute} method, * unless the executor has been shut down, in which case the task * is discarded. */ public static class CallerRunsPolicy implements RejectedExecutionHandler { /** * Creates a {@code CallerRunsPolicy}. */ public CallerRunsPolicy() { } /** * Executes task r in the caller's thread, unless the executor * has been shut down, in which case the task is discarded. * 调用者执行任务,如果线程池已关闭,则丢弃任务 * @param r the runnable task requested to be executed * @param e the executor attempting to execute this task */ public void rejectedExecution(Runnable r, ThreadPoolExecutor e) { if (!e.isShutdown()) { //如果线程池没关闭,则直接执行任务 r.run(); } } }
上面所讲的四种丢弃策略都是ThreadPoolExecutor的内部类,分别为默认的线程池拒绝任务处理策略AbortPolicy,直接抛出RejectedExecutionException;直接丢弃策略DiscardPolicy,丢弃旧的任务DiscardOldestPolicy,调用者执行任务CallerRunsPolicy。
DiscardOldestPolicy和CallerRunsPolicy都是在线程池没关闭时,策略才生效,否则关闭直接丢弃任务。
再看默认的线程工厂
//Executors
Executors.defaultThreadFactory()。
/** * Returns a default thread factory used to create new threads. * This factory creates all new threads used by an Executor in the * same {@link ThreadGroup}. If there is a {@link * java.lang.SecurityManager}, it uses the group of {@link * System#getSecurityManager}, else the group of the thread * invoking this <tt>defaultThreadFactory</tt> method. Each new * thread is created as a non-daemon thread with priority set to * the smaller of <tt>Thread.NORM_PRIORITY</tt> and the maximum * priority permitted in the thread group. New threads have names * accessible via {@link Thread#getName} of * [i]pool-N-thread-M[/i], where [i]N[/i] is the sequence * number of this factory, and [i]M[/i] is the sequence number * of the thread created by this factory. 返回一个默认的线程工厂用于创建线程,工厂创建分组相同的线程,交由执行器执行。 如果有java.lang.SecurityManager,则用System#getSecurityManager线程组, 否则用调用者的线程组。创建的新线程为非守护模式,优先级在 MIN_PRIORITY和MAX_PRIORITY之间, 默认为NORM_PRIORITY。可以通过Thread#getName获取线程name,默认为pool-N-thread-M, N为线程池编号,M为线程编号。 * @return a thread factory */ public static ThreadFactory defaultThreadFactory() { return new DefaultThreadFactory(); }
//Executors
/** * The default thread factory,默认的线程工厂 */ static class DefaultThreadFactory implements ThreadFactory { private static final AtomicInteger poolNumber = new AtomicInteger(1);//线程池编号 private final ThreadGroup group;//线程分组 private final AtomicInteger threadNumber = new AtomicInteger(1);//线程编号 private final String namePrefix; DefaultThreadFactory() { SecurityManager s = System.getSecurityManager(); group = (s != null) ? s.getThreadGroup() : Thread.currentThread().getThreadGroup(); //构造一个线程池工厂,线程池编号自增1 namePrefix = "pool-" + poolNumber.getAndIncrement() + "-thread-"; } //创建新线程,线程名为pool-1-thread-1, 1为当前线程工程的线程 public Thread newThread(Runnable r) { //根据分组,Runnable,名字创建线程,线程编号自增1 Thread t = new Thread(group, r, namePrefix + threadNumber.getAndIncrement(), 0); if (t.isDaemon()) t.setDaemon(false); if (t.getPriority() != Thread.NORM_PRIORITY) t.setPriority(Thread.NORM_PRIORITY); return t; } }
DefaultThreadFactory为Executors的内部类,
//Thread
public final static int MIN_PRIORITY = 1;//最小优先级 public final static int NORM_PRIORITY = 5;//默认 public final static int MAX_PRIORITY = 10;//最大优先级
再来看执行任务的工作线程Worker:
/** * Class Worker mainly maintains interrupt control state for * threads running tasks, along with other minor bookkeeping. * This class opportunistically extends AbstractQueuedSynchronizer * to simplify acquiring and releasing a lock surrounding each * task execution. This protects against interrupts that are * intended to wake up a worker thread waiting for a task from * instead interrupting a task being run. We implement a simple * non-reentrant mutual exclusion lock rather than use * ReentrantLock because we do not want worker tasks to be able to * reacquire the lock when they invoke pool control methods like * setCorePoolSize. Additionally, to suppress interrupts until * the thread actually starts running tasks, we initialize lock * state to a negative value, and clear it upon start (in * runWorker). Worker主要为任务线程维护中断控制状态和其他次要状态记录。 Worker简单实现了 AQS在任务线程执行前lock,任务执行完unlock。加锁的目的主要是保护任务线程的执行, 线程池唤醒一个任务线程等待任务,而不是中断当前正在执行任务的线程去执行任务。我们使用了一个 非重入互质锁,而不是ReentrantLock,这样做的目的是以防在任务执行的过程,线程池控制 方法的改变,对任务线程执行的影响,比如setCorePoolSize方法。另外为了防止任务线程在实际 执行前被中断,我们初始化锁状态为-1,在runWorker方法中,我们会清除它。 */ private final class Worker extends AbstractQueuedSynchronizer implements Runnable { /** * This class will never be serialized, but we provide a * serialVersionUID to suppress a javac warning. */ private static final long serialVersionUID = 6138294804551838833L; /** Thread this worker is running in. Null if factory fails. */ final Thread thread;//任务线程 /** Initial task to run. Possibly null. */ Runnable firstTask; //任务 /** Per-thread task counter */ volatile long completedTasks; //线程完成的任务计数 /** * Creates with given first task and thread from ThreadFactory. 根据给定的任务,用线程工厂创建任务线程 * @param firstTask the first task (null if none) * Worker(Runnable firstTask) { setState(-1); // inhibit interrupts until runWorker this.firstTask = firstTask; this.thread = getThreadFactory().newThread(this); } /** Delegates main run loop to outer runWorker */ public void run() { //运行线程 runWorker(this); } // Lock methods // // The value 0 represents the unlocked state. 开锁状态 // The value 1 represents the locked state.闭锁状态 protected boolean isHeldExclusively() { return getState() != 0;//是否持有锁 } //尝试获取锁 protected boolean tryAcquire(int unused) { if (compareAndSetState(0, 1)) { setExclusiveOwnerThread(Thread.currentThread()); return true; } return false; } //尝试释放锁 protected boolean tryRelease(int unused) { setExclusiveOwnerThread(null); setState(0); return true; } public void lock() { acquire(1); }//加锁 public boolean tryLock() { return tryAcquire(1); }//尝试加锁 public void unlock() { release(1); }//解锁 public boolean isLocked() { return isHeldExclusively(); }//是否锁住 //如果任务正在执行,则中断 void interruptIfStarted() { Thread t; if (getState() >= 0 && (t = thread) != null && !t.isInterrupted()) { try { //如果锁处于关闭状态,且任务线程不为null,非出于中断状态,则中断当前线程 t.interrupt(); } catch (SecurityException ignore) { } } } }
来看Worker的run方法:
/** Delegates main run loop to outer runWorker */ public void run() { //运行线程 runWorker(this); }
//ThreadPoolExecutor
/** * Main worker run loop. Repeatedly gets tasks from queue and * executes them, while coping with a number of issues: * 工作线程运行的主要方法。重复从任务队列获取任务,并执行,主要有一下几点注意事项: * 1. We may start out with an initial task, in which case we * don't need to get the first one. Otherwise, as long as pool is * running, we get tasks from getTask. If it returns null then the * worker exits due to changed pool state or configuration * parameters. Other exits result from exception throws in * external code, in which case completedAbruptly holds, which * usually leads processWorkerExit to replace this thread. * 1.我们也许在开始时有一个初始化的任务,这是不需要从任务队列获取任务。 只要线程池在运行,则从getTask方法获取任务,由于线程状态或配置参数的 改变,获取任务为null时,则工作线程退出。如果执行过程中抛出异常,则 调用processWorkerExit方法,取代当前线程 * 2. Before running any task, the lock is acquired to prevent * other pool interrupts while the task is executing, and * clearInterruptsForTaskRun called to ensure that unless pool is * stopping, this thread does not have its interrupt set. * 2.在执行任务前,获取lock,防止任务在执行的过程中被其他线程池中断。 clearInterruptsForTaskRun清除正在运行任务的中断位,在线程没有关闭的情况下。 * 3. Each task run is preceded by a call to beforeExecute, which * might throw an exception, in which case we cause thread to die * (breaking loop with completedAbruptly true) without processing * the task. * 3.在任务执行前,调用beforeExecute,这个可能会抛出异常,将导致工作线程 没有处理任务的情况下死掉。 * 4. Assuming beforeExecute completes normally, we run the task, * gathering any of its thrown exceptions to send to * afterExecute. We separately handle RuntimeException, Error * (both of which the specs guarantee that we trap) and arbitrary * Throwables. Because we cannot rethrow Throwables within * Runnable.run, we wrap them within Errors on the way out (to the * thread's UncaughtExceptionHandler). Any thrown exception also * conservatively causes thread to die. * 4.如果beforeExecute正常完成,执行任务,收集运行中的异常,交由afterExecute处理。 我们独立地处理运行异常,错误和任意异常。由于我们不能在Runnable.run中重新抛异常, 我们可以将异常包装在UncaughtExceptionHandler中。任何异常可能导致线程死掉 * 5. After task.run completes, we call afterExecute, which may * also throw an exception, which will also cause thread to * die. According to JLS Sec 14.20, this exception is the one that * will be in effect even if task.run throws. * 5.在任务执行完,我们可以调用afterExecute,方法可能抛出异常导致工作线程死掉。 根据JLS,Java语言规范,即使这个异常有执行任务抛出,已有可能导致线程死掉。 * The net effect of the exception mechanics is that afterExecute * and the thread's UncaughtExceptionHandler have as accurate * information as we can provide about any problems encountered by * user code. afterExecute方法和线程的UncaughtExceptionHandler,给我们提供了针对错误的便利。 * * @param w the worker */ final void runWorker(Worker w) { Thread wt = Thread.currentThread();//当前线程 Runnable task = w.firstTask;//工作线程任务 w.firstTask = null; //任务线程的锁状态默认为-1,此时解锁+1,变为0,即锁打开状态,允许中断,在任务未执行之前,不允许中断。 w.unlock(); // allow interrupts, boolean completedAbruptly = true;//完成后是否可以中断 try { while (task != null || (task = getTask()) != null) { w.lock(); // If pool is stopping, ensure thread is interrupted; // if not, ensure thread is not interrupted. This // requires a recheck in second case to deal with // shutdownNow race while clearing interrupt //如果线程池正在Stop,则确保线程中断; //如果非处于Stop之后的状态,则判断是否中断,如果中断则判断线程池是否已关闭 //如果线程池正在关闭,但没有中断,则中断线程池 if ((runStateAtLeast(ctl.get(), STOP) || (Thread.interrupted() && runStateAtLeast(ctl.get(), STOP))) && !wt.isInterrupted()) wt.interrupt(); try { //执行前工作 beforeExecute(wt, task); Throwable thrown = null; try { //执行任务 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; //任务线程完成任务数量加1,释放锁 w.completedTasks++; w.unlock(); } } //任务已执行完不可以中断 completedAbruptly = false; } finally { //处理任务完成后的工作 processWorkerExit(w, completedAbruptly); } }
runWorker有几点要关注:
1. //执行前工作
beforeExecute(wt, task);
2.//执行后工作
afterExecute(task, thrown);
3.
//处理任务完成后的工作
processWorkerExit(w, completedAbruptly);
分别来看每一点:
1. //执行前工作
beforeExecute(wt, task);
/** * Method invoked prior to executing the given Runnable in the * given thread. This method is invoked by thread {@code t} that * will execute task {@code r}, and may be used to re-initialize * ThreadLocals, or to perform logging. * * <p>This implementation does nothing, but may be customized in * subclasses. Note: To properly nest multiple overridings, subclasses * should generally invoke {@code super.beforeExecute} at the end of * this method. * 待子类扩展 * @param t the thread that will run task {@code r} * @param r the task that will be executed */ protected void beforeExecute(Thread t, Runnable r) { }
2.//执行后工作
afterExecute(task, thrown);
/** * Method invoked upon completion of execution of the given Runnable. * This method is invoked by the thread that executed the task. If * non-null, the Throwable is the uncaught {@code RuntimeException} * or {@code Error} that caused execution to terminate abruptly. * 处理任务完成工作,和在任务执行过程中,未捕捉的异常等 * <p>This implementation does nothing, but may be customized in * subclasses. Note: To properly nest multiple overridings, subclasses * should generally invoke {@code super.afterExecute} at the * beginning of this method. * * <p><b>Note:</b> When actions are enclosed in tasks (such as * {@link FutureTask}) either explicitly or via methods such as * {@code submit}, these task objects catch and maintain * computational exceptions, and so they do not cause abrupt * termination, and the internal exceptions are [i]not[/i] * passed to this method. If you would like to trap both kinds of * failures in this method, you can further probe for such cases, * as in this sample subclass that prints either the direct cause * or the underlying exception if a task has been aborted: * * <pre> {@code * class ExtendedExecutor extends ThreadPoolExecutor { * // ... * protected void afterExecute(Runnable r, Throwable t) { * super.afterExecute(r, t); * if (t == null && r instanceof Future<?>) { * try { * Object result = ((Future<?>) r).get(); * } catch (CancellationException ce) { * t = ce; * } catch (ExecutionException ee) { * t = ee.getCause(); * } catch (InterruptedException ie) { * Thread.currentThread().interrupt(); // ignore/reset * } * } * if (t != null) * System.out.println(t); * } * }}</pre> * 待子类扩展 * @param r the runnable that has completed * @param t the exception that caused termination, or null if * execution completed normally */ protected void afterExecute(Runnable r, Throwable t) { }
3.
从现在开始我们把以前的任务线程一律改为工作线程,下面的工作线程及我们前面所说的任务线程。
//处理任务完成后的工作
processWorkerExit(w, completedAbruptly);
/** * Performs cleanup and bookkeeping for a dying worker. Called * only from worker threads. Unless completedAbruptly is set, * assumes that workerCount has already been adjusted to account * for exit. This method removes thread from worker set, and * possibly terminates the pool or replaces the worker if either * it exited due to user task exception or if fewer than * corePoolSize workers are running or queue is non-empty but * there are no workers. * * @param w the worker * @param completedAbruptly if the worker died due to user exception */ private void processWorkerExit(Worker w, boolean completedAbruptly) { if (completedAbruptly) // If abrupt, then workerCount wasn't adjusted //如果任务线程被中断,则工作线程数量减1 decrementWorkerCount(); final ReentrantLock mainLock = this.mainLock; mainLock.lock(); try { //增加任务完成数量 completedTaskCount += w.completedTasks; //从工作线程集中,移除工作线程 workers.remove(w); } finally { mainLock.unlock(); } //尝试结束线程池,这个我们后面再说 //这个实在当前线程池正在关闭,但任务队列不为空,当前工作线程执行完任务,则尝试关闭空闲线程,更新线程池状态 tryTerminate(); int c = ctl.get(); //如果线程处于运行中 if (runStateLessThan(c, STOP)) { //如果线程已完成 if (!completedAbruptly) { //获取当前核心线程数 int min = allowCoreThreadTimeOut ? 0 : corePoolSize; if (min == 0 && ! workQueue.isEmpty()) //如果允许空闲工作线程等待任务,且任务队列不为空,则min为1 min = 1; //如果当前工作线程大于核心工作线程数,则返回 if (workerCountOf(c) >= min) return; // replacement not needed } addWorker(null, false); } }
本想以后再看,现在看了吧
我们再来看看tryTerminate
/** * Transitions to TERMINATED state if either (SHUTDOWN and pool * and queue empty) or (STOP and pool empty). If otherwise * eligible to terminate but workerCount is nonzero, interrupts an * idle worker to ensure that shutdown signals propagate. This * method must be called following any action that might make * termination possible -- reducing worker count or removing tasks * from the queue during shutdown. The method is non-private to * allow access from ScheduledThreadPoolExecutor. */ final void tryTerminate() { //自旋尝试关闭线程池 for (;;) { int c = ctl.get(); //如果线程池正在运行,或关闭但队列不为空,则返回 if (isRunning(c) || runStateAtLeast(c, TIDYING) || (runStateOf(c) == SHUTDOWN && ! workQueue.isEmpty())) return; if (workerCountOf(c) != 0) { // Eligible to terminate //如果工作线程不为空,则中断空闲工作线程 interruptIdleWorkers(ONLY_ONE); return; } final ReentrantLock mainLock = this.mainLock; mainLock.lock(); try { //线程池已关闭,任务队列为空,工作线程为0,更新线程池状态为TIDYING if (ctl.compareAndSet(c, ctlOf(TIDYING, 0))) { try { //执行结束工作 terminated(); } finally { //线程池已结束 ctl.set(ctlOf(TERMINATED, 0)); //唤醒等待线程池结束的线程 termination.signalAll(); } return; } } finally { mainLock.unlock(); } // else retry on failed CAS } }
来看尝试结束线程池的这一点,
//如果工作线程不为空,则中断空闲工作线程
interruptIdleWorkers(ONLY_ONE); /** * Interrupts threads that might be waiting for tasks (as * indicated by not being locked) so they can check for * termination or configuration changes. Ignores * SecurityExceptions (in which case some threads may remain * uninterrupted). * 中断等待任务的空闲非锁住状态的工作线程 * @param onlyOne If true, interrupt at most one worker. This is * called only from tryTerminate when termination is otherwise * enabled but there are still other workers. In this case, at * most one waiting worker is interrupted to propagate shutdown * signals in case all threads are currently waiting. * Interrupting any arbitrary thread ensures that newly arriving * workers since shutdown began will also eventually exit. * To guarantee eventual termination, it suffices to always * interrupt only one idle worker, but shutdown() interrupts all * idle workers so that redundant workers exit promptly, not * waiting for a straggler task to finish. 如果onlyOne为true以为着,只中断最多一个空闲工作线程,这个在关闭线程池时, 调用或关闭的过程中,工作线程完成任务调用。 */ private void interruptIdleWorkers(boolean onlyOne) { final ReentrantLock mainLock = this.mainLock; mainLock.lock(); try { for (Worker w : workers) { //遍历工作线程集 Thread t = w.thread; if (!t.isInterrupted() && w.tryLock()) {//锁打开说明,工作线程空闲 try { //如果工作线程非中断,且空闲,尝试获取锁,获取锁成功,则中断工作线程 t.interrupt(); } catch (SecurityException ignore) { } finally { w.unlock(); } } if (onlyOne) //如果是只中断一个空闲线程,则结束本次中断空闲线程任务 break; } } finally { mainLock.unlock(); } }
从上可以看出尝试结束线程池所做的工作为,首先检查线程池运行状态如果为运行中,关闭但任务队列不为空,或线程池工作线程为0,任务队列为空,则直接返回;否则查看工作线程是否为0,不为0,则根据onlyOne参数确定中断多少空闲线程,如果onlyOne为true,中断一个,否则中断所有空闲线程。
总结:
默认的线程池拒绝任务处理策略AbortPolicy,直接抛出RejectedExecutionException;直接丢弃策略DiscardPolicy,丢弃旧的任务DiscardOldestPolicy,调用者执行任务CallerRunsPolicy。DiscardOldestPolicy和CallerRunsPolicy都是在线程池没关闭时,策略才生效,否则关闭直接丢弃任务。
拒绝策略都为ThreadPoolExecutor的内部类。
默认的线程工厂DefaultThreadFactory为Executors的内部类, 用于创建线程,工厂创建分组相同的线程,交由执行器执行。如果有java.lang.SecurityManager,则用System#getSecurityManager线程组,否则用调用者的线程组。创建的新线程为非守护模式,优先级在 MIN_PRIORITY和MAX_PRIORITY之间,默认为NORM_PRIORITY。可以通过Thread#getName获取线程name,默认为pool-N-thread-M,N为线程池编号,M为线程编号。
Worker包装了任务线程,主要是为了维护中断控制状态和其他次要状态记录,及任务的执行。Worker同时继承了AQS,在任务线程执行前lock,任务执行完unlock。加锁的目的主要是保护任务线程的执行,线程池唤醒一个任务线程等待任务,而不是中断当前正在执行任务的线程去执行任务。Worker使用了一个 非重入互质锁,而不是ReentrantLock,这样做的目的是以防在任务执行的过程,线程池控制方法的改变,对任务线程执行的影响,比如setCorePoolSize方法。另外为了防止任务线程在实际执行前被中断,我们初始化锁状态为-1,在runWorker方法中,我们会清除它。runWorker执行任务时,首先释放锁,此时锁打开,允许中断,如果线程池正在stop,确保线程池已中断,否则
做执行前工作,执行任务,做执行后工作,如果任务被中断,则工作线程数量减1;
如果任务完成,则更新完成任务数量,从工作任务集中移除工作线程,尝试结束线程池。
尝试结束线程池,首先检查线程池运行状态如果为运行中,关闭但任务队列不为空,
或线程池工作线程为0,任务队列为空,则直接返回;否则查看工作线程是否为0,不为0,则根据onlyOne参数确定中断多少空闲线程,如果onlyOne为true,中断一个,否则中断所有空闲线程。
ThreadPoolExecutor解析三(线程池执行提交任务):
http://donald-draper.iteye.com/blog/2367199
//RejectedExecutionHandler
package java.util.concurrent; /** * A handler for tasks that cannot be executed by a {@link ThreadPoolExecutor}. * * @since 1.5 * @author Doug Lea */ public interface RejectedExecutionHandler { /** * Method that may be invoked by a {@link ThreadPoolExecutor} when * {@link ThreadPoolExecutor#execute execute} cannot accept a * task. This may occur when no more threads or queue slots are * available because their bounds would be exceeded, or upon * shutdown of the Executor. * 当线程池执行器不能接受任务时被调用。这种情况的发生可能由于没有任务线程可用, 或任务队列已满,或执行器关闭。 * <p>In the absence of other alternatives, the method may throw * an unchecked {@link RejectedExecutionException}, which will be * propagated to the caller of {@code execute}. * 在一些实现策略中,可能抛出一个RejectedExecutionException,也可能有调用者去执行任务。 * @param r the runnable task requested to be executed * @param executor the executor attempting to execute this task * @throws RejectedExecutionException if there is no remedy */ void rejectedExecution(Runnable r, ThreadPoolExecutor executor); }
//ThreadFactory
package java.util.concurrent; /** * An object that creates new threads on demand. Using thread factories * removes hardwiring of calls to {@link Thread#Thread(Runnable) new Thread}, * enabling applications to use special thread subclasses, priorities, etc. * 根据需求创建线程。用线程工厂取代Thread#Thread(Runnable)常见新线程,使应用可以使用 一些特殊的线程子类,比如具有优先级线程等 * <p> * The simplest implementation of this interface is just: * <pre> * class SimpleThreadFactory implements ThreadFactory { * public Thread newThread(Runnable r) { * return new Thread(r); * } * } * </pre> * * The {@link Executors#defaultThreadFactory} method provides a more * useful simple implementation, that sets the created thread context * to known values before returning it. * @since 1.5 * @author Doug Lea */ public interface ThreadFactory { /** * Constructs a new {@code Thread}. Implementations may also initialize * priority, name, daemon status, {@code ThreadGroup}, etc. * 创建一个线程初始化线程的优先级,名字,是否为守候,线程组 * @param r a runnable to be executed by new thread instance * @return constructed thread, or {@code null} if the request to * create a thread is rejected */ Thread newThread(Runnable r); }
//Thread 线程未捕捉异常处理器
// Added in JSR-166 /** * Interface for handlers invoked when a <tt>Thread</tt> abruptly * terminates due to an uncaught exception. * <p>When a thread is about to terminate due to an uncaught exception * the Java Virtual Machine will query the thread for its * <tt>UncaughtExceptionHandler</tt> using * {@link #getUncaughtExceptionHandler} and will invoke the handler's * <tt>uncaughtException</tt> method, passing the thread and the * exception as arguments. * If a thread has not had its <tt>UncaughtExceptionHandler</tt> * explicitly set, then its <tt>ThreadGroup</tt> object acts as its * <tt>UncaughtExceptionHandler</tt>. If the <tt>ThreadGroup</tt> object * has no * special requirements for dealing with the exception, it can forward * the invocation to the {@linkplain #getDefaultUncaughtExceptionHandler * default uncaught exception handler}. * * @see #setDefaultUncaughtExceptionHandler * @see #setUncaughtExceptionHandler * @see ThreadGroup#uncaughtException * @since 1.5 */ public interface UncaughtExceptionHandler { /** * Method invoked when the given thread terminates due to the * given uncaught exception. * <p>Any exception thrown by this method will be ignored by the * Java Virtual Machine. * @param t the thread * @param e the exception */ void uncaughtException(Thread t, Throwable e); } // null unless explicitly set private volatile UncaughtExceptionHandler uncaughtExceptionHandler; // null unless explicitly set private static volatile UncaughtExceptionHandler defaultUncaughtExceptionHandler;
发表评论
-
Executors解析
2017-04-07 14:38 1242ThreadPoolExecutor解析一(核心线程池数量、线 ... -
ScheduledThreadPoolExecutor解析三(关闭线程池)
2017-04-06 20:52 4448ScheduledThreadPoolExecutor解析一( ... -
ScheduledThreadPoolExecutor解析二(任务调度)
2017-04-06 12:56 2115ScheduledThreadPoolExecutor解析一( ... -
ScheduledThreadPoolExecutor解析一(调度任务,任务队列)
2017-04-04 22:59 4985Executor接口的定义:http://donald-dra ... -
ThreadPoolExecutor解析四(线程池关闭)
2017-04-03 23:02 9095Executor接口的定义:http: ... -
ThreadPoolExecutor解析三(线程池执行提交任务)
2017-04-03 12:06 6077Executor接口的定义:http://donald-dra ... -
ThreadPoolExecutor解析一(核心线程池数量、线程池状态等)
2017-03-31 22:01 20512Executor接口的定义:http://donald-dra ... -
ScheduledExecutorService接口定义
2017-03-29 12:53 1500Executor接口的定义:http://donald-dra ... -
AbstractExecutorService解析
2017-03-29 08:27 1070Executor接口的定义:http: ... -
ExecutorCompletionService解析
2017-03-28 14:27 1585Executor接口的定义:http://donald-dra ... -
CompletionService接口定义
2017-03-28 12:39 1059Executor接口的定义:http://donald-dra ... -
FutureTask解析
2017-03-27 12:59 1323package java.util.concurrent; ... -
Future接口定义
2017-03-26 09:40 1189/* * Written by Doug Lea with ... -
ExecutorService接口定义
2017-03-25 22:14 1157Executor接口的定义:http://donald-dra ... -
Executor接口的定义
2017-03-24 23:24 1671package java.util.concurrent; ... -
简单测试线程池拒绝执行任务策略
2017-03-24 22:37 2022线程池多余任务的拒绝执行策略有四中,分别是直接丢弃任务Disc ... -
JAVA集合类简单综述
2017-03-23 22:51 918Queue接口定义:http://donald-draper. ... -
DelayQueue解析
2017-03-23 11:00 1730Queue接口定义:http://donald-draper. ... -
SynchronousQueue解析下-TransferQueue
2017-03-22 22:20 2132Queue接口定义:http://donald-draper. ... -
SynchronousQueue解析上-TransferStack
2017-03-21 22:08 3046Queue接口定义:http://donald-draper. ...
相关推荐
- **ReentrantLock mainLock**:用于在更新内部数据(如线程数量、运行状态、工作线程增减等)时进行同步。 - **Condition termination**:支持awaitTermination方法的等待条件。 - **Worker类**:线程池的核心...
线程池原理-ThreadPoolExecutor源码解析 1.构造方法及参数 2.阻塞对列: BlockingQueue 3.线程工厂: DefaultThreadFactory 4.拒绝策略: RejectedExecutionHandler 5.执行线程 Executor
为了解决主线程的性能问题,开发者通常会在后台创建工作线程来执行耗时任务,例如网络请求、数据库操作、大量计算等。常见的工作线程实现方式有:子线程、IntentService、HandlerThread、AsyncTask以及使用线程池...
除了上述创建线程的方式,Java还提供了Callable和Future接口,用于创建带返回值的线程,以及ExecutorService和ThreadPoolExecutor等高级线程管理工具,以更灵活地控制线程的执行和管理。 在实际开发中,理解和掌握...
Java中的ExecutorService接口及其实现类如ThreadPoolExecutor,允许我们自定义线程池的核心参数,如核心线程数、最大线程数、线程存活时间等。线程池能有效控制并发量,防止过多线程导致系统资源耗尽,同时提供定时...
如果当前线程池中的线程数量已经达到 `maximumPoolSize`,那么将会根据 `RejectedExecutionHandler` 的拒绝策略来处理新任务。 4. **线程的销毁**:当线程池中的线程数量大于 `corePoolSize` 时,那些超过 `...
`ThreadPoolExecutor` 通过灵活配置核心线程数量、最大线程数量、任务队列等参数,实现了对线程的有效管理和任务的高效执行。通过对 `execute()` 方法的深入分析,我们能够更全面地理解 `ThreadPoolExecutor` 在实际...
本文将深入解析ThreadPoolExecutor的execute()方法执行流程,以帮助我们理解线程池的工作原理。 当一个任务被提交到线程池,线程池的执行策略主要分为四步: 1. 首先,线程池会检查当前的核心线程数是否已达到设定...
《JAVA线程(第三版)》是一本深入探讨Java多线程编程的权威书籍,针对Java线程的管理和优化提供了详尽的解析。线程在现代计算机编程中扮演着至关重要的角色,尤其是在并发处理和高性能应用中。Java以其强大的线程...
- **ThreadPoolExecutor**:最常见的线程池实现,可以配置核心线程数、最大线程数、线程存活时间等参数。 - **Executors工厂方法**:提供newFixedThreadPool、newSingleThreadExecutor、newCachedThreadPool和...
9. **线程间通信**:线程间通信通过共享数据或使用wait/notify机制协调工作。例如,生产者消费者模型、哲学家就餐问题等。 10. **线程优先级**:Java线程有优先级,但具体效果依赖于操作系统。线程优先级高的线程在...
- ThreadPoolExecutor:最常见的线程池实现,可以定制线程池大小、队列容量、拒绝策略等。 - Executors:静态工厂方法,提供预定义的线程池,如FixedThreadPool、SingleThreadExecutor、CachedThreadPool等。 6. ...
Java提供了`Thread`类来创建线程,`ExecutorService`和`ThreadPoolExecutor`等工具来管理线程池,以及`Future`接口来获取线程执行的结果。开发者通常会定义一个下载任务类,包含文件块的URL和本地保存路径,然后创建...
线程池的核心组件包括工作队列、线程池大小、核心线程数、最大线程数、空闲线程存活时间等。在多线程下载中,线程池用于调度多个下载任务,确保系统资源的有效利用。例如,Java中的`ExecutorService`接口及其实现如`...
Java的`ExecutorService`和`ThreadPoolExecutor`类实现了线程池功能,允许开发者根据需求调整线程池的大小、任务队列和拒绝策略等参数,以优化性能和资源利用。 最后,我们提一下死锁,这是一种线程间相互等待对方...
在实际开发中,结合`ThreadPoolExecutor`的其他方法,如`execute()`、`shutdown()`、`shutdownNow()`等,可以灵活地控制线程池的工作流程,以应对不同的并发场景。总之,掌握线程池源码不仅有助于优化代码,还能在...
1. **线程池管理**:为了有效控制和管理线程,源码可能会使用Java的`ExecutorService`或者Python的`concurrent.futures.ThreadPoolExecutor`等线程池实现。线程池可以预先创建一定数量的线程,当有任务需要执行时,...
### Java/Android线程使用深度解析 在计算机科学领域,线程与进程是核心概念,尤其是在多任务操作系统中。本文将深入探讨Java/Android环境下的线程管理,包括线程的创建、线程池的利用及`ThreadHandler`的运用等...
本文档将全面解析多线程的基础知识,从简单到复杂,帮助开发者深入理解并掌握这一核心技术。 一、多线程基础 1.1 线程与进程 线程是操作系统分配处理器时间的基本单元,而进程则是资源分配的基本单位。一个进程...
5. **并发集合**:讲解线程安全的集合类,如`ConcurrentHashMap`、`CopyOnWriteArrayList`和`BlockingQueue`等,以及它们在多线程环境下的使用策略。 6. **死锁与活锁**:分析线程间的同步问题可能导致的死锁和活锁...