瞭解一下JUC的內部實現

時間 2019-11-10

標籤瞭解一下 juc 內部實現欄目 Java 简体版

原文原文鏈接

AQS

關於CLH大量使用到的Unsafe的CAS用法，頭兩個入參是this和xxOffset，翻了一下牛逼網友的給的代碼大概是處理一個內存對齊的問題，整個操做中涉及到offset(dest)有兩個部分java
```
mov edx, dest
.....
cmpxchg dword ptr [edx], ecx   ;ecx寄存器放置exchange_value
```
Unsafe不面向普通開發者，上來就檢查你的類加載器是否是null(native)node

先mark一下這句話，其中AbstractOwnableSynchronizer就是保存有排斥用的Thread成員數組

* You may also find the inherited methods from {@link
* AbstractOwnableSynchronizer} useful to keep track of the thread
* owning an exclusive synchronizer.  You are encouraged to use them
* -- this enables monitoring and diagnostic tools to assist users in
* determining which threads hold locks.

A thread may try to acquire if it is first in the queue.（這是一個FIFO機制）併發
CLH鎖的入隊是原子性的（源碼中使用CAS（新的節點，tail）實現替換） Insertion into a CLH queue requires only a single atomic operation on "tail"，且出隊也是原子性的，dequeuing involves only updating the "head"，但還須要處理後繼 in part to deal with possible cancellation due to timeouts and interrupts，全部的信息都用volatile的waitStatus來表示，比方說取消（timeout or interrupt）是1，SIGNAL（當前的後繼須要喚醒，注意有特殊的要求， unpark its successor when it releases or cancels）爲-1，而-2 / -3 涉及到Condition的設計，這裏暫且保留說明app
鏈表設計中prev用於處理取消操做（LAZY），next用於處理阻塞操做，當須要wakeup時就沿着next來跑（其中有一點checking backwards from the atomically updated "tail" when a node's successor appears to be null的情形暫留-->UPD.問題後面說明了）less
nextWaiter和next有必定區別，前者是一個簡單的node，然後者是volatile，具體用途彷佛不止一種，有一種用法是判斷是否共享/獨佔nextWaiter == SHARED工具
state和status又有啥區別啊（The synchronization state.好含糊啊，推測是可重入設計中的資源狀態（UPD.確實是個含糊的概念，這是要交由實現類來具體使用的））ui
CLH隊列有獨佔(null)和共享（一個空的Node()）兩種模式，分別爲ReentranceLock和Semaphore/CyclicBarrier等線程通訊工具提供實現基類this
CLH locks are normally used forspinlocksatom
A node is signalled when its predecessor releases.
enqueue操做是經過CAS來實現雙向鏈表的，詳見line583:enq（好奇隊列爲空時設立head的操做，大概是一種lazy設計）

爲何unpark須要從後往前遍歷，須要看併發狀況下的原子性，當CAStail爲新的node時，原tail的next並不指向真正的tail，而prev保證了必然能遍歷到全部的node（再次給大佬跪了，懵逼了很久orz），UPD.還有一點是cancelAcquire時node.next=node，此時若是有unpark的衝突會死掉，而prev是正常工做的

private Node enq(final Node node) {
        for (;;) {
            Node t = tail;
            if (t == null) { // Must initialize
                if (compareAndSetHead(new Node()))
                    tail = head;
            } else {
                node.prev = t; 
                if (compareAndSetTail(t, node)) {
                    // 恰好發生意外 新的tail.prev確定有了，但舊的tail.next仍是null
                    t.next = node;
                    return t;
                }
            }
        }
    }


private void unparkSuccessor(Node node) {
    /*
     * If status is negative (i.e., possibly needing signal) try
     * to clear in anticipation of signalling.  It is OK if this
     * fails or if status is changed by waiting thread.
     */
    int ws = node.waitStatus;
    if (ws < 0)
        compareAndSetWaitStatus(node, ws, 0);

    /*
     * Thread to unpark is held in successor, which is normally
     * just the next node.  But if cancelled or apparently null,
     * traverse backwards from tail to find the actual
     * non-cancelled successor.
     */
    Node s = node.next;
    if (s == null || s.waitStatus > 0) {
        s = null;
        for (Node t = tail; t != null && t != node; t = t.prev)
            if (t.waitStatus <= 0)
                s = t;
    }
    if (s != null)
        LockSupport.unpark(s.thread);
}

tryAcquire/tryRelease/tryAcquireShared/tryReleaseShared是要複寫的方法，JUC衍生的一批線程通訊工具就是靠這個
cancelAcquire(node)的操做也是比較費解啊

1.取消node的線程

2.獲取第一個waiteStatuse<=0的前驅pred，且把中間全部cancelled的節點所有置空，node.prev=pred

3.把node也給cancelled掉

4.node已經廢了，若是node原本是tail那還要把tail給CAS成pred，predNext爲空

5.若是pred是head,那就喚醒後繼（就是倒着跑的那個unparkSuccessor）

6.除此之外把pred的ws改成SIGNAL（合適的時候就喚醒，由於此時還不是頭的後繼就不須要這麼快喚醒）

7.廢棄的node.next=node（注意此時也是一條長長的死循環鏈表，內部所有Cancelled），用於快速GC
暫時就這麼多了，還有Node.EXCLUSIVE的模式是怎麼用的我有空查查，好累哦

Atomic

原理就是volatile int value + CAS
AtomicStampedReference使用Pair來維護一個reference和stamp
對象也能CAS，也是特殊的受限類提供UNSAFE.compareAndSwapObject

Executor

查閱ThreadPoolExecutor
原子整型ctl提供兩個信息，一個是workerCount，另外一個是runState，因爲使用了位壓縮因此線程數最多隻能到達(2^29)-1，文檔中提到可能會換成AtomicLong，只是爲了快而使用普通整型大小
runState提供5種狀態，
- RUNNING : 接收且執行
- SHUTDOWN : 不接收但執行隊列中任務
- STOP : 不接收也不執行
- TIDYING : 全部任務中斷，隊列爲0
- TERMINATED : terminated()完成
當有mainLock時可訪問工做者線程池HashSet<Worker> works，其中Worker繼承自AQS
awaitTermination由Condition提供支持

CountDownLatch / CyclicBarrier / Condition / ...

CountDownLatch經過Sync繼承AQS實現tryAcquireShared來實現共享鎖
CyclicBarrier相對複雜，使用了ReentranceLock和Condition來組合，主要是用於signalAll

CopyOnWrite...

沒啥特別的，看了一下CopyOnWriteArrayList的實現，就是寫時上鎖且複製，final ReentrantLock lock = this.lock;

BlockingQueue

這裏查閱的是ArrayBlockingQueue
很是保守的類，使用了Reentrance和Condition(notEmpty|notFull)，任意操做幾乎都上鎖
Itrs類做爲Iterator的代理，內部的Node是WeakReference類型，提供弱一致性（這裏不是特別懂具體的操做。。）
put()時lock會lockInterruptibly()，且會輪詢while (count == items.length) notFull.await();
take()時同理，但後面是輪詢while (count == 0) notEmpty.await();(話說內部的數組是循環數組啊)

感覺一下take的內部調用dequeue()

E x = (E) items[takeIndex];
items[takeIndex] = null; // effective Java中推薦的作法
if (++takeIndex == items.length)
    takeIndex = 0;
count--; // 真正的個數
if (itrs != null)
    itrs.elementDequeued(); // itrs相關
notFull.signal();
return x;

ReentranceLock

state = 0 沒鎖， state != 0 上鎖 state > 1 進入可重入狀態當state == 0時會執行setExclusiveOwnerThread來釋放資源
是否公平交由Sync使用策略模式調度，比方說NonfairSync就是非公平的調度，此時若是調用lock.lock()其實就是sync.lock()，默認下是NonfairSync

看看關鍵的地方

static final class NonfairSync extends Sync {
    private static final long serialVersionUID = 7316153563782823691L;

    /**
         * Performs lock.  Try immediate barge, backing up to normal
         * acquire on failure.
         */
    final void lock() {
        if (compareAndSetState(0, 1))
            setExclusiveOwnerThread(Thread.currentThread());
        else
            acquire(1);
    }

    protected final boolean tryAcquire(int acquires) {
        return nonfairTryAcquire(acquires);
    }
}

/**
     * Sync object for fair locks
     */
static final class FairSync extends Sync {
    private static final long serialVersionUID = -3000897897090466540L;

    final void lock() {
        acquire(1);
    }

    /**
         * Fair version of tryAcquire.  Don't grant access unless
         * recursive call or no waiters or is first.
         */
    protected final boolean tryAcquire(int acquires) {
        final Thread current = Thread.currentThread();
        int c = getState();
        if (c == 0) {
            if (!hasQueuedPredecessors() &&
                compareAndSetState(0, acquires)) {
                setExclusiveOwnerThread(current);
                return true;
            }
        }
        else if (current == getExclusiveOwnerThread()) {
            int nextc = c + acquires;
            if (nextc < 0)
                throw new Error("Maximum lock count exceeded");
            setState(nextc);
            return true;
        }
        return false;
    }
}

lock():能夠看出公平調度是很是乖巧的交給AQS來得到資源，而非公平調度則是經過CAS來搶先得到，不行再給AQS

tryAcquire:非公平經過CAS得到，而公平須要判斷是否hasQueuedPredecessors()