public ThreadPoolExecutor(int corePoolSize,

int maximumPoolSize,

long keepAliveTime,

TimeUnit timeUnit,

BlockingQueue<Runnable> workQueue,

ThreadFactory threadFactory,

RejectedExecutionHandler handler)

new ThreadFactoryBuilder().setNameFormat
("general-detail-batch-%d").build()

public void execute(Runnable command) {

if (command == null)

throw new NullPointerException();

/*

* Proceed in 3 steps:

*

* 1. If fewer than corePoolSize threads
are running, try to

* start a new thread with the given command
as its first

* task. The call to addWorker atomically checks
runState and

* workerCount, and so prevents false alarms
that would add

* threads when it shouldn't, by returning false.

*

* 2. If a task can be successfully queued, then
we still need

* to double-check whether we should have added
a thread

* (because existing ones died since last
checking) or that

* the pool shut down since entry into this
method. So we

* recheck state and if necessary roll back the
enqueuing if

* stopped, or start a new thread if there are none.

*

* 3. If we cannot queue task, then we try
to add a new

* thread. If it fails, we know we are shut
down or saturated

* and so reject the task.

*/

int c = ctl.get();

if (workerCountOf(c) < corePoolSize) {

if (addWorker(command, true))

return;

c = ctl.get();

}

if (isRunning(c) && workQueue.offer
(command)) {

int recheck = ctl.get();

if (! isRunning(recheck) && remove
(command))

reject(command);

else if (workerCountOf(recheck) == 0)

addWorker(null, false);

}

else if (!addWorker(command, false))

reject(command);

}

public void createExecutor() {

internalExecutor = true;

TaskQueue taskqueue = new TaskQueue();

TaskThreadFactory tf = new TaskThreadFactory
(getName() + "-exec-", daemon, getThreadPriority());

executor = new ThreadPoolExecutor
(getMinSpareThreads(), getMaxThreads(), 60,
TimeUnit.SECONDS,taskqueue, tf);

taskqueue.setParent( (ThreadPoolExecutor) executor);

}

/**

* Same as a java.util.concurrent.
ThreadPoolExecutor but implements a much more efficient

* {@link #getSubmittedCount()} method, to be
used to properly handle the work queue.

* If a RejectedExecutionHandler is not
specified a default one will be configured

* and that one will always throw a
RejectedExecutionException

*

*/

public class ThreadPoolExecutor extends
java.util.concurrent.ThreadPoolExecutor {

/**

* The number of tasks submitted but not yet
finished. This includes tasks

* in the queue and tasks that have been
handed to a worker thread but the

* latter did not start executing the task yet.

* This number is always greater or equal to
{@link #getActiveCount()}.

*/

// 新增的submittedCount成员变量，用于
统计已提交但还未完成的任务数

private final AtomicInteger submittedCount =
new AtomicInteger(0);

private final AtomicLong lastContextStopped
Time = new AtomicLong(0L);

// 构造函数

public ThreadPoolExecutor(int corePoolSize,
int maximumPoolSize, long keepAliveTime,
TimeUnit unit, BlockingQueue<Runnable>
workQueue, ThreadFactory threadFactory,

RejectedExecutionHandler handler) {

super(corePoolSize, maximumPoolSize,
keepAliveTime, unit, workQueue,
threadFactory, handler);

// 预启动所有核心线程

prestartAllCoreThreads();

}

}

@Override

public void execute(Runnable command) {

execute(command,0,TimeUnit.MILLISECONDS);

}

public void execute(Runnable command, long
timeout, TimeUnit unit) {

submittedCount.incrementAndGet();

try {

super.execute(command);

} catch (RejectedExecutionException rx) {

if (super.getQueue() instanceof TaskQueue) {

final TaskQueue queue = (TaskQueue)super.getQueue();

try {

if (!queue.force(command, timeout, unit)) {

submittedCount.decrementAndGet();

throw new RejectedExecution
Exception("Queue capacity is full.");

}

} catch (InterruptedException x) {

submittedCount.decrementAndGet();

throw new RejectedExecutionException(x);

}

} else {

submittedCount.decrementAndGet();

throw rx;

}

}

}

/**

* As task queue specifically designed to run
with a thread pool executor. The

* task queue is optimised to properly utilize
threads within a thread pool

* executor. If you use a normal queue, the
executor will spawn threads when

* there are idle threads and you wont be
able to force items onto the queue

* itself.

*/

public class TaskQueue extends Linked
BlockingQueue<Runnable> {

public boolean force(Runnable o, long timeout,
TimeUnit unit) throws InterruptedException {

if ( parent==null || parent.isShutdown() ) throw new
RejectedExecutionException("Executor not
running, can't force a command into the queue");

return super.offer(o,timeout,unit); //forces the
item onto the queue, to be used if the task is rejected

}

@Override

public boolean offer(Runnable o) {

// 1. parent为线程池，Tomcat中为自定义线程池实例

//we can't do any checks

if (parent==null) return super.offer(o);

// 2. 当线程数达到最大线程数时，新提交任务入队

//we are maxed out on threads, simply queue the object

if (parent.getPoolSize() == parent.getMaximumPoolSize
()) return super.offer(o);

// 3. 当提交的任务数小于线程池中已有的
线程数时，即有空闲线程，任务入队即可

//we have idle threads, just add it to the queue

if (parent.getSubmittedCount()<=(parent.
getPoolSize())) return super.offer(o);

// 4. 【关键点】如果当前线程数量未达到最
大线程数，直接返回false，让线程池创建新线程

//if we have less threads than maximum force
creation of a new thread

if (parent.getPoolSize()<parent.
getMaximumPoolSize()) return false;

// 5. 最后的兜底，放入队列

//if we reached here, we need to add it to the queue

return super.offer(o);

}

}

/**

* A Thread implementation that records the time at which it was created.

*

*/

public class TaskThread extends Thread {

private final long creationTime;

public TaskThread(ThreadGroup group, Runnable target, String name) {

super(group, new WrappingRunnable(target), name);

this.creationTime = System.currentTimeMillis();

}

/**

* Wraps a {@link Runnable} to swallow any {@link StopPooledThreadException}

* instead of letting it go and potentially trigger a break in a debugger.

*/

private static class WrappingRunnable implements Runnable {

private Runnable wrappedRunnable;

WrappingRunnable(Runnable wrappedRunnable) {

this.wrappedRunnable = wrappedRunnable;

}

@Override

public void run() {

try {

wrappedRunnable.run();

} catch(StopPooledThreadException exc) {

//expected : we just swallow the exception to avoid disturbing

//debuggers like eclipse's

log.debug("Thread exiting on purpose", exc);

}

}

}

}

/**

* Allows the server developer to specify the acceptCount (backlog) that

* should be used for server sockets. By default, this value

* is 100.

*/

private int acceptCount = 100;

private int maxConnections = 10000;

public void destroy() {

try {

poolExecutor.shutdown();

if (!poolExecutor.awaitTermination(AWAIT_TIMEOUT, TimeUnit.SECONDS)) {

poolExecutor.shutdownNow();

}

} catch (InterruptedException e) {

// 如果当前线程被中断，重新取消所有任务

pool.shutdownNow();

// 保持中断状态

Thread.currentThread().interrupt();

}

}

/**

* 在主线程中，开启鹰眼异步模式，并将ctx传递给多线程任务

**/

// 防止鹰眼链路丢失，需要传递

RpcContext_inner ctx = EagleEye.getRpcContext();

// 开启异步模式

ctx.setAsyncMode(true);

/**

* 在线程池任务线程中，设置鹰眼rpc环境

**/

private void runTask() {

try {

EagleEye.setRpcContext(ctx);

// do something...

} catch (Exception e) {

log.error("requestError, params: {}", this.params, e);

} finally {

// 判断当前任务是否是主线程在运行，当Rejected策略为CallerRunsPolicy的时候，核对当前线程

if (mainThread != Thread.currentThread()) {

EagleEye.clearRpcContext();

}

}

}

/**

* 使用 ThreadLocal 定义一个全局的 SimpleDateFormat

*/

private static ThreadLocal<SimpleDateFormat> simpleDateFormatThreadLocal = new

ThreadLocal<SimpleDateFormat>() {

@Override

protected SimpleDateFormat initialValue() {

return new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");

}

};

// 用法

String dateString = simpleDateFormatThreadLocal.get().format(calendar.getTime());

/**

* Returns the value in the current thread's copy of this

* thread-local variable. If the variable has no value for the

* current thread, it is first initialized to the value returned

* by an invocation of the {@link #initialValue} method.

*

* @return the current thread's value of this thread-local

*/

public T get() {

// 1. 获取当前线程

Thread t = Thread.currentThread();

// 2. 获取当前线程内部的ThreadLocalMap变量t.threadLocals;

ThreadLocalMap map = getMap(t);

// 3. 判断map是否为null

if (map != null) {

// 4. 使用当前threadLocal变量获取entry

ThreadLocalMap.Entry e = map.getEntry(this);

// 5. 判断entry是否为null

if (e != null) {

// 6.返回Entry.value

@SuppressWarnings("unchecked")

T result = (T)e.value;

return result;

}

}

// 7. 如果map/entry为null设置初始值

return setInitialValue();

}

/**

* Variant of set() to establish initialValue. Used instead

* of set() in case user has overridden the set() method.

*

* @return the initial value

*/

private T setInitialValue() {

// 1. 初始化value，如果重写就用重写后的value，默认null

T value = initialValue();

// 2. 获取当前线程

Thread t = Thread.currentThread();

// 3. 获取当前线程内部的ThreadLocalMap变量

ThreadLocalMap map = getMap(t);

if (map != null)

// 4. 不为null就set, key: threadLocal, value: value

map.set(this, value);

else

// 5. map若为null则创建ThreadLocalMap对象

createMap(t, value);

return value;

}

/**

* Create the map associated with a ThreadLocal. Overridden in

* InheritableThreadLocal.

*

* @param t the current thread

* @param firstValue value for the initial entry of the map

*/

void createMap(Thread t, T firstValue) {

t.threadLocals = new ThreadLocalMap(this, firstValue);

}

/**

* Construct a new map initially containing (firstKey, firstValue).

* ThreadLocalMaps are constructed lazily, so we only create

* one when we have at least one entry to put in it.

*/

ThreadLocalMap(ThreadLocal<?> firstKey, Object firstValue) {

// 1. 初始化entry数组，size: 16

table = new Entry[INITIAL_CAPACITY];

// 2. 计算value的index

int i = firstKey.threadLocalHashCode & (INITIAL_CAPACITY - 1);

// 3. 在对应index位置赋值

table[i] = new Entry(firstKey, firstValue);

// 4. entry size

size = 1;

// 5. 设置threshold: threshold = len * 2 / 3;

setThreshold(INITIAL_CAPACITY);

}

/**

* Set the resize threshold to maintain at worst a 2/3 load factor.

*/

private void setThreshold(int len) {

threshold = len * 2 / 3;

}

/**

* Sets the current thread's copy of this thread-local variable

* to the specified value. Most subclasses will have no need to

* override this method, relying solely on the {@link #initialValue}

* method to set the values of thread-locals.

*

* @param value the value to be stored in the current thread's copy of

* this thread-local.

*/

public void set(T value) {

// 1. 获取当前线程

Thread t = Thread.currentThread();

// 2. 获取当前线程内部的ThreadLocalMap变量

ThreadLocalMap map = getMap(t);

if (map != null)

// 3. 设置value

map.set(this, value);

else

// 4. 若map为null则创建ThreadLocalMap

createMap(t, value);

}

static class ThreadLocalMap {

/**

* The entries in this hash map extend WeakReference, using

* its main ref field as the key (which is always a

* ThreadLocal object). Note that null keys (i.e. entry.get()

* == null) mean that the key is no longer referenced, so the

* entry can be expunged from table. Such entries are referred to

* as "stale entries" in the code that follows.

*/

static class Entry extends WeakReference<ThreadLocal<?>> {

/** The value associated with this ThreadLocal. */

Object value;

Entry(ThreadLocal<?> k, Object v) {

super(k);

value = v;

}

}

// ...

}

EagleEyeHttpRequest eagleEyeHttpRequest = this.convertHttpRequest(httpRequest);

// 1. 初始化，将traceId、rpcId等数据存储到鹰眼的ThreadLocal变量中

EagleEyeRequestTracer.startTrace(eagleEyeHttpRequest, false);

try {

chain.doFilter(httpRequest, httpResponse);

} finally {

// 2. 清理ThreadLocal变量值

EagleEyeRequestTracer.endTrace(this.convertHttpResponse(httpResponse));

}

public class ThreadLocalTest {

public static class ThreadLocalDemo {

private ThreadLocal<String> threadLocalHolder = new ThreadLocal();

public void setValue(String value) {

threadLocalHolder.set(value);

}

public String getValue() {

return threadLocalHolder.get();

}

}

public static void main(String[] args) {

int count = 3;

List<ThreadLocalDemo> list = new LinkedList<>();

for (int i = 0; i < count; i++) {

ThreadLocalDemo demo = new ThreadLocalDemo();

demo.setValue("demo-" + i);

list.add(demo);

}

System.out.println();

}

}

// 反例，实际上使用了共享对象obj而并未隔离，

private static ThreadLocal<Obj> threadLocal = ThreadLocal.withIntitial(() -> obj);