- Callable 接口
public interface Callable<V> {
V call() throws Exception;
}
- Runnable 接口
public interface Runnable {
public abstract void run();
}
- Future接口
public interface Future<V> {
// 用来取消任务，如果取消任务成功则返回true，如果取消任务失败则返回false
boolean cancel(boolean mayInterruptIfRunning);
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();
}
- FutureTask类
public class FutureTask<V> implements RunnableFuture<V> {
// 线程执行状态
private volatile int state;
private static final int NEW = 0;
private static final int COMPLETING = 1;
private static final int NORMAL = 2;
private static final int EXCEPTIONAL = 3;
private static final int CANCELLED = 4;
private static final int INTERRUPTING = 5;
private static final int INTERRUPTED = 6;
private Callable<V> callable;
private Object outcome;
private volatile Thread runner;
/**
* 显然，waiters保留了在FutureTask上的等待线程列表，设计链表的意义？ 因为Future的get()/get(timeout)在task处于非完成状态时是需要阻塞等待的，
* 如果多个线程进行get操作，显然需要一个链表/队列来维护这些等待线程，这就是waiters的意义所在。
*/
private volatile WaitNode waiters;
static final class WaitNode {
volatile Thread thread;
volatile WaitNode next;
WaitNode() { thread = Thread.currentThread(); }
}
public FutureTask(Callable<V> callable) {
if (callable == null) throw new NullPointerException();
this.callable = callable;
this.state = NEW;
}
// 把Runnable转换为Callable接口
public FutureTask(Runnable runnable, V result) {
// Executors.callable(Runnable task, T result) 是用来把Runnable包装成Callable<T>的。
// 包装出来的Callable<T>只能返回传入的result
this.callable = Executors.callable(runnable, result);
this.state = NEW;
}
// Executors.callable方法源码
//====================================================================
static final class RunnableAdapter<T> implements Callable<T> {
final Runnable task;
final T result;
RunnableAdapter(Runnable task, T result) {
this.task = task;
this.result = result;
}
public T call() {
task.run();
return result;
}
}
public static <T> Callable<T> callable(Runnable task, T result) {
if (task == null)
throw new NullPointerException();
return new RunnableAdapter<T>(task, result);
}
//====================================================================
// End
public void run() {
try {
Callable<V> c = callable;
if (c != null && state == NEW) {
V result;
boolean ran;
try {
result = c.call();
ran = true;
} catch (Throwable ex) {
result = null;
ran = false;
setException(ex); // 保存异常，当调用get方法的时候抛出异常
}
if (ran)
set(result);
}
} finally {
runner = null;
int s = state;
if (s >= INTERRUPTING)
handlePossibleCancellationInterrupt(s);
}
}
protected void set(V v) {
// 尝试将statec状态从new修改为COMPLETING状态
if (UNSAFE.compareAndSwapInt(this, stateOffset, NEW, COMPLETING)) {
// 修改成功，将结果保存到outcome中，再将state状态修改为normal状态
outcome = v;
UNSAFE.putOrderedInt(this, stateOffset, NORMAL); // final state
finishCompletion(); // 唤醒等待队列中所有的线程
}
}
// LockSupport.unpark(thread)唤醒线程
private void finishCompletion() {
// assert state > COMPLETING;
for (WaitNode q; (q = waiters) != null;) {
// 将waiters置为null
if (UNSAFE.compareAndSwapObject(this, waitersOffset, q, null)) {
// 循环遍历等待队列，唤醒所有等待线程
for (;;) {
Thread t = q.thread;
if (t != null) {
q.thread = null;
LockSupport.unpark(t);
}
WaitNode next = q.next;
if (next == null)
break;
q.next = null; // unlink to help gc
q = next;
}
break;
}
}
done();
callable = null; // to reduce footprint
}
public V get() throws InterruptedException, ExecutionException {
int s = state;
if (s <= COMPLETING)
s = awaitDone(false, 0L); // 阻塞当前线程
return report(s);
}
private V report(int s) throws ExecutionException {
Object x = outcome;
if (s == NORMAL)
return (V)x;
if (s >= CANCELLED)
throw new CancellationException();
throw new ExecutionException((Throwable)x);
}
// 加入等待队列过程
private int awaitDone(boolean timed, long nanos)
throws InterruptedException {
final long deadline = timed ? System.nanoTime() + nanos : 0L;
WaitNode q = null;
boolean queued = false;
for (;;) {
if (Thread.interrupted()) {
removeWaiter(q);
throw new InterruptedException();
}
int s = state;
if (s > COMPLETING) {
if (q != null)
q.thread = null;
return s;
}
else if (s == COMPLETING) // cannot time out yet
Thread.yield();
else if (q == null)
q = new WaitNode();
else if (!queued)
// 入队
queued = UNSAFE.compareAndSwapObject(this, waitersOffset,
q.next = waiters, q);
else if (timed) {
nanos = deadline - System.nanoTime();
if (nanos <= 0L) {
removeWaiter(q);
return state;
}
LockSupport.parkNanos(this, nanos);
} else {
LockSupport.park(this);// 阻塞当先线程
}
}
}
// Unsafe mechanics
private static final sun.misc.Unsafe UNSAFE;
private static final long stateOffset;
private static final long runnerOffset;
private static final long waitersOffset;
static {
try {
UNSAFE = sun.misc.Unsafe.getUnsafe();
Class<?> k = FutureTask.class;
// 获取object对象的属性Field的偏移量
stateOffset = UNSAFE.objectFieldOffset
(k.getDeclaredField("state"));
runnerOffset = UNSAFE.objectFieldOffset
(k.getDeclaredField("runner"));
waitersOffset = UNSAFE.objectFieldOffset
(k.getDeclaredField("waiters"));
} catch (Exception e) {
throw new Error(e);
}
}
}
【补充】sun.misc.Unsafe类
public class VO
{
public int a = 0;
public long b = 0;
public static String c= "123";
public static Object d= null;
public static int e = 100;
}
1.获取实例字段的偏移地址
// 获取实例字段的偏移地址,偏移最小的那个字段(仅挨着头部)就是对象头的大小
System.out.println(unsafe.objectFieldOffset(VO.class.getDeclaredField("a")));
System.out.println(unsafe.objectFieldOffset(VO.class.getDeclaredField("b")));
// fieldOffset与objectFieldOffset功能一样,fieldOffset是过时方法,最好不要再使用
System.out.println(unsafe.fieldOffset(VO.class.getDeclaredField("b")));
2.获取数组的头部大小和元素大小
// 数组第一个元素的偏移地址,即数组头占用的字节数
int[] intarr = new int[0];
System.out.println(unsafe.arrayBaseOffset(intarr.getClass()));
// 数组中每个元素占用的大小
System.out.println(unsafe.arrayIndexScale(intarr.getClass()));
Unsafe类中有很多以BASE_OFFSET结尾的常量，比如ARRAY_INT_BASE_OFFSET等，这些常量值是通过arrayBaseOffset方法得到的。arrayBaseOffset方法是一个本地方法，可以获取数组第一个元素的偏移地址。Unsafe类中还有很多以INDEX_SCALE结尾的常量，比如 ARRAY_INT_INDEX_SCALE 等，这些常量值是通过arrayIndexScale方法得到的。将arrayBaseOffset与arrayIndexScale配合使用，可以定位数组中每个元素在内存中的位置。
3.获取类的静态字段偏移
// 获取类的静态字段偏地址
System.out.println(unsafe.staticFieldOffset(VO.class.getDeclaredField("c")));
System.out.println(unsafe.staticFieldOffset(VO.class.getDeclaredField("d")));
// 获取静态字段的起始地址,通过起始地址和偏移地址,就可以获取静态字段的值了
// 只不过静态字段的起始地址,类型不是long,而是Object类型
Object base1 = unsafe.staticFieldBase(VO.class);
Object base2 = unsafe.staticFieldBase(VO.class.getDeclaredField("d"));
System.out.println(base1==base2);//true
4.获取操作系统的位数
// Report the size in bytes of a native pointer.
// 返回4或8,代表是32位还是64位操作系统。
System.out.println(unsafe.addressSize());
// 返回32或64,获取操作系统是32位还是64位
System.out.println(System.getProperty("sun.arch.data.model"));
通过上面的几段代码，我们可以成功获取类中各个字段的偏移地址，这跟jol工具的输出结果和我们的结论是一致的。有了字段的偏移地址，在加上对象的起始地，我们就能够通过Unsafe直接获取字段的值了。
5.读取对象实例字段的值
//获取实例字段的属性值
VO vo = new VO();
vo.a = 10000;
long aoffset = unsafe.objectFieldOffset(VO.class.getDeclaredField("a"));
int va = unsafe.getInt(vo, aoffset);
System.out.println("va="+va);
6.获取静态字段的属性值
VO.e = 1024;
Field sField = VO.class.getDeclaredField("e");
Object base = unsafe.staticFieldBase(sField);
long offset = unsafe.staticFieldOffset(sField);
System.out.println(unsafe.getInt(base, offset));//1024
- Executor 接口
public interface Executor {
void execute(Runnable command);
}
- ExecutorService 接口
public interface ExecutorService extends Executor {
void shutdown();
<T> Future<T> submit(Callable<T> task);
<T> Future<T> submit(Runnable task, T result);
Future<?> submit(Runnable task);
}
- AbstractExecutorService类
public abstract class AbstractExecutorService implements ExecutorService {
protected <T> RunnableFuture<T> newTaskFor(Runnable runnable, T value) {
return new FutureTask<T>(runnable, value);
}
protected <T> RunnableFuture<T> newTaskFor(Callable<T> callable) {
return new FutureTask<T>(callable);
}
public Future<?> submit(Runnable task) {
if (task == null) throw new NullPointerException();
RunnableFuture<Void> ftask = newTaskFor(task, null);
execute(ftask);
return ftask;
}
public <T> Future<T> submit(Runnable task, T result) {
if (task == null) throw new NullPointerException();
RunnableFuture<T> ftask = newTaskFor(task, result);
execute(ftask);
return ftask;
}
public <T> Future<T> submit(Callable<T> task) {
if (task == null) throw new NullPointerException();
RunnableFuture<T> ftask = newTaskFor(task);
execute(ftask);
return ftask;
}
}
- ThreadPoolExecutor类
一、ThreadPoolExecutor中的重要成员变量
- 1、AtomicInteger ctl
AtomicInteger类型的ctl代表了ThreadPoolExecutor中的控制状态，它是一个复核类型的成员变量，是一个原子整数，借助高低位包装了两个概念：
（1）workerCount：线程池中当前活动的线程数量，占据ctl的低29位；
（2）runState：线程池运行状态，占据ctl的高3位，有RUNNING、SHUTDOWN、STOP、TIDYING、TERMINATED五种状态。
AtomicInteger ctl的定义如下：
private final AtomicInteger ctl = new AtomicInteger(ctlOf(RUNNING, 0));
先说下workerCount：线程池中当前活动的线程数量，它占据ctl的低29位，这样，每当活跃线程数增加或减少时，ctl直接做相应数目的增减即可，十分方便。而ThreadPoolExecutor中COUNT_BITS就代表了workerCount所占位数，定义如下：
private static final int COUNT_BITS = Integer.SIZE - 3;
在Java中，一个int占据32位，而COUNT_BITS的结果不言而喻，Integer大小32减去3，就是29；另外，既然workerCount代表了线程池中当前活动的线程数量，那么
它肯定有个上下限阈值，下限很明显就是0，上限呢？ThreadPoolExecutor中CAPACITY就代表了workerCount的上限，它是ThreadPoolExecutor中理论上的最大活跃线程数，其定义如下：
private static final int CAPACITY = (1 << COUNT_BITS) - 1;
运算过程为1左移29位，也就是00000000 00000000 00000000 00000001 --> 001 0000 00000000 00000000 00000000，再减去1的话，就是 000 11111 11111111 11111111 11111111，前三位代表线程池运行状态runState，所以这里workerCount的理论最大值就应该是29个1，即536870911；
既然workerCount作为其中一个概念复合在AtomicInteger ctl中，那么ThreadPoolExecutor理应提供从AtomicInteger ctl中解析出workerCount的方法，如下：
private static int workerCountOf(int c) { return c & CAPACITY; }
计算逻辑很简单，传入的c代表的是ctl的值，即高3位为线程池运行状态runState，低29位为线程池中当前活动的线程数量workerCount，将其与CAPACITY进行与操作&，也就是与000 11111 11111111 11111111 11111111进行与操作，c的前三位通过与000进行与操作，无论c前三位为何值，最终都会变成000，也就是舍弃前三位的值，而c的低29位与29个1进行与操作，c的低29位还是会保持原值，这样就从AtomicInteger ctl中解析出了workerCount的值。
接下来，我们再看下runState：线程池运行状态，它占据ctl的高3位，有RUNNING、SHUTDOWN、STOP、TIDYING、TERMINATED五种状态。我们先分别解释下这五种状态：
（1）RUNNING：接受新任务，并处理队列任务
private static final int RUNNING = -1 << COUNT_BITS;
-1在Java底层是由32个1表示的，左移29位的话，即111 00000 00000000 00000000 00000000，也就是低29位全部为0，高3位全部为1的话，表示RUNNING状态，即-536870912；
（2）SHUTDOWN：不接受新任务，但会处理队列任务
private static final int SHUTDOWN = 0 << COUNT_BITS;
0在Java底层是由32个0表示的，无论左移多少位，还是32个0，即000 00000 00000000 00000000 00000000，也就是低29位全部为0，高3位全部为0的话，表示SHUTDOWN状态，即0；
（3）STOP：不接受新任务，不会处理队列任务，而且会中断正在处理过程中的任务
private static final int STOP = 1 << COUNT_BITS;
1在Java底层是由前面的31个0和1个1组成的，左移29位的话，即001 00000 00000000 00000000 00000000，也就是低29位全部为0，高3位为001的话，表示STOP状态，即536870912；
（4）TIDYING：所有的任务已结束，workerCount为0，线程过渡到TIDYING状态，将会执行terminated()钩子方法
private static final int TIDYING = 2 << COUNT_BITS;
2在Java底层是由前面的30个0和1个10组成的，左移29位的话，即010 00000 00000000 00000000 00000000，也就是低29位全部为0，高3位为010的话，表示TIDYING状态，即1073741824；
（5）TERMINATED：terminated()方法已经完成
private static final int TERMINATED = 3 << COUNT_BITS;
2在Java底层是由前面的30个0和1个11组成的，左移29位的话，即011 00000 00000000 00000000 00000000，也就是低29位全部为0，高3位为011的话，表示TERMINATED状态，即1610612736；
由上面我们可以得知，运行状态的值按照RUNNING-->SHUTDOWN-->STOP-->TIDYING-->TERMINATED顺序值是递增的，这些值之间的数值顺序很重要。随着时间的推移，运行状态单调增加，但是不需要经过每个状态。那么，可能存在的线程池状态的转换是什么呢？如下：
（1）RUNNING -> SHUTDOWN：调用shutdownNow()方法后，或者线程池实现了finalize方法，在里面调用了shutdown方法，即隐式调用；
（2）(RUNNING or SHUTDOWN) -> STOP：调用shutdownNow()方法后；
（3）SHUTDOWN -> TIDYING：线程池和队列均为空时；
（4）STOP -> TIDYING：线程池为空时；
（5）TIDYING -> TERMINATED：terminated()钩子方法完成时。
我们再来看下是实现获取运行状态的runStateOf()方法，代码如下：
private static int runStateOf(int c) { return c & ~CAPACITY; }
是按位取反的意思，CAPACITY表示的是高位的3个0，和低位的29个1，而CAPACITY则表示高位的3个1，2低位的9个0，然后再与入参c执行按位与操作，即高3位保持原样，低29位全部设置为0，也就获取了线程池的运行状态runState。
最后，我们再看下原子变量ctl的初始化方法ctlOf()，代码如下：
private static int ctlOf(int rs, int wc) { return rs | wc; }
很简单，传入的rs表示线程池运行状态runState，其是高3位有值，低29位全部为0的int，而wc则代表线程池中有效线程的数量workerCount，其为高3位全部为0，而低29位有值得int，将runState和workerCount做或操作|处理，即用runState的高3位，workerCount的低29位填充的数字，而默认传入的runState、workerCount分别为RUNNING和0。
2、BlockingQueue<Runnable> workQueue
workQueue是用于持有任务并将其转换成工作线程worker的队列；
3、HashSet<Worker> workers
workers是包含线程池中所有工作线程worker的集合，仅仅当拥有mainLock锁时才能访问它；
4、long completedTaskCount
completedTaskCount是已完成任务的计数器，只有在worker线程的终止，仅仅当拥有mainLock锁时才能访问它；
public class ThreadPoolExecutor extends AbstractExecutorService {
private final AtomicInteger ctl = new AtomicInteger(ctlOf(RUNNING, 0));
/**
* （1）workerCount：线程池中当前活动的线程数量，占据ctl的低29位；
* （2）runState：线程池运行状态，占据ctl的高3位，有RUNNING、SHUTDOWN、STOP、TIDYING、TERMINATED五种状态。
*
*/
private static int runStateOf(int c) { return c & ~CAPACITY; }
private static int workerCountOf(int c) { return c & CAPACITY; }
private static int ctlOf(int rs, int wc) { return rs | wc; }
// 多线程添加Runnable的时候需要加锁
private final ReentrantLock mainLock = new ReentrantLock();
private final Condition termination = mainLock.newCondition();
//
private final HashSet<Worker> workers = new HashSet<>();
// 等待队列
private final BlockingQueue<Runnable> workQueue;
// 默认值为false，如果为false，core线程在空闲时依然存活；如果为true，则core线程等待工作，直到时间超时至keepAliveTime
private volatile boolean allowCoreThreadTimeOut;
//空闲线程等待工作的超时时间（纳秒），即空闲线程存活时间
private volatile long keepAliveTime;
//核心线程池大小，保持存活的工作线程的最小数目，当小于corePoolSize时，会直接启动新的一个线程来处理任务，而不管线程池中是否有空闲线程；
private volatile int corePoolSize;
private volatile int maximumPoolSize; // 线程池中线程的最大数量
private volatile ThreadFactory threadFactory;
private volatile RejectedExecutionHandler handler;
private static final RejectedExecutionHandler defaultHandler =
new AbortPolicy();
public ThreadPoolExecutor(int corePoolSize,
int maximumPoolSize,
long keepAliveTime,
TimeUnit unit,
BlockingQueue<Runnable> workQueue) {
this(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue,
Executors.defaultThreadFactory(), defaultHandler);
}
// 往线程池里面添加Runnable
public void execute(Runnable command) {
int c = ctl.get();
if (workerCountOf(c) < corePoolSize) {
if (addWorker(command, true))
return;
c = ctl.get();
}
// 将command添加到阻塞队列，并不会改变核心线程池数量
if (isRunning(c) && workQueue.offer(command)) {
int recheck = ctl.get();
// 如果在我们添加到阻塞队列之后，状态不是RUNNING状态，会将当前任务从阻塞队列移除，并拒绝这次任务
if (!isRunning(recheck) && remove(command))
reject(command);
//这种情况是由于corePoolSize允许为0，当corePoolSize为0时，第一次会运行到这步，并添加线程到线程池中。当corePoolSize等于0时，会相当于只在核心线程池中添加一个线程用于消费阻塞队列的任务，这里也会在2.1结合不同阻塞队列说
else if (workerCountOf(recheck) == 0)
addWorker(null, false);
}
else if (!addWorker(command, false))
reject(command);
}
private boolean addWorker(Runnable firstTask, boolean core) {
retry:
for (;;) {
int c = ctl.get();
int rs = runStateOf(c);
if (rs >= SHUTDOWN &&
! (rs == SHUTDOWN &&
firstTask == null &&
! workQueue.isEmpty()))
return false;
for (;;) {
int wc = workerCountOf(c);
if (wc >= CAPACITY ||
wc >= (core ? corePoolSize : maximumPoolSize))
return false;
if (compareAndIncrementWorkerCount(c))
break retry;
c = ctl.get();
if (runStateOf(c) != rs)
continue retry;
// else CAS failed due to workerCount change; retry inner loop
}
}
// 满足所有的检测条件之后，才能执行以下代码
boolean workerStarted = false;
boolean workerAdded = false;
Worker w = null;
try {
w = new Worker(firstTask); // 创建Worker(Worker extends AbstractQueuedSynchronizer implements Runnable)
final Thread t = w.thread;
if (t != null) {
final ReentrantLock mainLock = this.mainLock;
mainLock.lock();
try {
int rs = runStateOf(ctl.get());
if (rs < SHUTDOWN ||
(rs == SHUTDOWN && firstTask == null)) {
if (t.isAlive())
throw new IllegalThreadStateException();
workers.add(w);
int s = workers.size();
if (s > largestPoolSize)
largestPoolSize = s;
workerAdded = true;
}
} finally {
mainLock.unlock();
}
if (workerAdded) {
t.start(); // 开始启动线程
workerStarted = true;
}
}
} finally {
if (!workerStarted)
addWorkerFailed(w);
}
return workerStarted;
}
final void runWorker(Worker w) {
Thread wt = Thread.currentThread();
Runnable task = w.firstTask;
w.firstTask = null;
w.unlock(); // allow interrupts
boolean completedAbruptly = true;
try {
while (task != null || (task = getTask()) != null) {
w.lock();
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;
w.completedTasks++;
w.unlock();
}
}
completedAbruptly = false;
} finally {
processWorkerExit(w, completedAbruptly);
}
}
// 从阻塞队列取Runnable
private Runnable getTask() {
boolean timedOut = false;]
for (;;) {
int c = ctl.get();
int rs = runStateOf(c);
if (rs >= SHUTDOWN && (rs >= STOP || workQueue.isEmpty())) {
decrementWorkerCount();
return null;
}
int wc = workerCountOf(c);
boolean timed = allowCoreThreadTimeOut || wc > corePoolSize;
if ((wc > maximumPoolSize || (timed && timedOut))
&& (wc > 1 || workQueue.isEmpty())) {
if (compareAndDecrementWorkerCount(c))
return null;
continue;
}
try {
Runnable r = timed ?
workQueue.poll(keepAliveTime, TimeUnit.NANOSECONDS) :
workQueue.take();
if (r != null)
return r;
timedOut = true;
} catch (InterruptedException retry) {
timedOut = false;
}
}
}
private final class Worker
extends AbstractQueuedSynchronizer
implements Runnable
{
private static final long serialVersionUID = 6138294804551838833L;
final Thread thread;
Runnable firstTask;
volatile long completedTasks;
Worker(firstTask) {
setState(-1);
this.firstTask = firstTask;
this.thread = getThreadFactory().newThread(this);
}
public void run() {
runWorker(this);
}
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 {
t.interrupt();
} catch (SecurityException ignore) {
}
}
}
}
}
- Executors 工具类
public class Executors {
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>());
}
}
- 阻塞队列
10.1 LinkedBlockingQueue 默认大小为Integer.MAX_VALUE
LinkedBlockingQueue，当我们初始化的时候没有给他初始容量，那么这里，他每次offer都可以添加到我们的阻塞队列中，因为LinkedBlockingQueue是基于链表结构的无界阻塞队列。那么我们如果corePoolSize不是0，则相当于只有当前workers中只有CorePool，当workerCountOf(c) > corePoolSize的时候，我们只是向阻塞队列中添加任务，供之后线程消费，而不会再添加新的worker到workers了，所以这个时候的MaxPool和CorePool是一样大的，maxmumPoolSize参数也就没有了意义。如果corePoolSize是0，则相当于只有一个线程在线程池中，之后的任务都直接进入到阻塞队列
LinkedBlockingQueue赋予了初始化容量，那么我的理解是和ArrayBlockingQueue作用是一样的。当我们的数量达到了核心线程数，接下来会向阻塞队列中添加任务，当我们的阻塞队列也满了。则再创建新的worker加入到workers中，当达到最大线程数时，最后会reject。
当我们当前的线程池核心线程数大小小于corePoolSize的时候，每次都会创建新的woker来执行，当我们等于核心线程数的时候，如果这个时候存在空闲的worker，那么会直接使用空闲的worker执行，当没有空闲worker的时候会向阻塞队列中添加command
10.2 ArrayBlockingQueue 必须指定队列大小
10.3 PriorityBlockingQueue 基于优先级阻塞队列
10.4 SynchronousQueue
** 拒接执行策略 **
final void reject(Runnable command) {
handler.rejectedExecution(command, this);
}
public static class CallerRunsPolicy implements RejectedExecutionHandler {
public CallerRunsPolicy() { }
public void rejectedExecution(Runnable r, ThreadPoolExecutor e) {
if (!e.isShutdown()) {
r.run();
}
}
}
public static class AbortPolicy implements RejectedExecutionHandler {
public AbortPolicy() { }
public void rejectedExecution(Runnable r, ThreadPoolExecutor e) {
throw new RejectedExecutionException("Task " + r.toString() +
" rejected from " + e.toString());
}
}
public static class DiscardPolicy implements RejectedExecutionHandler {
public DiscardPolicy() { }
public void rejectedExecution(Runnable r, ThreadPoolExecutor e) {}
}
public static class DiscardOldestPolicy implements RejectedExecutionHandler {
public DiscardOldestPolicy() { }
public void rejectedExecution(Runnable r, ThreadPoolExecutor e) {
if (!e.isShutdown()) {
e.getQueue().poll();
e.execute(r);
}
}
}
【总结】
其中workQueue代表的是提交但未执行的队列，它是BlockingQueue接口的对象，用于存放Runable对象，主要分为以下几种类型：
直接提交的队列：SynchronousQueue队列，它是一个没有容量的队列，前面我有对其进行讲解，当线程池进行入队offer操作的时候，本身是无容量的，所以直接返回false，并没有保存下来，而是直接提交给线程来进行执行，如果没有空余的线程则执行拒绝策略。
有界的任务队列：可以使用ArrayBlockingQueue队列，因为它内部是基于数组来进行实现的，初始化时必须指定容量参数，当使用有界任务队列时，当有任务进行提交时，线程池的线程数量小于corePoolSize则创建新的线程来执行任务，当线程池的线程数量大于corePoolSize的时候，则将提交的任务放入到队列中，当提交的任务塞满队列后，如果线程池的线程数量没有超过maximumPoolSize，则创建新的线程执行任务，如果超过了maximumPoolSize则执行拒绝策略。
无界的任务队列：可以使用LinkedBlockingQueue队列，它内部是基于链表的形式，默认队列的长度是Integer.MAX_VALUE，也可以指定队列的长度，当队列满时进行阻塞操作，当然线程池中采用的是offer方法并不会阻塞线程，当队列满时则返回false，入队成功则则返回true，当使用LinkedBlockingQueue队列时，有任务提交到线程池时，如果线程池的数量小于corePoolSize，线程池会产生新的线程来执行任务，当线程池的线程数量大于corePoolSize时，则将提交的任务放入到队列中，等待执行任务的线程执行完之后进行消费队列中的任务，若后续仍有新的任务提交，而没有空闲的线程时，它会不断往队列中入队提交的任务，直到资源耗尽。
优先任务队列：t有限任务队列是带有执行优先级的队列，他可以使用PriorityBlockingQueue队列，可以控制任务的执行先后顺序，它是一个无界队列，该队列可以根据任务自身的优先级顺序先后执行，在确保性能的同时，也能有很好的质量保证。
【源码】AsyncTask<Params, Progress, Result> 源码浅尝
public abstract class AsyncTask<Params, Progress, Result> {
private static final int CPU_COUNT = Runtime.getRuntime().availableProcessors();
private static final int CORE_POOL_SIZE = Math.max(2, Math.min(CPU_COUNT - 1, 4));
private static final int MAXIMUM_POOL_SIZE = CPU_COUNT * 2 + 1;
private static final int KEEP_ALIVE_SECONDS = 30;
private static final BlockingQueue<Runnable> sPoolWorkQueue =
new LinkedBlockingQueue<Runnable>(128);
public static final Executor SERIAL_EXECUTOR = new SerialExecutor();
private static class SerialExecutor implements Executor {
final ArrayDeque<Runnable> mTasks = new ArrayDeque<Runnable>();
Runnable mActive;
public synchronized void execute(final Runnable r) {
mTasks.offer(new Runnable() {
public void run() {
try {
r.run();
} finally {
scheduleNext();
}
}
});
if (mActive == null) {
scheduleNext();
}
}
protected synchronized void scheduleNext() {
if ((mActive = mTasks.poll()) != null) {
THREAD_POOL_EXECUTOR.execute(mActive);
}
}
}
public static final Executor THREAD_POOL_EXECUTOR;
static {
ThreadPoolExecutor threadPoolExecutor = new ThreadPoolExecutor(
CORE_POOL_SIZE, MAXIMUM_POOL_SIZE, KEEP_ALIVE_SECONDS, TimeUnit.SECONDS,
sPoolWorkQueue, sThreadFactory);
threadPoolExecutor.allowCoreThreadTimeOut(true);
THREAD_POOL_EXECUTOR = threadPoolExecutor;
}
public AsyncTask(@Nullable Looper callbackLooper) {
mHandler = callbackLooper == null || callbackLooper == Looper.getMainLooper()
? getMainHandler()
: new Handler(callbackLooper);
mWorker = new WorkerRunnable<Params, Result>() {
public Result call() throws Exception {
mTaskInvoked.set(true);
Result result = null;
try {
Process.setThreadPriority(Process.THREAD_PRIORITY_BACKGROUND);
//noinspection unchecked
result = doInBackground(mParams);
Binder.flushPendingCommands();
} catch (Throwable tr) {
mCancelled.set(true);
throw tr;
} finally {
postResult(result);
}
return result;
}
};
mFuture = new FutureTask<Result>(mWorker) {
@Override
protected void done() {
try {
postResultIfNotInvoked(get());
} catch (InterruptedException e) {
android.util.Log.w(LOG_TAG, e);
} catch (ExecutionException e) {
throw new RuntimeException("An error occurred while executing doInBackground()",
e.getCause());
} catch (CancellationException e) {
postResultIfNotInvoked(null);
}
}
};
}
@MainThread
public final AsyncTask<Params, Progress, Result> execute(Params... params) {
return executeOnExecutor(sDefaultExecutor, params);
}
@MainThread
public final AsyncTask<Params, Progress, Result> executeOnExecutor(Executor exec,
Params... params) {
if (mStatus != Status.PENDING) {
switch (mStatus) {
case RUNNING:
throw new IllegalStateException("Cannot execute task:"
+ " the task is already running.");
case FINISHED:
throw new IllegalStateException("Cannot execute task:"
+ " the task has already been executed "
+ "(a task can be executed only once)");
}
}
mStatus = Status.RUNNING;
onPreExecute();
mWorker.mParams = params;
exec.execute(mFuture);
return this;
}
private Result postResult(Result result) {
@SuppressWarnings("unchecked")
Message message = getHandler().obtainMessage(MESSAGE_POST_RESULT,
new AsyncTaskResult<Result>(this, result));
message.sendToTarget();
return result;
}
@WorkerThread
protected abstract Result doInBackground(Params... params);
@MainThread
protected void onPreExecute() {}
@MainThread
protected void onPostExecute(Result result) {}
@MainThread
protected void onProgressUpdate(Progress... values) {}
}
