http://blog.dengqinghua.net

BLOCKED 是在等待排他锁, 而 WAITING 是被调用了 `Object#wait()`, `Thread#join()` or `LockSupport#park()` 方法,
而处于等待状态, 并且可以通过 `notify` 或者 `notifyAll` 方法进行唤醒.

while (!needHandle) {
    // doNothing
}

doThing

synchronized { // monitor region begin, 即 monitorenter
    doThingA;
    ...
}              // monitor region end, 即 monitorexit

synchronized(this) {
    doSthA

    synchronized(this) {
        doSthA
    }
}

public class ThreadPool {
    private static final int THREAD_COUNT = 100;
    private static final Executor executor = Executors.newFixedThreadPool(THREAD_COUNT);
    public static void runMuiltThreadServerWithThreadPool() throws IOException {
        ServerSocket socket = new ServerSocket(10080);

        while (true) {
            final Socket connection = socket.accept();
            // 这里采用了线程池的方式
            executor.execute(() -> handleConnection(connection));
        }
    }
}

public class ThreadPool {
    private static final int THREAD_COUNT = 100;
    private static final Executor executor = Executors.newFixedThreadPool(THREAD_COUNT);

    public static void runMuiltThreadServerWithThreadPool() throws IOException {
        ServerSocket socket = new ServerSocket(10080);

        while (true) {
            final Socket connection = socket.accept();
            // 这里采用了线程池的方式
            executor.execute(() -> handleConnection(connection));
        }
    }
}

 RUNNING -> SHUTDOWN
    On invocation of shutdown(), perhaps implicitly in finalize()
 (RUNNING or SHUTDOWN) -> STOP
    On invocation of shutdownNow()
 SHUTDOWN -> TIDYING
    When both queue and pool are empty
 STOP -> TIDYING
    When pool is empty
 TIDYING -> TERMINATED
    When the terminated() hook method has completed

// 状态信息存储在第30位-第32位
int COUNT_BITS = Integer.SIZE - 3, // 29位
        CAPACITY = (1 << count_bits) - 1;

        int RUNNING = -1 << COUNT_BITS,       // 11100000000000000000000000000000 -536870912
                SHUTDOWN = 0,
                STOP = 1 << COUNT_BITS,       // 00100000000000000000000000000000 536870912
                TIDYING = 2 << COUNT_BITS,    // 01000000000000000000000000000000 1073741824
                TERMINATED = 3 << COUNT_BITS; // 01100000000000000000000000000000 1610612736

// assertThat(Integeer.toBinaryString(CAPACITY), is("11111111111111111111111111111"));
// assertThat(Integer.toBinaryString(CAPACITY).length(), is(29));

// ctl 包含两部分: runState 为 第30位 至 第32位, workerCount 为 第1位 到 第29位
// 故获取 runState 只需要高位(30-32)信息 为 ctl & ~CAPACITY
// 故获取 workerCount 只需要低位(1-29)信息 为 ctl & CAPACITY

// 这里的 c 为 ctl, 该方法从 ctl 解析出当前的状态, 如 running/shutdown等
private static int runStateOf(int c)     { return c & ~CAPACITY; }

// 这里的 c 为 ctl, 该方法从 ctl 解析出当前的workerCount
private static int workerCountOf(int c)  { return c & CAPACITY; }

// rs 代表 runState, 如上所述的 running/shutdown/stop/tidying/termiated 等值
// wc 代表 workerCount
// 通过 rs 和 wc 得到 ctl 的值
private static int ctlOf(int rs, int wc) { return rs | wc; }

// 初始化值为 11100000000000000000000000000000 -536870912, 即 -2^29
private final AtomicInteger ctl = new AtomicInteger(ctlOf(RUNNING, 0));
// rs 代表 runState, 如上所述的 running/shutdown/stop/tidying/termiated 等值
// wc 代表 workerCount
private static int ctlOf(int rs, int wc) { return rs | wc; }

/**
 * Set containing all worker threads in pool. Accessed only when
 * holding mainLock.
 */
private final HashSet<Worker> workers = new HashSet<Worker>();

/**
 * The queue used for holding tasks and handing off to worker
 * threads.  We do not require that workQueue.poll() returning
 * null necessarily means that workQueue.isEmpty(), so rely
 * solely on isEmpty to see if the queue is empty (which we must
 * do for example when deciding whether to transition from
 * SHUTDOWN to TIDYING).  This accommodates special-purpose
 * queues such as DelayQueues for which poll() is allowed to
 * return null even if it may later return non-null when delays
 * expire.
 */
private final BlockingQueue<Runnable> workQueue;

final void runWorker(Worker w) {
    Thread wt = Thread.currentThread();
    Runnable task = w.firstTask;
    w.firstTask = null;
    w.unlock(); // allow interrupts
    boolean completedAbruptly = true;
    try {
        // 不停地获取任务, 直到没有任务可以获取
        while (task != null || (task = getTask()) != null) {
            ...
            task.run();
            ...
        }
    }
}

private Runnable getTask() {
    ...
    try {
        Runnable r = timed ?
            // 如果设置了超时, 就使用 poll 方法获取 task
            workQueue.poll(keepAliveTime, TimeUnit.NANOSECONDS) :
            // 如果没有设置超时, 就使用 take 方法获取 task
            workQueue.take();
        if (r != null)
            return r;
    } catch (InterruptedException retry) {
        ...
    }
}

private final class Worker
    extends AbstractQueuedSynchronizer
    // 这个Worker 实现了 Runnable 接口!
    implements Runnable {

    // 这个线程是用来执行任务的
    final Thread thread;

    public void run() {
        runWorker(this);
    }
}

# @mgr 即为他对应的 Manager 对象
class Processor
  def run
    begin
      while !@done
        # 调用 perform 方法进行处理
        process_one
      end

      # 一旦结束了, 则将 Processor对象中的manager对应的worker去掉, 即是改变了上述 Manager的 @workers 数组
      @mgr.processor_stopped(self)
    rescue Sidekiq::Shutdown
      # 在接收到TERM SIGNAL之后, 等待超时的时候sidekiq会抛出异常 Sidekiq::Shutdown, 见下文分析
      # 线程被关闭.
      @mgr.processor_stopped(self)
    rescue Exception => ex
      # 程序报错了, Manager#processor_died 会重新生成一个新的Processor线程
      @mgr.processor_died(self, ex)
    end
  end
end

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

    public boolean offer(E e, long timeout, TimeUnit unit)
        throws InterruptedException {

        if (e == null) throw new NullPointerException();
        long nanos = unit.toNanos(timeout);

        // Note: convention in all put/take/etc is to preset local var
        // holding count negative to indicate failure unless set.
        // 这个是约定, 设置为一个负值, 并没有特殊的含义...
        int c = -1;
        final ReentrantLock putLock = this.putLock;
        final AtomicInteger count = this.count;

        // 添加 putLock 的锁
        putLock.lockInterruptibly();

        try {
            // 如果队列已经满了, 则进行等待一段时间
            // 直到队列的count小于capacity 为止
            while (count.get() == capacity) {
                if (nanos <= 0)
                    return false;
                nanos = notFull.awaitNanos(nanos);
            }

            // 进行入队列操作, 该操作是不添加锁的, 仅仅修改了 last
            // last = last.next = node;
            enqueue(new Node<E>(e));

            // 原子性的进行加1
            c = count.getAndIncrement();

            // 如果发现队列未满, 则发 notFull 的 signal
            // TODO: signal的信号机制, ReentrantLock 还需要再看一下
            if (c + 1 < capacity)
                notFull.signal();
        } finally {
            // 解锁
            putLock.unlock();
        }
        if (c == 0)
            signalNotEmpty();
        return true;
    }
}

public class LinkedBlockingQueue<E> extends AbstractQueue<E>
    implements BlockingQueue<E>, java.io.Serializable {
    public E poll(long timeout, TimeUnit unit) throws InterruptedException {
        E x = null;
        int c = -1;
        long nanos = unit.toNanos(timeout);
        final AtomicInteger count = this.count;
        final ReentrantLock takeLock = this.takeLock;

        // 添加 takeLock 锁
        takeLock.lockInterruptibly();
        try {
            while (count.get() == 0) {
                if (nanos <= 0)
                    return null;
                nanos = notEmpty.awaitNanos(nanos);
            }

            // 从队列头部获得数据
            x = dequeue();
            c = count.getAndDecrement();
            if (c > 1)
                notEmpty.signal();
        } finally {
            takeLock.unlock();
        }
        if (c == capacity)
            signalNotFull();
        return x;
    }
}

public void shutdown() {
    final ReentrantLock mainLock = this.mainLock;
    mainLock.lock();
    try {
        checkShutdownAccess();
        advanceRunState(SHUTDOWN);
        interruptIdleWorkers();
        onShutdown(); // hook for ScheduledThreadPoolExecutor
    } finally {
        mainLock.unlock();
    }
    tryTerminate();
}

while (true) {
    final Socket connection = socket.accept();
    Future task = executorFuture.submit(() -> handleConnection(connection));
    System.out.println(task.get());    // 如果没有结果则堵塞
    System.out.println(task.isDone()); // 返回true
}

Future task = executorFuture.submit(() -> handleConnection(connection));

// java/util/concurrent/AbstractExecutorService.java:114
public Future<?> submit(Runnable task) {
    if (task == null) throw new NullPointerException();
    RunnableFuture<Void> ftask = newTaskFor(task, null);
    execute(ftask);
    return ftask;
}

public FutureTask(Runnable runnable, V result) {
    // 将需要执行的任务存储起来
    this.callable = Executors.callable(runnable, result);
    this.state = NEW;       // ensure visibility of callable
}

public class FutureTask<V> implements RunnableFuture<V> {
    public void run() {
        if (state != NEW ||
            !UNSAFE.compareAndSwapObject(this, runnerOffset,
                                         null, Thread.currentThread()))
            return;
        try {
            Callable<V> c = callable;
            if (c != null && state == NEW) {
                V result;
                boolean ran;
                try {
                    // 在这里调用真正的需要执行的任务
                    result = c.call();
                    // 到这儿就说明已经执行完了
                    ran = true;
                } catch (Throwable ex) {
                    result = null;
                    ran = false;
                    setException(ex);
                }
                if (ran)
                    set(result);
            }
        } finally {
            ....
        }
    }
}

/**
 * The run state of this task, initially NEW.  The run state
 * transitions to a terminal state only in methods set,
 * setException, and cancel.  During completion, state may take on
 * transient values of COMPLETING (while outcome is being set) or
 * INTERRUPTING (only while interrupting the runner to satisfy a
 * cancel(true)). Transitions from these intermediate to final
 * states use cheaper ordered/lazy writes because values are unique
 * and cannot be further modified.
 *
 * Possible state transitions:
 * NEW -> COMPLETING -> NORMAL
 * NEW -> COMPLETING -> EXCEPTIONAL
 * NEW -> CANCELLED
 * NEW -> INTERRUPTING -> INTERRUPTED
 */
private volatile int state;
private static final int NEW          = 0;
private static final int COMPLETING   = 1;
private static final int NORMAL       = 2;
private static final int EXCEPTIONAL  = 3;
private static final int CANCELLED    = 4;
private static final int INTERRUPTING = 5;
private static final int INTERRUPTED  = 6;