了解一下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()

ReentrantReadWriteLock实现ReadWriteLock接口，区别在于readLock的lock()实际是委托给sync的acquireShared(1)，而WriteLock是sync.acquire(1)