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 lastContextStoppedTime = 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 RejectedExecutionException("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 LinkedBlockingQueue<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 = newThreadLocal<SimpleDateFormat>() {@Overrideprotected 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);