/**
 * 实现线程安全的DateFormat
 *
 * @author cdrcool
 */
class SafeDateFormat {
    private static final ThreadLocal<DateFormat> TL =
            ThreadLocal.withInitial(() -> new SimpleDateFormat("yyyy-MM-dd HH:mm:ss"));

    public static DateFormat get() {
        return TL.get();
    }
}

//不同线程执⾏下⾯代码，返回的dateFormat是不同的
DateFormat dateFormat = SafeDateFormat.get();

class MyThreadLocal<T> {
    Map<Thread, T> locals = new ConcurrentHashMap<>();
    //获取线程变量
    T get() {
        return locals.get(Thread.currentThread());
    }
    //设置线程变量
    void	set(T	t)	{
        locals.put(Thread.currentThread(),	t);
    }
}

class Thread {
    // 内部持有ThreadLocalMap
    ThreadLocal.ThreadLocalMap threadLocals;
}

class ThreadLocal<T> {
    ...

    public T get()	{
        // ⾸先获取线程持有的ThreadLocalMap，等价于：ThreadLocalMap map = Thread.currentThread().threadLocals;
        Thread t = Thread.currentThread();
        ThreadLocalMap map = getMap(t);
        if (map != null) {
            // 在ThreadLocalMap中查找变量
            ThreadLocalMap.Entry e = map.getEntry(this);
            if (e != null) {
                @SuppressWarnings("unchecked")
                T result = (T)e.value;
                return result;
            }
        }
        return setInitialValue();
    }

    static class ThreadLocalMap {
        // Entry定义
        static class Entry extends WeakReference<ThreadLocal<?>>{
            Object value;
        }

        ...

        // 内部是数组⽽不是Map（线性探测法）
        Entry[]	table;

        ...

        // 根据ThreadLocal查找Entry
        private Entry getEntry(ThreadLocal<?> key) {
            // 省略查找逻辑
        }

        ...
    }
}

ExecutorService es;
ThreadLocal tl;
es.execute(() -> {
    // ThreadLocal增加变量
    tl.set(obj);
    try {
    // 省略业务逻辑代码
    } finally {
        // ⼿动清理ThreadLocal
        tl.remove();
    }
});

public class Heap {
    private int[] a; // 数组，从下标 1 开始存储数据
    private int n;  // 堆可以存储的最大数据个数
    private int count; // 堆中已经存储的数据个数
    
    public Heap(int capacity) {
        a = new int[capacity + 1];
        n = capacity;
        count = 0;
    }
    
    public void insert(int data) {
        if (count >= n) return; // 堆满了
        ++count;
        a[count] = data;
        int i = count;
        while (i / 2 > 0 && a[i] > a[i / 2]) { // 自下往上堆化
            swap(a, i, i / 2); // swap() 函数作用：交换下标为 i 和 i/2 的两个元素
            i = i / 2;
        }
    }
}

public void removeMax() {
    if (count == 0) return -1; // 堆中没有数据
    a[1] = a[count];
    --count;
    heapify(a, count, 1);
}

private void heapify(int[] a, int n, int i) { // 自上往下堆化
    while (true) {
        int maxPos = i;
        if (i*2 <= n && a[i] < a[i*2]) maxPos = i*2;
        if (i*2+1 <= n && a[maxPos] < a[i*2+1]) maxPos = i*2+1;
        if (maxPos == i) break;
        swap(a, i, maxPos);
        i = maxPos;
    }
}

private static void buildHeap(int[] a, int n) {
    for (int i = n/2; i >= 1; --i) {
        heapify(a, n, i);
    }
}

private static void heapify(int[] a, int n, int i) {
    while (true) {
        int maxPos = i;
        if (i*2 <= n && a[i] < a[i*2]) maxPos = i*2;
        if (i*2+1 <= n && a[maxPos] < a[i*2+1]) maxPos = i*2+1;
        if (maxPos == i) break;
        swap(a, i, maxPos);
        i = maxPos;
    }
}

// n 表示数据的个数，数组 a 中的数据从下标 1 到 n 的位置。
public static void sort(int[] a, int n) {
  buildHeap(a, n);
  int k = n;
  while (k > 1) {
    swap(a, 1, k);
    --k;
    heapify(a, k, 1);
  }
}

public ThreadPoolExecutor(int corePoolSize,
                          int maximumPoolSize,
                          long keepAliveTime,
                          TimeUnit unit,
                          BlockingQueue<Runnable> workQueue) {
    this(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue,
         Executors.defaultThreadFactory(), defaultHandler);
}

public ThreadPoolExecutor(int corePoolSize,
                          int maximumPoolSize,
                          long keepAliveTime,
                          TimeUnit unit,
                          BlockingQueue<Runnable> workQueue,
                          ThreadFactory threadFactory) {
    this(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue,
         threadFactory, defaultHandler);
}

public ThreadPoolExecutor(int corePoolSize,
                          int maximumPoolSize,
                          long keepAliveTime,
                          TimeUnit unit,
                          BlockingQueue<Runnable> workQueue,
                          RejectedExecutionHandler handler) {
    this(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue,
         Executors.defaultThreadFactory(), handler);
}

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.acc = System.getSecurityManager() == null ?
            null :
            AccessController.getContext();
    this.corePoolSize = corePoolSize;
    this.maximumPoolSize = maximumPoolSize;
    this.workQueue = workQueue;
    this.keepAliveTime = unit.toNanos(keepAliveTime);
    this.threadFactory = threadFactory;
    this.handler = handler;
}

try {
    // 业务逻辑
} catch (RuntimeException x) {
    // 按需处理
} catch (Throwable x) {
    // 按需处理
}

/**
 * 关闭线程池
 * 不再接受新的任务，同时等待已经提交的任务执行完成——包括那些还未开始执行的任务
 */
void shutdown();

/**
 * 关闭线程池
 * 将尝试取消所有运行中的任务，并且不再启动队列中尚未开始执行的任务
 */
void shutdownNow();
``` 
其原理都是遍历线程池中的工作线程，然后逐个调用线程的`interrupt`方法来中断线程，所以无法响应中断的任务可能永远无法终止。
不过它们也存在一定的区别，`shutdownNow`首先将线程池的状态设置成STOP，然后尝试停止所有的正在执行或暂停任务的线程，并且等待执行任务的列表，而`shutdown`只是将线程池的状态设置成SHUTDOWN状态，然后中断所有没有正在执行任务的线程。

### 线程池种类
`Executors`提供了一系列静态方法，利用它们我们可以很便捷的创建各种类型的线程池，如`FixedThreadPool`、`SingleThreadExecutor`、`CachedThreadPool`，又如具备调度功能的`ScheduledExecutorService`，以及遵循“工作窃取”模式的`ForkJoinPool`等。

#### FixedThreadPool
FixedThreadPool被称为可重用固定线程数的线程池，下面是源码实现：
```java
public static ExecutorService newFixedThreadPool(int nThreads) {
    return new ThreadPoolExecutor(nThreads, nThreads,
                                  0L, TimeUnit.MILLISECONDS,
                                  new LinkedBlockingQueue<Runnable>());
}

public static ExecutorService newFixedThreadPool(int nThreads, ThreadFactory threadFactory) {
    return new ThreadPoolExecutor(nThreads, nThreads,
                                  0L, TimeUnit.MILLISECONDS,
                                  new LinkedBlockingQueue<Runnable>(),
                                  threadFactory);
}

public static ExecutorService newSingleThreadExecutor() {
    return new FinalizableDelegatedExecutorService
        (new ThreadPoolExecutor(1, 1,
                                0L, TimeUnit.MILLISECONDS,
                                new LinkedBlockingQueue<Runnable>()));
}

public static ExecutorService newSingleThreadExecutor(ThreadFactory threadFactory) {
    return new FinalizableDelegatedExecutorService
        (new ThreadPoolExecutor(1, 1,
                                0L, TimeUnit.MILLISECONDS,
                                new LinkedBlockingQueue<Runnable>(),
                                threadFactory));
}

public static ExecutorService newCachedThreadPool() {
    return new ThreadPoolExecutor(0, Integer.MAX_VALUE,
                                  60L, TimeUnit.SECONDS,
                                  new SynchronousQueue<Runnable>());
}

public static ExecutorService newCachedThreadPool(ThreadFactory threadFactory) {
    return new ThreadPoolExecutor(0, Integer.MAX_VALUE,
                                  60L, TimeUnit.SECONDS,
                                  new SynchronousQueue<Runnable>(),
                                  threadFactory);
}

public class CustomThreadPoolExecutor extends ThreadPoolExecutor {
       
   public CustomThreadPoolExecutor() {
       super(10, 10,
               0L, TimeUnit.MILLISECONDS,
               new LinkedBlockingQueue<Runnable>());
   }

   @Override
   protected void beforeExecute(Thread t, Runnable r) {
       System.out.println(t.getName() + ": before execute");
       super.beforeExecute(t, r);
   }

   @Override
   protected void afterExecute(Runnable r, Throwable t) {
       System.out.println("after execute");
       super.afterExecute(r, t);
   }

   @Override
   protected void terminated() {
       System.out.println("terminated");
       super.terminated();
   }
}

@Test
public void test1() throws InterruptedException {
    CountDownLatch countDownLatch = new CountDownLatch(10);

    CustomThreadPoolExecutor executor = new CustomThreadPoolExecutor();

    IntStream.range(0, 10)
            .forEach(index -> executor.execute(() -> {
                System.out.println(String.format("线程%s运行中", index + 1));
                System.out.println("completedTaskCount: " + executor.getCompletedTaskCount());
                countDownLatch.countDown();
            }));

    countDownLatch.await();

    executor.shutdown();
    
    System.out.println("completedTaskCount: " + executor.getCompletedTaskCount());
}

/**
 * 在将来的某个时候执行给定的任务。这个任务可能在新线程或现有的池线程中执行。
 *
 * 如果线程池被关闭，又或者线程池的容量达到了，那么这个任务将无法被提交执行，而是交由当前的{@code RejectedExecutionHandler}处理。
 *
 * @param command 要执行的任务
 * @throws RejectedExecutionException 如果任务不能被接受执行
 * @throws NullPointerException 如果{@code command}是null
 */
public void execute(Runnable command) {
    // 如果任务为空，抛出空异常
    if (command == null)
        throw new NullPointerException();
        
    int c = ctl.get();
    // 如果运行的线程数小于corePoolSize，则创建线程并执行任务
    if (workerCountOf(c) < corePoolSize) {
        if (addWorker(command, true))
            return;
        c = ctl.get();
    }
    
    // 如果任务能够被添加到队列里（仍需做double-check）
    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); // is shut down or saturated
}

final void runWorker(Worker w) {
    Thread wt = Thread.currentThread();
    Runnable task = w.firstTask;
    w.firstTask = null;
    w.unlock(); // allow interrupts
    boolean completedAbruptly = true;
    try {
        // 循环获取工作队列（BlockingQueue<Runnable>）里的任务
        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
            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);
    }
}

public List<List<Integer>> levelOrder(TreeNode root) {
    if (root == null) return new ArrayList<>(0);
    
    List<List<Integer>> result = new ArrayList<>();
    
    Queue<TreeNode> queue = new LinkedList<TreeNode>();
    queue.offer(root);
    
    Queue<TreeNode> curLevelNodes = new LinkedList<TreeNode>();
    
    while (!queue.isEmpty()) {
        TreeNode node = queue.poll();
        curLevelNodes.offer(node);
        
        if (queue.isEmpty()) {
            List<Integer> list = new ArrayList<>(curLevelNodes.size());
            while (!curLevelNodes.isEmpty()) {
                TreeNode curNode = curLevelNodes.poll();
                list.add(curNode.val);
                
                if (curNode.left != null) {
                    queue.offer(curNode.left);
                }
                
                if (curNode.right != null) {
                    queue.offer(curNode.right);
                }
                
            }
            result.add(list);
        }
    }
    
    
    return result;
}

public BinaryTree find(BinaryTree tree, int data) {
    /*
        循环查找
     */
    BinaryTree root = tree;

    while (root != null) {
        if (tree.data == data) {
            return tree;
        } else if (tree.data < data) {
            root = root.right;
        } else {
            root = root.left;
        }
    }

    return null;

    /*
        递归查找
     */
//  if (tree == null) {
//      return null;
//  }
//  if (tree.data ==  data) {
//      return tree;
//  } else if (tree.data < data) {
//      return find(tree.right, data);
//  } else {
//      return find(tree.left, data);
//  }
}

public void insert(BinaryTree tree, int data) {
    if (tree == null) {
        tree = new BinaryTree(data);
        return;
    }

    BinaryTree root = tree;
    while (true) {
        if (data > tree.data) {
            if (root.right == null) {
                root.right = new BinaryTree(data);
                return;
            }
            root = root.right;
        } else if (data < tree.data) {
            if (root.left == null) {
                root.left = new BinaryTree(data);
                return;
            }
            root = root.left;
        }
    }
}

public void delete(BinaryTree tree, int data) {
    BinaryTree p = tree;
    BinaryTree pp = null;

    while (p != null && p.data != data) {
        pp = p;

        if (p.data > data) {
            p = p.left;
        } else {
            p = p.right;
        }
    }

    // 没有找到
    if (p == null) {
        return;
    }

    // 如果同时有左节点和右节点，找到右节点中最小的
    if (p.left != null && p.right != null) {
        BinaryTree minp = p.right;
        BinaryTree minpp = p;
        while (minp.left != null) {
            minpp = minp;
            minp = minp.left;
        }

        p.data = minp.data;

        p = minp;
        pp = minpp;
    }

    // 删除节点是叶子节点或者仅有一个子节点
    BinaryTree child;
    if (p.left != null) {
        child = p.left;
    } else if (p.right != null) {
        child = p.right;
    } else {
        child = null;
    }

    // 删除的是根节点
    if (pp == null) {
        tree = child;
    }
    // 同时有左节点和右节点，或仅有一个子节点且删除的是左节点
    else if (pp.left == p) {
        pp.left = child;
    }
    // 仅有一个子节点且删除的是右节点
    else {
        pp.right = child;
    }
}

/**
 * 模拟电商询价系统
 */
@Test
public void enquiry() {
    // 创建线程池
    ExecutorService executor = Executors.newFixedThreadPool(3);
    // 创建CompletionService
    CompletionService<Integer> service = new ExecutorCompletionService<>(executor);

    // 异步向电商S1询价
    service.submit(this::getPriceByS1);
    // 异步向电商S2询价
    service.submit(this::getPriceByS2);
    // 异步向电商S3询价
    service.submit(this::getPriceByS3);

    // 获取任务结果
    IntStream.range(0, 3).forEach(index -> {
        try {
            // 按任务执行完毕的顺序，获取任务结果
            Future<Integer> future = service.take();
            Integer value = future.get();
            System.out.println("value: " + value);

            // 将询价结果异步保存到数据库
            executor.execute(() -> save(value));
        } catch (InterruptedException | ExecutionException e) {
            e.printStackTrace();
        }
    });
}

private Integer getPriceByS1() {
    return 100;
}

private Integer getPriceByS2() {
    return 200;
}

private Integer getPriceByS3() {
    return 300;
}

private void save(Integer value) {
    System.out.println("save value: " + value);
}

/**
 * 模拟从多个途径获取值，如果有一个途径获取成功，就取消所有任务并返回结果
 */
@Test
public void multiPath() {
    // 创建线程池
    ExecutorService executor = Executors.newFixedThreadPool(3);
    // 创建CompletionService
    CompletionService<Integer> service = new ExecutorCompletionService<>(executor);

    // ⽤于保存Future对象
    List<Future<Integer>> futures = new ArrayList<>(3);

    // 提交异步任务，并保存future到futures
    futures.add(service.submit(this::getPriceByS1));
    futures.add(service.submit(this::getPriceByS2));
    futures.add(service.submit(this::getPriceByS3));

    // 获取最快返回的任务执⾏结果
    Integer value = 0;
    try {
        // 只要有⼀个成功返回，则break
        for (int i = 0; i < 3; ++i) {
            try {
                value = service.take().get();
            } catch (InterruptedException | ExecutionException e) {
                e.printStackTrace();
            }

            // 简单地通过判空来检查是否成功返回
            if (value != null) {
                break;
            }
        }
    } finally {
        System.out.println("value: " + value);

        // 取消所有任务
        futures.forEach(future -> future.cancel(true));
    }
}

public class ExecutorCompletionService<V> implements CompletionService<V> {
    /**
     * 线程池，由构造函数传入
     */
    private final Executor executor;
    /**
     * executor类型为AbstractExecutorService，aes的值就是executor，否则为null
    */
    private final AbstractExecutorService aes;
    /**
     * 阻塞队列，线程执行完毕后往该队列中插入值，主线程每次从该队列中获取值
     */
    private final BlockingQueue<Future<V>> completionQueue;

    /**
     * FutureTask extension to enqueue upon completion
     * 继承FutureTask，主要是实现了里面的空方法done
     */
    private class QueueingFuture extends FutureTask<Void> {
        QueueingFuture(RunnableFuture<V> task) {
            super(task, null);
            this.task = task;
        }
        /**
         * futureTask执行完毕后，就往阻塞队列中插入值
         */
        protected void done() { completionQueue.add(task); }
        private final Future<V> task;
    }

    private RunnableFuture<V> newTaskFor(Callable<V> task) {
        if (aes == null)
            return new FutureTask<V>(task);
        else
            return aes.newTaskFor(task);
    }

    private RunnableFuture<V> newTaskFor(Runnable task, V result) {
        if (aes == null)
            return new FutureTask<V>(task, result);
        else
            return aes.newTaskFor(task, result);
    }

    public ExecutorCompletionService(Executor executor) {
        if (executor == null)
            throw new NullPointerException();
        this.executor = executor;
        this.aes = (executor instanceof AbstractExecutorService) ?
            (AbstractExecutorService) executor : null;
        this.completionQueue = new LinkedBlockingQueue<Future<V>>();
    }

    public ExecutorCompletionService(Executor executor,
                                     BlockingQueue<Future<V>> completionQueue) {
        if (executor == null || completionQueue == null)
            throw new NullPointerException();
        this.executor = executor;
        this.aes = (executor instanceof AbstractExecutorService) ?
            (AbstractExecutorService) executor : null;
        this.completionQueue = completionQueue;
    }

    /**
     * 提交任务
     */
    public Future<V> submit(Callable<V> task) {
        if (task == null) throw new NullPointerException();
        RunnableFuture<V> f = newTaskFor(task);
        // 包装原始task，以实现task执行完毕后，会往阻塞队列中插入task
        executor.execute(new QueueingFuture(f));
        return f;
    }

    /**
     * 获取执行完毕的任务：从阻塞队列中获取，如果队列为空就会一直阻塞
     */
    public Future<V> submit(Runnable task, V result) {
        if (task == null) throw new NullPointerException();
        RunnableFuture<V> f = newTaskFor(task, result);
        executor.execute(new QueueingFuture(f));
        return f;
    }

    /**
     * 获取执行完毕的任务：从阻塞队列中获取，如果队列为空就会一直阻塞
     */
    public Future<V> take() throws InterruptedException {
        return completionQueue.take();
    }

    /**
     * 获取执行完毕的任务：从阻塞队列中获取，如果队列为空就会返回null
     */
    public Future<V> poll() {
        return completionQueue.poll();
    }

    /**
     * 获取执行完毕的任务：从阻塞队列中获取，如果队列为空就会先阻塞timeout，超时后还为空就返回null
     */
    public Future<V> poll(long timeout, TimeUnit unit)
            throws InterruptedException {
        return completionQueue.poll(timeout, unit);
    }

}