JUC

public class ThreadA extends Thread {
    @Override
    public void run() {
    }
    public static void main(String[] args){
      new ThreadA().start();
    }
}

public class ThreadB implements Runnable {
    @Override
    public void run() {
    }
    public static void main(String[] args){
      new Thread(new ThreadB()).start();
    }
}

import java.util.concurrent.Callable;
import java.util.concurrent.FutureTask;

public class ThreadC implements Callable<String> {
    @Override
    public String call() {
        return "Callable";
    }
    public static void main(String[] args) throws Exception {
      FutureTask<String> task = new FutureTask<>(new ThreadC());
      task.run();
      task.get();
    }
}

public class ThreadStopDemo1 extends Thread {

    private volatile boolean stopped = false;

    @Override
    public void run() {
        while (!stopped){
            // working here. (1)
        }
    }

    public void stopWork() {
        this.stopped = true;
    }
}

public class ThreadStopDemo2 extends Thread {

    @Override
    public void run() {
        try {
            Thread.sleep(10000L);
        } catch (InterruptedException e) {
            // stop working. (1)
        }
    }

    public static void main(String[] args) {
        Thread thread = new ThreadStopDemo2();
        thread.start();
        thread.interrupt(); (2)
    }
}

abstract static class Sync extends AbstractQueuedSynchronizer {
    final void lock() {
        if (!initialTryLock()) // 首先尝试先获得锁, 如果获取失败则与其他线程竞争来获取锁
            acquire(1);
    }
}

public abstract class AbstractQueuedSynchronizer {

        public final void acquire(int arg) {
            if (!tryAcquire(arg)) // tryAcquire与initialTryLock类似, 会尝试去获取锁
                acquire(null, arg, false, false, false, 0L); // 竞争锁
        }

        final int acquire(Node node, int arg, boolean shared,
                      boolean interruptible, boolean timed, long time) {
        Thread current = Thread.currentThread();
        byte spins = 0, postSpins = 0;   // retries upon unpark of first thread
        boolean interrupted = false, first = false;
        Node pred = null;                // predecessor of node when enqueued
        for (;;) {
            if (!first && (pred = (node == null) ? null : node.prev) != null &&
                !(first = (head == pred))) {
                if (pred.status < 0) {
                    cleanQueue();           // predecessor cancelled
                    continue;
                } else if (pred.prev == null) {
                    Thread.onSpinWait();    // ensure serialization
                    continue;
                }
            }
            if (first || pred == null) {
                boolean acquired;
                try {
                    if (shared)
                        acquired = (tryAcquireShared(arg) >= 0);
                    else
                        acquired = tryAcquire(arg); // 尝试去竞争锁, 如果竞争成功则返回true
                } catch (Throwable ex) {
                    cancelAcquire(node, interrupted, false);
                    throw ex;
                }
                if (acquired) {
                    if (first) {
                        node.prev = null;
                        head = node;
                        pred.next = null;
                        node.waiter = null;
                        if (shared)
                            signalNextIfShared(node); // 如果当前竞争锁成功, 并且当前节点是头节点, 那么通知调用LockSupport.unpark唤醒当前节点的next节点
                        if (interrupted)
                            current.interrupt();
                    }
                    return 1;
                }
            }
            if (node == null) {                 // allocate; retry before enqueue
                if (shared)
                    node = new SharedNode();
                else
                    node = new ExclusiveNode(); // 初始化Node
            } else if (pred == null) {          // 尝试将当前节点插入到等待队列末端, 并将当前tail的next指向成当前节点
                node.waiter = current;
                Node t = tail;
                node.setPrevRelaxed(t);         // avoid unnecessary fence
                if (t == null)
                    tryInitializeHead(); // 如果head为null, 那么初始化head和tail为一个哑元节点
                else if (!casTail(t, node)) // 将当前节点设置成tail
                    node.setPrevRelaxed(null);  // back out
                else
                    t.next = node;
            } else if (first && spins != 0) {
                --spins;                        // reduce unfairness on rewaits
                Thread.onSpinWait();
            } else if (node.status == 0) {
                node.status = WAITING;          // enable signal and recheck
            } else {
                long nanos;
                spins = postSpins = (byte)((postSpins << 1) | 1);
                if (!timed)
                    LockSupport.park(this); // 当前节点已经进入队列, 调用LockSupport.park暂停当前线程
                else if ((nanos = time - System.nanoTime()) > 0L)
                    LockSupport.parkNanos(this, nanos);
                else
                    break;
                node.clearStatus();
                if ((interrupted |= Thread.interrupted()) && interruptible)
                    break;
            }
        }
        return cancelAcquire(node, interrupted, interruptible);
    }
}

public abstract class AbstractQueuedSynchronizer {
    public final boolean release(int arg) {
        if (tryRelease(arg)) {
            signalNext(head);
            return true;
        }
        return false;
    }

    private static void signalNext(Node h) {
        Node s;
        if (h != null && (s = h.next) != null && s.status != 0) {
            s.getAndUnsetStatus(WAITING);
            LockSupport.unpark(s.waiter); (1)
        }
    }
}

abstract static class Sync extends AbstractQueuedSynchronizer {
    protected final boolean tryRelease(int releases) {
        int c = getState() - releases;
        if (getExclusiveOwnerThread() != Thread.currentThread())
            throw new IllegalMonitorStateException();
        boolean free = (c == 0);
        if (free)
            setExclusiveOwnerThread(null); (1)
        setState(c); (2)
        return free;
    }
}

package me.jy.lang.thread.juc;

import java.util.concurrent.locks.StampedLock;
import java.util.stream.IntStream;

/**
 * @author jy
 */
public class StampedLockDemo {

    private final StampedLock lock = new StampedLock();

    private double x;
    private double y;

    public static void main(String[] args) {
        StampedLockDemo stampedLockDemo = new StampedLockDemo();
        IntStream.rangeClosed(1, 100)
            .parallel()
            .forEach(i -> stampedLockDemo.move(i, i + 1));
        // 7212
        System.out.println(stampedLockDemo.computeDistance());
    }

    public void move(double deltaX, double deltaY) {
        // 获取写锁
        long stamp = lock.writeLock();
        x += deltaX;
        y += deltaY;
        // 释放写锁
        lock.unlockWrite(stamp);
    }

    public double computeDistance() {
        // 尝试获取读锁(乐观锁)
        long stamp = lock.tryOptimisticRead();
        // 操作数据
        double currentX = x;
        double currentY = y;
        // 校验是否有线程获取过写锁
        if (!lock.validate(stamp)) {
            // 重新获取读锁(悲观锁)
            stamp = lock.readLock();
            currentX = x;
            currentY = y;
            // 释放锁
            lock.unlockRead(stamp);
        }
        return Math.sqrt(currentX * currentX + currentY * currentY);
    }

}

public class StampedLock implements java.io.Serializable {

    public long writeLock() {
        // 重置低8位, 只有当低8位都为0的时候可以直接获得锁.
        long s = U.getLongOpaque(this, STATE) & ~ABITS, nextState;
        // 设置第8位为1.
        if (casState(s, nextState = s | WBIT)) {
            U.storeStoreFence();
            return nextState;
        }
        return acquireWrite(false, false, 0L);
    }

    private long acquireWrite(boolean interruptible, boolean timed, long time) {
        byte spins = 0, postSpins = 0;   // retries upon unpark of first thread
        boolean interrupted = false, first = false;
        WriterNode node = null;
        Node pred = null;
        for (long s, nextState;;) {
            if (!first && (pred = (node == null) ? null : node.prev) != null &&
                !(first = (head == pred))) {
                if (pred.status < 0) {
                    cleanQueue();           // predecessor cancelled
                    continue;
                } else if (pred.prev == null) {
                    Thread.onSpinWait();    // ensure serialization
                    continue;
                }
            }
            if ((first || pred == null) && ((s = state) & ABITS) == 0L &&
                casState(s, nextState = s | WBIT)) {
                U.storeStoreFence();
                if (first) {
                    node.prev = null;
                    head = node;
                    pred.next = null;
                    node.waiter = null;
                    if (interrupted)
                        Thread.currentThread().interrupt();
                }
                return nextState;
            } else if (node == null) {          // retry before enqueuing
                node = new WriterNode();
            } else if (pred == null) {          // try to enqueue
                Node t = tail;
                node.setPrevRelaxed(t);
                if (t == null)
                    tryInitializeHead();
                else if (!casTail(t, node))
                    node.setPrevRelaxed(null);  // back out
                else
                    t.next = node;
            } else if (first && spins != 0) {   // reduce unfairness
                --spins;
                Thread.onSpinWait();
            } else if (node.status == 0) {      // enable signal
                if (node.waiter == null)
                    node.waiter = Thread.currentThread();
                node.status = WAITING;
            } else {
                long nanos;
                spins = postSpins = (byte)((postSpins << 1) | 1);
                if (!timed)
                    LockSupport.park(this);
                else if ((nanos = time - System.nanoTime()) > 0L)
                    LockSupport.parkNanos(this, nanos);
                else
                    break;
                node.clearStatus();
                if ((interrupted |= Thread.interrupted()) && interruptible)
                    break;
            }
        }
        return cancelAcquire(node, interrupted);
    }

    private long cancelAcquire(Node node, boolean interrupted) {
        if (node != null) {
            node.waiter = null;
            node.status = CANCELLED;
            cleanQueue();
            if (node instanceof ReaderNode)
                signalCowaiters((ReaderNode)node);
        }
        return (interrupted || Thread.interrupted()) ? INTERRUPTED : 0L;
    }

    private static void signalCowaiters(ReaderNode node) {
        if (node != null) {
            // 只要有一个ReadLock被唤醒, 就会唤醒所有的ReadLock.
            for (ReaderNode c; (c = node.cowaiters) != null; ) {
                if (node.casCowaiters(c, c.cowaiters))
                    LockSupport.unpark(c.waiter);
            }
        }
    }

    private long releaseWrite(long s) {
        long nextState = state = unlockWriteState(s);
        // 唤醒等待队列第一个节点.
        signalNext(head);
        return nextState;
    }

    private static long unlockWriteState(long s) {
        // 将state加上1<<7, 这样第8位就置0了.
        return ((s += WBIT) == 0L) ? ORIGIN : s;
    }

    // 获取乐观锁
    public long tryOptimisticRead() {
        long s;
        // state & WBIT 不为0时, 方法直接返回0, validate(0) 会返回false.
        return (((s = state) & WBIT) == 0L) ? (s & SBITS) : 0L;
    }

    public boolean validate(long stamp) {
        U.loadFence();
        return (stamp & SBITS) == (state & SBITS);
    }

}

package me.jy.lang.thread.juc;

import lombok.extern.slf4j.Slf4j;

import java.util.concurrent.Semaphore;
import java.util.concurrent.TimeUnit;
import java.util.stream.IntStream;

/**
 * @author jy
 */
@Slf4j
public class SemaphoreDemo {

    private static final Semaphore SEMAPHORE = new Semaphore(5);

    public static void main(String[] args) {
        IntStream
            .rangeClosed(1, 100)
            .parallel()
            .forEach(i -> {
                try {
                    SEMAPHORE.acquire();
                    log.info("I am in.");
                    TimeUnit.SECONDS.sleep(1);
                } catch (InterruptedException ignored) {
                } finally {
                    SEMAPHORE.release();
                }
            });
    }
}

public abstract class AbstractQueuedSynchronizer {
    public final void acquireSharedInterruptibly(int arg)
        throws InterruptedException {
        if (Thread.interrupted() ||
            (tryAcquireShared(arg) < 0 &&
             acquire(null, arg, true, true, false, 0L) < 0))
            throw new InterruptedException();
    }
}

public class Semaphore implements java.io.Serializable {

    public void acquire() throws InterruptedException {
        sync.acquireSharedInterruptibly(1);
    }

    public void release() {
        sync.releaseShared(1);
    }

    abstract static class Sync extends AbstractQueuedSynchronizer {

        Sync(int permits) {
            setState(permits);
        }

        final int getPermits() {
            return getState();
        }

        final int nonfairTryAcquireShared(int acquires) {
            for (;;) {
                int available = getState();
                int remaining = available - acquires;
                // CAS修改state, remaining小于0时代表剩余的state不够用了, 线程会去竞争锁.
                if (remaining < 0 ||
                    compareAndSetState(available, remaining))
                    return remaining;
            }
        }

        protected final boolean tryReleaseShared(int releases) {
            for (;;) {
                int current = getState();
                int next = current + releases;
                if (next < current)
                    throw new Error("Maximum permit count exceeded");
                // 因为有多个线程持有该锁, 所以释放锁时需要不停CAS修改state.
                if (compareAndSetState(current, next))
                    return true;
            }
        }
    }
}

package me.jy.lang.thread.juc;

import lombok.extern.slf4j.Slf4j;

import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;
import java.util.stream.IntStream;

/**
 * @author jy
 */
@Slf4j
public class CountDownLatchDemo {

    private static final int WORKER_COUNT = 10;

    private static final ExecutorService executorService = Executors.newFixedThreadPool(4);

    public static void main(String[] args) throws InterruptedException {

        CountDownLatch latch = new CountDownLatch(WORKER_COUNT);
        Runnable work = () -> {
            try {
                log.info("working");
                TimeUnit.SECONDS.sleep(1);
                latch.countDown();
            } catch (InterruptedException ignored) {
            }
        };
        IntStream.rangeClosed(1, WORKER_COUNT).forEach(i -> executorService.execute(work));
        latch.await();

        log.info("workers done!");
        executorService.shutdownNow();
    }
}

public class CountDownLatch {

    public void await() throws InterruptedException {
        sync.acquireSharedInterruptibly(1);
    }

    public void countDown() {
        sync.releaseShared(1);
    }

    private static final class Sync extends AbstractQueuedSynchronizer {

        Sync(int count) {
            setState(count);
        }

        // 一直等到state为0时才能竞争到锁.
        protected int tryAcquireShared(int acquires) {
            return (getState() == 0) ? 1 : -1;
        }

        protected boolean tryReleaseShared(int releases) {
            // 不停地CAS将state减1
            for (;;) {
                int c = getState();
                if (c == 0)
                    return false;
                int nextc = c - 1;
                if (compareAndSetState(c, nextc))
                    // state减到0, 可以唤醒等待线程.
                    return nextc == 0;
            }
        }
    }
}

package me.jy.lang.thread.juc;

import lombok.Data;
import lombok.extern.slf4j.Slf4j;

import java.time.Instant;
import java.util.Comparator;
import java.util.List;
import java.util.concurrent.*;
import java.util.stream.IntStream;

/**
 * @author jy
 */
@Slf4j
public class CyclicBarrierDemo {

    private static final int THREAD_COUNT = 5;

    private static final ExecutorService THREAD_POOL = Executors.newFixedThreadPool(THREAD_COUNT);

    private static final List<Player> PLAYERS = new CopyOnWriteArrayList<>();

    public static void main(String[] args) {
        CyclicBarrier cyclicBarrier = new CyclicBarrier(THREAD_COUNT, () -> {
            log.info("====== Result ======");
            PLAYERS.stream()
                .sorted(Comparator.comparingLong(a -> a.finishedAt))
                .forEach(System.out::println);
            try {
                TimeUnit.SECONDS.sleep(2);
            } catch (InterruptedException e) {
                throw new RuntimeException(e);
            }
            log.info("====== End ======");
            THREAD_POOL.shutdownNow();
        });

        IntStream.rangeClosed(1, THREAD_COUNT)
            .forEach(i -> THREAD_POOL.execute(() -> {
                try {
                    log.info(" is running!");
                    TimeUnit.SECONDS.sleep(i);
                    PLAYERS.add(new Player().setName(Thread.currentThread().getName()).setFinishedAt(Instant.now().getEpochSecond()));
                    cyclicBarrier.await();
                    log.info(" over!"); // 执行完barrierAction后才释放锁.
                } catch (Exception ignored) {
                }
            }));
    }

    @Data
    private static class Player {
        private String name;
        private long finishedAt;
    }

}

public class CyclicBarrier {

    private static class Generation {
        Generation() {}
        // 标识当前线程是否被打断.
        boolean broken;
    }

    public int await() throws InterruptedException, BrokenBarrierException {
        try {
            return dowait(false, 0L);
        } catch (TimeoutException toe) {
            throw new Error(toe); // cannot happen
        }
    }

    private int dowait(boolean timed, long nanos)
        throws InterruptedException, BrokenBarrierException,
               TimeoutException {
        final ReentrantLock lock = this.lock;
        lock.lock();
        try {
            final Generation g = generation;

            if (g.broken)
                throw new BrokenBarrierException();

            // 捕获外部中断信号
            if (Thread.interrupted()) {
                breakBarrier();
                throw new InterruptedException();
            }

            int index = --count;
            // 减到0, 则表示最后一个线程到达执行点
            if (index == 0) {  // tripped
                Runnable command = barrierCommand;
                if (command != null) {
                    try {
                        // 最后一个线程执行回调方法
                        command.run();
                    } catch (Throwable ex) {
                        breakBarrier();
                        throw ex;
                    }
                }
                nextGeneration();
                return 0;
            }

            // loop until tripped, broken, interrupted, or timed out
            for (;;) {
                try {
                    if (!timed)
                        // 当前线程阻塞, 等待最后一个线程唤醒
                        // await会释放锁, 让其他线程也在这里阻塞
                        trip.await();
                    else if (nanos > 0L)
                        nanos = trip.awaitNanos(nanos);
                } catch (InterruptedException ie) {
                    if (g == generation && ! g.broken) {
                        breakBarrier();
                        throw ie;
                    } else {
                        // We're about to finish waiting even if we had not
                        // been interrupted, so this interrupt is deemed to
                        // "belong" to subsequent execution.
                        Thread.currentThread().interrupt();
                    }
                }

                if (g.broken)
                    throw new BrokenBarrierException();

                // 被唤醒后, generation被重新初始化为新的对象, 此处不相等, await方法就会结束.
                if (g != generation)
                    return index;

                if (timed && nanos <= 0L) {
                    breakBarrier();
                    throw new TimeoutException();
                }
            }
        } finally {
            lock.unlock();
        }
    }

    private void nextGeneration() {
        // 唤醒其他阻塞线程
        trip.signalAll();
        // 重置count计数, 下轮可以继续使用
        count = parties;
        // 重置标志位
        generation = new Generation();
    }

    private void breakBarrier() {
        generation.broken = true;
        count = parties;
        trip.signalAll();
    }
}

package me.jy.lang.thread.juc;

import lombok.extern.slf4j.Slf4j;

import java.util.concurrent.Exchanger;
import java.util.concurrent.TimeUnit;

/**
 * @author jy
 */
@Slf4j
public class ExchangerDemo {

    public static void main(String[] args) {
        Exchanger<String> exchanger = new Exchanger<>();

        new Thread(() -> {
            try {
                // 阻塞等待线程B发送数据
                String data = exchanger.exchange("A");
                log.info(data);
            } catch (InterruptedException e) {
                throw new RuntimeException(e);
            }
        }, "A").start();

        new Thread(() -> {
            try {
                TimeUnit.SECONDS.sleep(3);
                // 与线程A交换数据
                String data = exchanger.exchange("B");
                log.info(data);
            } catch (InterruptedException e) {
                throw new RuntimeException(e);
            }
        }, "B").start();
    }
}

public class Exchanger<V> {

    private volatile Node[] arena;

    private volatile Node slot;

    private final Participant participant;

    static final class Participant extends ThreadLocal<Node> {
        public Node initialValue() { return new Node(); }
    }

    public Exchanger() {
        participant = new Participant();
    }

    @jdk.internal.vm.annotation.Contended static final class Node {
        int index;              // Arena index
        int bound;              // Last recorded value of Exchanger.bound
        int collides;           // Number of CAS failures at current bound
        int hash;               // Pseudo-random for spins
        Object item;            // This thread's current item
        volatile Object match;  // Item provided by releasing thread
        volatile Thread parked; // Set to this thread when parked, else null
    }

    public V exchange(V x) throws InterruptedException {
        Object v;
        Node[] a;
        Object item = (x == null) ? NULL_ITEM : x;
        // 先用slot交换数据, 如果slot被占用了, 就会用arena交换数据.
        if (((a = arena) != null ||
             (v = slotExchange(item, false, 0L)) == null) &&
            (Thread.interrupted() || // disambiguates null return
             (v = arenaExchange(item, false, 0L)) == null))
            throw new InterruptedException();
        return (v == NULL_ITEM) ? null : (V)v;
    }

    private final Object slotExchange(Object item, boolean timed, long ns) {
        Node p = participant.get();
        Thread t = Thread.currentThread();
        if (t.isInterrupted()) // preserve interrupt status so caller can recheck
            return null;

        for (Node q;;) {
            // slot不为null表示有别的线程在等待交换数据
            if ((q = slot) != null) {
                // 重置SLOT, 拿到对方线程想要交换的数据
                if (SLOT.compareAndSet(this, q, null)) {
                    Object v = q.item;
                    // 将自己的数据设置到对方线程的match属性上.
                    q.match = item;
                    Thread w = q.parked;
                    if (w != null)
                        // 唤醒该线程
                        LockSupport.unpark(w);
                    return v;
                }
                // 有其他线程修改了SLOT, CAS失败, 创建arena数组.
                // create arena on contention, but continue until slot null
                if (NCPU > 1 && bound == 0 &&
                    BOUND.compareAndSet(this, 0, SEQ)) // bound设置为256
                    arena = new Node[(FULL + 2) << ASHIFT]; // (CPU数/2+2)*32
            }
            // arena不为空则转为使用arena数组交换数据
            else if (arena != null)
                return null; // caller must reroute to arenaExchange
            else {
                p.item = item;
                // 将自己线程数据设置到item属性里, 然后尝试放入到SLOT中.
                if (SLOT.compareAndSet(this, null, p))
                    break;
                p.item = null;
            }
        }

        // 当前线程数据放入到slot中, 此处自旋, 直到对方线程取出.
        int h = p.hash;
        long end = timed ? System.nanoTime() + ns : 0L;
        // 自旋1024次
        int spins = (NCPU > 1) ? SPINS : 1;
        Object v;
        // 如果p.match为null, 说明还没有别的线程过来交换数据
        while ((v = p.match) == null) {
            if (spins > 0) {
                h ^= h << 1; h ^= h >>> 3; h ^= h << 10;
                if (h == 0)
                    h = SPINS | (int)t.getId();
                else if (h < 0 && (--spins & ((SPINS >>> 1) - 1)) == 0)
                    Thread.yield();
            }
            else if (slot != p)
                spins = SPINS;
            else if (!t.isInterrupted() && arena == null &&
                     (!timed || (ns = end - System.nanoTime()) > 0L)) {
                p.parked = t;
                // 如果slot还没有被拿走, 阻塞自己.
                if (slot == p) {
                    if (ns == 0L)
                        LockSupport.park(this);
                    else
                        LockSupport.parkNanos(this, ns);
                }
                p.parked = null;
            }
            // ns小于0或者被中断, 取消交换直接返回.
            else if (SLOT.compareAndSet(this, p, null)) {
                v = timed && ns <= 0L && !t.isInterrupted() ? TIMED_OUT : null;
                break;
            }
        }
        // 清空双方交换的数据
        MATCH.setRelease(p, null);
        p.item = null;
        // 保留随机数, 供下次使用
        p.hash = h;
        return v;
    }
}

package me.jy.lang.thread.juc;

import lombok.SneakyThrows;
import lombok.extern.slf4j.Slf4j;

import java.util.concurrent.Phaser;
import java.util.concurrent.TimeUnit;
import java.util.stream.IntStream;

/**
 * @author jy
 */
@Slf4j
public class PhaserDemo {

    private static final int PARTIES = 10;
    private static final Phaser PHASER = new Phaser(PARTIES);

    public static void main(String[] args) {
        PhaserDemo demo = new PhaserDemo();
        IntStream.rangeClosed(1, PARTIES)
            .parallel()
            .forEach(i -> new Thread(demo::countDownLatch).start());
        // 模拟CountDownLatch
        // 等待parties减到0
        PHASER.awaitAdvance(0);
        log.info("awaitAdvance passed");

        // 模拟CyclicBarrier
        IntStream.rangeClosed(1, PARTIES)
            .parallel()
            .forEach(i -> new Thread(demo::cyclicBarrier).start());
    }

    @SneakyThrows
    public void countDownLatch() {
        TimeUnit.SECONDS.sleep(200000);
        PHASER.arrive();
        log.info("arrived");
    }

    @SneakyThrows
    public void cyclicBarrier() {
        log.info("phase0");
        TimeUnit.SECONDS.sleep(2);
        // 等待其他线程执行完毕
        PHASER.arriveAndAwaitAdvance();
        log.info("phase1");
        TimeUnit.SECONDS.sleep(3);
        // 等待其他线程执行完毕
        PHASER.arriveAndAwaitAdvance();
        log.info("phase2");
    }

}

public class Phaser {

    // 第1位标识是否同步完成, 第2~32为phase数, 第33~48位为总线程数, 第49~64位为未到达线程数.
    private volatile long state;

    private final Phaser parent;
    private final Phaser root;

    private final AtomicReference<QNode> evenQ;
    private final AtomicReference<QNode> oddQ;

    public Phaser(Phaser parent, int parties) {
        if (parties >>> PARTIES_SHIFT != 0)
            throw new IllegalArgumentException("Illegal number of parties");
        int phase = 0;
        this.parent = parent;
        if (parent != null) {
            final Phaser root = parent.root;
            this.root = root;
            this.evenQ = root.evenQ;
            this.oddQ = root.oddQ;
            if (parties != 0)
                phase = parent.doRegister(1);
        }
        else {
            this.root = this;
            this.evenQ = new AtomicReference<QNode>();
            this.oddQ = new AtomicReference<QNode>();
        }
        // 设置2~32位, 33~48位, 49~64位 为parties
        this.state = (parties == 0) ? (long)EMPTY :
            ((long)phase << PHASE_SHIFT) |
            ((long)parties << PARTIES_SHIFT) |
            ((long)parties);
    }

    public int awaitAdvance(int phase) {
        final Phaser root = this.root;
        long s = (root == this) ? state : reconcileState();
        // 右移32位得到state里的phase数
        int p = (int)(s >>> PHASE_SHIFT);
        if (phase < 0)
            return phase;
        if (p == phase)
            return root.internalAwaitAdvance(phase, null);
        return p;
    }

    private int internalAwaitAdvance(int phase, QNode node) {
        // 唤醒阻塞在前面phase阶段的线程
        releaseWaiters(phase-1);          // ensure old queue clean
        boolean queued = false;           // true when node is enqueued
        int lastUnarrived = 0;            // to increase spins upon change
        int spins = SPINS_PER_ARRIVAL;
        long s;
        int p;
        while ((p = (int)((s = state) >>> PHASE_SHIFT)) == phase) {
            if (node == null) {           // spinning in noninterruptible mode
                int unarrived = (int)s & UNARRIVED_MASK;
                if (unarrived != lastUnarrived &&
                    (lastUnarrived = unarrived) < NCPU)
                    spins += SPINS_PER_ARRIVAL;
                boolean interrupted = Thread.interrupted();
                // 如果被打断, 或者自旋结束, 则创建一个QNode
                if (interrupted || --spins < 0) { // need node to record intr
                    node = new QNode(this, phase, false, false, 0L);
                    node.wasInterrupted = interrupted;
                }
                else
                    // 首先空自旋
                    Thread.onSpinWait();
            }
            else if (node.isReleasable()) // done or aborted
                break;
            else if (!queued) {           // push onto queue
                AtomicReference<QNode> head = (phase & 1) == 0 ? evenQ : oddQ;
                QNode q = node.next = head.get();
                // 加入到队列头部, phase为奇数时加入到evenQ, phase为偶数时加入到oddQ.
                if ((q == null || q.phase == phase) &&
                    (int)(state >>> PHASE_SHIFT) == phase) // avoid stale enq
                    queued = head.compareAndSet(q, node);
            }
            else {
                try {
                    // 将当前node阻塞.
                    ForkJoinPool.managedBlock(node);
                } catch (InterruptedException cantHappen) {
                    node.wasInterrupted = true;
                }
            }
        }

        if (node != null) {
            if (node.thread != null)
                node.thread = null;       // avoid need for unpark()
            if (node.wasInterrupted && !node.interruptible)
                Thread.currentThread().interrupt();
            if (p == phase && (p = (int)(state >>> PHASE_SHIFT)) == phase)
                return abortWait(phase); // possibly clean up on abort
        }
        // 唤醒阻塞在当前phase阶段的线程
        releaseWaiters(phase);
        return p;
    }

    private void releaseWaiters(int phase) {
        QNode q;   // first element of queue
        Thread t;  // its thread
        AtomicReference<QNode> head = (phase & 1) == 0 ? evenQ : oddQ;
        while ((q = head.get()) != null &&
               q.phase != (int)(root.state >>> PHASE_SHIFT)) {
            if (head.compareAndSet(q, q.next) &&
                (t = q.thread) != null) {
                q.thread = null;
                LockSupport.unpark(t);
            }
        }
    }

    public int arrive() {
        // arrive后将未到达线程数减1.
        return doArrive(1);
    }

    public int arriveAndDeregister() {
        // arrive后将未到达线程数和总线程数都减1
        return doArrive(65537);
    }

    private int doArrive(int adjust) {
        final Phaser root = this.root;
        for (;;) {
            long s = (root == this) ? state : reconcileState();
            int phase = (int)(s >>> PHASE_SHIFT);
            if (phase < 0)
                return phase;
            int counts = (int)s;
            int unarrived = (counts == EMPTY) ? 0 : (counts & UNARRIVED_MASK);
            if (unarrived <= 0)
                throw new IllegalStateException(badArrive(s));
            // 将state减去adjust
            if (STATE.compareAndSet(this, s, s-=adjust)) {
                // 减到最后一个线程
                if (unarrived == 1) {
                    long n = s & PARTIES_MASK;
                    // 将总线程数赋值到未到达的线程数, 从而实现Phaser复用
                    int nextUnarrived = (int)n >>> PARTIES_SHIFT;
                    if (root == this) {
                        if (onAdvance(phase, nextUnarrived))
                            n |= TERMINATION_BIT;
                        else if (nextUnarrived == 0)
                            n |= EMPTY;
                        else
                            n |= nextUnarrived;
                        // 先将phase+1, 再修改当前state
                        int nextPhase = (phase + 1) & MAX_PHASE;
                        n |= (long)nextPhase << PHASE_SHIFT;
                        STATE.compareAndSet(this, s, n);
                        // 释放当前phase的线程
                        releaseWaiters(phase);
                    }
                    // 当前Phaser总线程数为0, 说明当前线程已经全部取消注册, 所以通知父级Phaser减65537(反注册一个总线程同时将未到达线程数减一)
                    else if (nextUnarrived == 0) { // propagate deregistration
                        phase = parent.doArrive(ONE_DEREGISTER);
                        STATE.compareAndSet(this, s, s | EMPTY);
                    }
                    else
                        // 通知父级Phaser减1.
                        phase = parent.doArrive(ONE_ARRIVAL);
                }
                return phase;
            }
        }
    }
}

public class ArrayBlockingQueue<E> extends AbstractQueue<E> implements BlockingQueue<E>, java.io.Serializable {

    final Object[] items;
    int takeIndex;
    int putIndex;
    int count;

    final ReentrantLock lock;
    private final Condition notEmpty;
    private final Condition notFull;

    public ArrayBlockingQueue(int capacity, boolean fair) {
        if (capacity <= 0)
            throw new IllegalArgumentException();
        this.items = new Object[capacity];
        lock = new ReentrantLock(fair);
        notEmpty = lock.newCondition();
        notFull =  lock.newCondition();
    }

    public boolean add(E e) {
        if (offer(e))
            return true;
        else
            // 队列满了抛出异常
            throw new IllegalStateException("Queue full");
    }

    public boolean offer(E e) {
        Objects.requireNonNull(e);
        final ReentrantLock lock = this.lock;
        lock.lock();
        try {
            if (count == items.length)
                return false;
            else {
                enqueue(e);
                return true;
            }
        } finally {
            lock.unlock();
        }
    }

    private void enqueue(E e) {
        final Object[] items = this.items;
        // 将元素插入到items数组里
        items[putIndex] = e;
        // putIndex即为队列尾部索引
        if (++putIndex == items.length) putIndex = 0;
        count++;
        notEmpty.signal();
    }

    public void put(E e) throws InterruptedException {
        Objects.requireNonNull(e);
        final ReentrantLock lock = this.lock;
        lock.lockInterruptibly();
        try {
            // 如果数组满了则阻塞住
            while (count == items.length)
                notFull.await();
            enqueue(e);
        } finally {
            lock.unlock();
        }
    }

    public E remove() {
        E x = poll();
        if (x != null)
            return x;
        // 如果数组满了则抛出异常
        else
            throw new NoSuchElementException();
    }

    public E poll() {
        final ReentrantLock lock = this.lock;
        lock.lock();
        try {
            return (count == 0) ? null : dequeue();
        } finally {
            lock.unlock();
        }
    }

    private E dequeue() {
        final Object[] items = this.items;
        @SuppressWarnings("unchecked")
        E e = (E) items[takeIndex];
        // 删除数组元素
        items[takeIndex] = null;
        // takeIndex即为队列头部索引
        if (++takeIndex == items.length) takeIndex = 0;
        count--;
        if (itrs != null)
            itrs.elementDequeued();
        // 唤醒阻塞在notFull的线程
        notFull.signal();
        return e;
    }

    public E take() throws InterruptedException {
        final ReentrantLock lock = this.lock;
        lock.lockInterruptibly();
        try {
            // 数组为空时阻塞
            while (count == 0)
                notEmpty.await();
            return dequeue();
        } finally {
            lock.unlock();
        }
    }

    public E element() {
        E x = peek();
        if (x != null)
            return x;
        // 数组为空时抛出异常
        else
            throw new NoSuchElementException();
    }

    public E peek() {
        final ReentrantLock lock = this.lock;
        lock.lock();
        try {
            // 直接返回takeIndex所在元素(队列头部)
            return itemAt(takeIndex); // null when queue is empty
        } finally {
            lock.unlock();
        }
    }

}

public class LinkedBlockingQueue<E> extends AbstractQueue<E> implements BlockingQueue<E>, java.io.Serializable {

    // 单向链表节点
    static class Node<E> {
        E item;
        Node<E> next;
        Node(E x) { item = x; }
    }

    private final int capacity;
    private final AtomicInteger count = new AtomicInteger();
    transient Node<E> head;
    private transient Node<E> last;

    private final ReentrantLock takeLock = new ReentrantLock();
    private final ReentrantLock putLock = new ReentrantLock();
    private final Condition notEmpty = takeLock.newCondition();
    private final Condition notFull = putLock.newCondition();

    public boolean offer(E e) {
        if (e == null) throw new NullPointerException();
        final AtomicInteger count = this.count;
        if (count.get() == capacity)
            return false;
        final int c;
        final Node<E> node = new Node<E>(e);
        final ReentrantLock putLock = this.putLock;
        putLock.lock();
        try {
            // 判断队列是否满了
            if (count.get() == capacity)
                return false;
            // 加入到队尾
            enqueue(node);
            c = count.getAndIncrement();
            if (c + 1 < capacity)
                notFull.signal();
        } finally {
            putLock.unlock();
        }
        if (c == 0)
            signalNotEmpty();
        return true;
    }

    public void put(E e) throws InterruptedException {
        if (e == null) throw new NullPointerException();
        final int c;
        final Node<E> node = new Node<E>(e);
        final ReentrantLock putLock = this.putLock;
        final AtomicInteger count = this.count;
        putLock.lockInterruptibly();
        try {
            // 队列满了阻塞
            while (count.get() == capacity) {
                notFull.await();
            }
            enqueue(node);
            c = count.getAndIncrement();
            if (c + 1 < capacity)
                // LinkedBlockingQueue的put也会唤醒其他的put操作
                notFull.signal();
        } finally {
            putLock.unlock();
        }
        if (c == 0)
            signalNotEmpty();
    }

    public E poll() {
        final AtomicInteger count = this.count;
        if (count.get() == 0)
            return null;
        final E x;
        final int c;
        final ReentrantLock takeLock = this.takeLock;
        takeLock.lock();
        try {
            // 队列为空返回null
            if (count.get() == 0)
                return null;
            x = dequeue();
            c = count.getAndDecrement();
            if (c > 1)
                // LinkedBlockingQueue的poll也会唤醒其他的poll操作
                notEmpty.signal();
        } finally {
            takeLock.unlock();
        }
        if (c == capacity)
            // 唤醒阻塞在notFull的线程
            signalNotFull();
        return x;
    }

    private void enqueue(Node<E> node) {
        last = last.next = node;
    }

    private E dequeue() {
        Node<E> h = head;
        Node<E> first = h.next;
        h.next = h; // help GC
        head = first;
        // head为标识队列头的哑元节点, 此处将head.next变为head, 并将item清除
        E x = first.item;
        first.item = null;
        return x;
    }

}

public class PriorityBlockingQueue<E> extends AbstractQueue<E> implements BlockingQueue<E>, java.io.Serializable {

    private transient Object[] queue;

    private transient Comparator<? super E> comparator;

    private final ReentrantLock lock = new ReentrantLock();
    // PriorityBlockingQueue因为会无限增长, 所以只会有notEmpty的等待条件
    private final Condition notEmpty = lock.newCondition();

    public PriorityBlockingQueue(int initialCapacity,
                                 Comparator<? super E> comparator) {
        if (initialCapacity < 1)
            throw new IllegalArgumentException();
        this.comparator = comparator;
        this.queue = new Object[Math.max(1, initialCapacity)];
    }

    // 不存在插入不了的情况, 所以add/put都直接调用的offer
    public boolean add(E e) {
        return offer(e);
    }

    public void put(E e) {
        offer(e); // never need to block
    }

    public boolean offer(E e) {
        if (e == null)
            throw new NullPointerException();
        final ReentrantLock lock = this.lock;
        lock.lock();
        int n, cap;
        Object[] es;
        // 如果queue数组满了, 则扩容
        while ((n = size) >= (cap = (es = queue).length))
            tryGrow(es, cap);
        try {
            final Comparator<? super E> cmp;
            if ((cmp = comparator) == null)
                siftUpComparable(n, e, es);
            else
                siftUpUsingComparator(n, e, es, cmp);
            size = n + 1;
            notEmpty.signal();
        } finally {
            lock.unlock();
        }
        return true;
    }

    private void tryGrow(Object[] array, int oldCap) {
        lock.unlock(); // must release and then re-acquire main lock
        Object[] newArray = null;
        if (allocationSpinLock == 0 &&
            ALLOCATIONSPINLOCK.compareAndSet(this, 0, 1)) {
            try {
                // 扩容两倍或直接+2
                int growth = oldCap < 64 ? oldCap + 2 : oldCap >> 1;
                int newCap = ArraysSupport.newLength(oldCap, 1, growth);
                if (queue == array)
                    newArray = new Object[newCap];
            } finally {
                allocationSpinLock = 0;
            }
        }
        if (newArray == null) // back off if another thread is allocating
            Thread.yield();
        lock.lock();
        if (newArray != null && queue == array) {
            queue = newArray;
            System.arraycopy(array, 0, newArray, 0, oldCap);
        }
    }

    public E remove() {
        E x = poll();
        if (x != null)
            return x;
        else
            throw new NoSuchElementException();
    }

    public E poll() {
        final ReentrantLock lock = this.lock;
        lock.lock();
        try {
            return dequeue();
        } finally {
            lock.unlock();
        }
    }

    public E take() throws InterruptedException {
        final ReentrantLock lock = this.lock;
        lock.lockInterruptibly();
        E result;
        try {
            // 如果队列没有元素, 则阻塞
            while ( (result = dequeue()) == null)
                notEmpty.await();
        } finally {
            lock.unlock();
        }
        return result;
    }

    // 二叉堆删除元素
    private E dequeue() {
        // assert lock.isHeldByCurrentThread();
        final Object[] es;
        final E result;

        if ((result = (E) ((es = queue)[0])) != null) {
            final int n;
            final E x = (E) es[(n = --size)];
            es[n] = null;
            if (n > 0) {
                final Comparator<? super E> cmp;
                if ((cmp = comparator) == null)
                    siftDownComparable(0, x, es, n);
                else
                    siftDownUsingComparator(0, x, es, n, cmp);
            }
        }
        return result;
    }

}

public interface Delayed extends Comparable<Delayed> {
    long getDelay(TimeUnit unit);
}

public class DelayQueue<E extends Delayed> extends AbstractQueue<E> implements BlockingQueue<E> {
    private Thread leader;
    private final PriorityQueue<E> q = new PriorityQueue<E>();
    private final transient ReentrantLock lock = new ReentrantLock();
    // queue非空的条件变量
    private final Condition available = lock.newCondition();

    // 与PriorityBlokingQueue一样, 没有容量的限制, 所以add/put都直接调用的offer
    public boolean add(E e) {
        return offer(e);
    }

    public void put(E e) {
        offer(e);
    }

    public boolean offer(E e) {
        final ReentrantLock lock = this.lock;
        lock.lock();
        try {
            q.offer(e);
            // 只有当堆顶的元素为新插入的元素才需要唤醒阻塞在take的线程
            if (q.peek() == e) {
                leader = null;
                available.signal();
            }
            return true;
        } finally {
            lock.unlock();
        }
    }

    public E remove() {
        E x = poll();
        if (x != null)
            return x;
        else
            throw new NoSuchElementException();
    }

    public E poll() {
        final ReentrantLock lock = this.lock;
        lock.lock();
        try {
            E first = q.peek();
            // 当第一个元素的超时时间大于0的时候, 直接返回null, 否则返回第一个元素
            return (first == null || first.getDelay(NANOSECONDS) > 0)
                ? null
                : q.poll();
        } finally {
            lock.unlock();
        }
    }

    public E take() throws InterruptedException {
        final ReentrantLock lock = this.lock;
        lock.lockInterruptibly();
        try {
            for (;;) {
                // 取出延迟时间最小的元素
                E first = q.peek();
                // 如果为空则阻塞
                if (first == null)
                    available.await();
                else {
                    long delay = first.getDelay(NANOSECONDS);
                    if (delay <= 0L)
                        return q.poll();
                    first = null; // don't retain ref while waiting
                    if (leader != null)
                        // 如果有其他线程在等待元素, 则阻塞住.
                        available.await();
                    else {
                        Thread thisThread = Thread.currentThread();
                        leader = thisThread;
                        try {
                            // 等待队列头部元素的delay时间
                            available.awaitNanos(delay);
                        } finally {
                            if (leader == thisThread)
                                leader = null;
                        }
                    }
                }
            }
        } finally {
            if (leader == null && q.peek() != null)
                // 取出队列头部的元素后, 唤醒其他阻塞在take的线程
                available.signal();
            lock.unlock();
        }
    }
}

public class SynchronousQueue<E> extends AbstractQueue<E> implements BlockingQueue<E>, java.io.Serializable {
    public SynchronousQueue() {
        this(false);
    }

    public SynchronousQueue(boolean fair) {
        // 默认为非公平
        transferer = fair ? new TransferQueue<E>() : new TransferStack<E>();
    }

    public void put(E e) throws InterruptedException {
        if (e == null) throw new NullPointerException();
        // 插入元素时第一个参数传这个元素
        if (transferer.transfer(e, false, 0) == null) {
            Thread.interrupted();
            throw new InterruptedException();
        }
    }

    public boolean offer(E e) {
        if (e == null) throw new NullPointerException();
        return transferer.transfer(e, true, 0) != null;
    }

    public E poll() {
        // 取出元素时第一个参数传null
        return transferer.transfer(null, true, 0);
    }

    public E take() throws InterruptedException {
        E e = transferer.transfer(null, false, 0);
        if (e != null)
            return e;
        Thread.interrupted();
        throw new InterruptedException();
    }

}

public class LinkedBlockingDeque<E> extends AbstractQueue<E> implements BlockingDeque<E>, java.io.Serializable {

    // 双向链表节点
    static final class Node<E> {
        E item;
        Node<E> prev;
        Node<E> next;
        Node(E x) {
            item = x;
        }
    }

    // 头节点
    transient Node<E> first;
    // 尾节点
    transient Node<E> last;
    // 节点总数
    private transient int count;
    // 链表容量
    private final int capacity;

    final ReentrantLock lock = new ReentrantLock();
    private final Condition notEmpty = lock.newCondition();
    private final Condition notFull = lock.newCondition();

    public LinkedBlockingDeque() {
        this(Integer.MAX_VALUE);
    }

    public LinkedBlockingDeque(int capacity) {
        if (capacity <= 0) throw new IllegalArgumentException();
        this.capacity = capacity;
    }

    public void putFirst(E e) throws InterruptedException {
        if (e == null) throw new NullPointerException();
        Node<E> node = new Node<E>(e);
        final ReentrantLock lock = this.lock;
        lock.lock();
        try {
            // 队列满了会阻塞
            while (!linkFirst(node))
                notFull.await();
        } finally {
            lock.unlock();
        }
    }

    private boolean linkFirst(Node<E> node) {
        if (count >= capacity)
            return false;
        Node<E> f = first;
        // node的next为当前的头节点
        node.next = f;
        // 将头节点改为node
        first = node;
        if (last == null)
            last = node;
        else
            f.prev = node;
        ++count;
        // 唤醒阻塞在notEmpty条件变量的线程
        notEmpty.signal();
        return true;
    }

    public E takeLast() throws InterruptedException {
        final ReentrantLock lock = this.lock;
        lock.lock();
        try {
            E x;
            // 链表空时会阻塞
            while ( (x = unlinkLast()) == null)
                notEmpty.await();
            return x;
        } finally {
            lock.unlock();
        }
    }

    private E unlinkLast() {
        Node<E> l = last;
        if (l == null)
            return null;
        Node<E> p = l.prev;
        E item = l.item;
        l.item = null;
        // 改掉last.prev指针, 防止强引用无法回收last节点
        l.prev = l;
        // last = last.prev
        last = p;
        if (p == null)
            first = null;
        else
            p.next = null;
        --count;
        // 唤醒阻塞在notFull条件变量的线程
        notFull.signal();
        return item;
    }
}

public class CopyOnWriteArrayList<E> implements List<E>, RandomAccess, Cloneable, java.io.Serializable {

    // 写操作的锁对象(synchronized锁)
    final transient Object lock = new Object();

    private transient volatile Object[] array;

    public boolean add(E e) {
        synchronized (lock) {
            Object[] es = getArray();
            int len = es.length;
            // 先拷贝出来一份数组
            es = Arrays.copyOf(es, len + 1);
            // 写入到拷贝出来的数组
            es[len] = e;
            // 重新赋值数组
            setArray(es);
            return true;
        }
    }

    public boolean addIfAbsent(E e) {
        Object[] snapshot = getArray();
        // 先不加锁判断数组是否包含该元素, 然后加锁再判断一次, 如果不存在则插入元素
        return indexOfRange(e, snapshot, 0, snapshot.length) < 0
            && addIfAbsent(e, snapshot);
    }

    public E remove(int index) {
        synchronized (lock) {
            Object[] es = getArray();
            int len = es.length;
            E oldValue = elementAt(es, index);
            int numMoved = len - index - 1;
            Object[] newElements;
            if (numMoved == 0)
                // 如果删除的是最后一个元素, 则直接复制数组的0~len-1位
                newElements = Arrays.copyOf(es, len - 1);
            else {
                // 将删除位置的左右两侧复制到新的数组里
                newElements = new Object[len - 1];
                System.arraycopy(es, 0, newElements, 0, index);
                System.arraycopy(es, index + 1, newElements, index,
                                 numMoved);
            }
            setArray(newElements);
            return oldValue;
        }
    }
}

public class CopyOnWriteArraySet<E> extends AbstractSet<E> implements java.io.Serializable {

    private final CopyOnWriteArrayList<E> al;

    public CopyOnWriteArraySet() {
        al = new CopyOnWriteArrayList<E>();
    }

    // 判断不存在则插入元素
    public boolean add(E e) {
        return al.addIfAbsent(e);
    }
}

    private static void shutdown() {
        ExecutorService executorService = Executors.newFixedThreadPool(2);
        executorService.execute(() -> {
            SleepUtil.sleep(1000);
            log.info("sleep 1s");
        });
        executorService.execute(() -> {
            SleepUtil.sleep(2000);
            log.info("sleep 2s");
        });
        executorService.execute(() -> {
            SleepUtil.sleep(3000);
            log.info("sleep 3s");
        });
        executorService.shutdown(); (1)
        log.info("shutdown called");
        executorService.execute(() -> System.out.println("no more")); (2)
        (3)
    }

    private static void shutdownNow() {
        ExecutorService executorService = Executors.newFixedThreadPool(2);
        executorService.execute(() -> {
            SleepUtil.sleep(10000);
            log.info("sleep 10s");
        });
        executorService.execute(() -> {
            SleepUtil.sleep(20000);
            log.info("sleep 20s");
        });
        executorService.execute(() -> {
            SleepUtil.sleep(30000);
            log.info("sleep 30s");
        });
        List<Runnable> runnables = executorService.shutdownNow(); (1)
        log.info("shutdownNow: {}", runnables); (2)
        log.info("shutdown called");
        executorService.execute(() -> System.out.println("no more")); (3)
        (4)
    }

public class ThreadPoolExecutor extends AbstractExecutorService {

    // 前3位标识线程池的状态, 后29位标识线程数
    private static final int COUNT_BITS = Integer.SIZE - 3;
    private static final int COUNT_MASK = (1 << COUNT_BITS) - 1;

    // 线程池的5种状态常量
    private static final int RUNNING    = -1 << COUNT_BITS;
    private static final int SHUTDOWN   =  0 << COUNT_BITS;
    private static final int STOP       =  1 << COUNT_BITS;
    private static final int TIDYING    =  2 << COUNT_BITS;
    private static final int TERMINATED =  3 << COUNT_BITS;

    // 线程池的当前状态
    private final AtomicInteger ctl = new AtomicInteger(ctlOf(RUNNING, 0));

    private final ReentrantLock mainLock = new ReentrantLock();
    private final Condition termination = mainLock.newCondition();

    public void shutdown() {
        final ReentrantLock mainLock = this.mainLock;
        mainLock.lock();
        try {
            checkShutdownAccess();
            // 设置线程池状态为SHUTDOWN
            advanceRunState(SHUTDOWN);
            // 打断空闲线程
            interruptIdleWorkers();
            onShutdown(); // hook for ScheduledThreadPoolExecutor
        } finally {
            mainLock.unlock();
        }
        tryTerminate();
    }

    public List<Runnable> shutdownNow() {
        List<Runnable> tasks;
        final ReentrantLock mainLock = this.mainLock;
        mainLock.lock();
        try {
            checkShutdownAccess();
            // 设置线程池状态为STOP
            advanceRunState(STOP);
            // 打断所有线程
            interruptWorkers();
            tasks = drainQueue();
        } finally {
            mainLock.unlock();
        }
        tryTerminate();
        return tasks;
    }

    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();
        }
    }

    private void interruptWorkers() {
        for (Worker w : workers)
            // 不管能不能获得锁, 都直接打断该线程
            w.interruptIfStarted();
    }

    private List<Runnable> drainQueue() {
        BlockingQueue<Runnable> q = workQueue;
        ArrayList<Runnable> taskList = new ArrayList<>();
        // 将阻塞队列里的线程对象放入到taskList里
        q.drainTo(taskList);
        if (!q.isEmpty()) {
            for (Runnable r : q.toArray(new Runnable[0])) {
                // 如果是DelayQueue, drainTo方法可能会没有把队列元素全放入taskList里, 此处手动删除再加入到taskList里
                if (q.remove(r))
                    taskList.add(r);
            }
        }
        return taskList;
    }

    final void tryTerminate() {
        for (;;) {
            int c = ctl.get();
            if (isRunning(c) ||
                runStateAtLeast(c, TIDYING) ||
                (runStateLessThan(c, STOP) && ! workQueue.isEmpty()))
                return;
            if (workerCountOf(c) != 0) { // Eligible to terminate
                interruptIdleWorkers(ONLY_ONE);
                return;
            }

            final ReentrantLock mainLock = this.mainLock;
            mainLock.lock();
            try {
                // 阻塞队列为空并且没有工作线程时, 设置线程池状态为TIDYING
                if (ctl.compareAndSet(c, ctlOf(TIDYING, 0))) {
                    try {
                        // 执行terminated回调
                        terminated();
                    } finally {
                        // 设置线程池状态为TERMINATED
                        ctl.set(ctlOf(TERMINATED, 0));
                        // 唤醒阻塞在awaitTermination方法的线程
                        termination.signalAll();
                    }
                    return;
                }
            } finally {
                mainLock.unlock();
            }
            // else retry on failed CAS
        }
    }

    // 等待线程池状态变为TERMINATED
    public boolean awaitTermination(long timeout, TimeUnit unit) throws InterruptedException {
        long nanos = unit.toNanos(timeout);
        final ReentrantLock mainLock = this.mainLock;
        mainLock.lock();
        try {
            // 如果当前线程池的状态小于TERMINATED(3), 则阻塞
            while (runStateLessThan(ctl.get(), TERMINATED)) {
                if (nanos <= 0L)
                    return false;
                nanos = termination.awaitNanos(nanos);
            }
            return true;
        } finally {
            mainLock.unlock();
        }
    }

}

public class ThreadPoolExecutor extends AbstractExecutorService {

    public void execute(Runnable command) {
        if (command == null)
            throw new NullPointerException();
        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();
            // 如果线程池状态不是RUNNING, 则拒绝执行新提交的任务
            if (! isRunning(recheck) && remove(command))
                reject(command);
            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();;) {
            if (runStateAtLeast(c, SHUTDOWN)
                && (runStateAtLeast(c, STOP)
                    || firstTask != null
                    || workQueue.isEmpty()))
                return false;

            for (;;) {
                // 大于核心线程数or最大线程数
                if (workerCountOf(c)
                    >= ((core ? corePoolSize : maximumPoolSize) & COUNT_MASK))
                    return false;
                // 工作线程计数加1
                if (compareAndIncrementWorkerCount(c))
                    break retry;
                c = ctl.get();  // Re-read ctl
                if (runStateAtLeast(c, SHUTDOWN))
                    continue retry;
            }
        }

        boolean workerStarted = false;
        boolean workerAdded = false;
        Worker w = null;
        try {
            w = new Worker(firstTask);
            final Thread t = w.thread;
            if (t != null) {
                final ReentrantLock mainLock = this.mainLock;
                mainLock.lock();
                try {
                    int c = ctl.get();

                    if (isRunning(c) ||
                        (runStateLessThan(c, STOP) && firstTask == null)) {
                        if (t.getState() != Thread.State.NEW)
                            throw new IllegalThreadStateException();
                        // 将线程对象添加到workers集合中
                        workers.add(w);
                        workerAdded = true;
                        int s = workers.size();
                        if (s > largestPoolSize)
                            largestPoolSize = s;
                    }
                } finally {
                    mainLock.unlock();
                }
                if (workerAdded) {
                    // 执行该工作线程
                    t.start();
                    workerStarted = true;
                }
            }
        } finally {
            if (! workerStarted)
                addWorkerFailed(w);
        }
        return workerStarted;
    }
}

public class ThreadPoolExecutor extends AbstractExecutorService {

    final void runWorker(Worker w) {
        Thread wt = Thread.currentThread();
        Runnable task = w.firstTask;
        w.firstTask = null;
        w.unlock();
        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);
                    try {
                        task.run();
                        afterExecute(task, null);
                    } catch (Throwable ex) {
                        afterExecute(task, ex);
                        throw ex;
                    }
                } finally {
                    task = null;
                    w.completedTasks++;
                    w.unlock();
                }
            }
            completedAbruptly = false;
        } finally {
            processWorkerExit(w, completedAbruptly);
        }
    }
}