git clone https://github.com/chewiebug/GCViewer.git

mvn clean pacakge

cd target

java -jar -verbose:gc -XX:+PrintGCDetails -XX:+PrintGCTimeStamps -Xloggc:./gc.log xxx.jar

private void doAcquireSharedInterruptibly(int arg)
    throws InterruptedException {
    //将当前线程封装成SHARDED类型的节点，加入到链表中
    final Node node = addWaiter(Node.SHARED);
    boolean failed = true;
    try {
        for (;;) {
            //自旋来获取锁，获取前置节点，如果前置节点是头结点，尝试获取锁，将当前节点设置为头节点，开始传播为后置节点解锁
            final Node p = node.predecessor();
            if (p == head) {
                int r = tryAcquireShared(arg);
                if (r >= 0) {
                    /**自旋解锁：唤醒头结点的线程并且开始根据链表顺序传播解锁。
                    *   因为是公平锁，所以无论现在在哪个线程都会根据链表顺序从head节点开始唤醒
                    **/
                    setHeadAndPropagate(node, r);
                    p.next = null; // help GC
                    failed = false;
                    return;
                }
            }

            //通独占锁，根据前置节点的情况判断是否应该休眠（前置节点是SINGAL，当前节点的线程会block，如果前置节点是cancel会跳过所有cancel的节点）
            if (shouldParkAfterFailedAcquire(p, node) &&
                parkAndCheckInterrupt())
                throw new InterruptedException();
        }
    } finally {
        if (failed)
            cancelAcquire(node);
    }
}

private void setHeadAndPropagate(Node node, int propagate) {
    //当前的头节点当中间变量同时设置头结点
    Node h = head;
    setHead(node);
    /*
     * Try to signal next queued node if:
     *   Propagation was indicated by caller,
     *     or was recorded (as h.waitStatus either before
     *     or after setHead) by a previous operation
     *     (note: this uses sign-check of waitStatus because
     *      PROPAGATE status may transition to SIGNAL.)
     * and
     *   The next node is waiting in shared mode,
     *     or we don't know, because it appears null
     *
     * The conservatism in both of these checks may cause
     * unnecessary wake-ups, but only when there are multiple
     * racing acquires/releases, so most need signals now or soon
     * anyway.
     */
     //propagate>0（这里是state==0），或者头结点为空，或者头结点不是CANCELED状态，或者现在的头节点为空，或者头结点不是CANCELED状态
    if (propagate > 0 || h == null || h.waitStatus < 0 ||
        (h = head) == null || h.waitStatus < 0) {
        Node s = node.next;//把当前Node状态之后的节点Shared都唤醒
        if (s == null || s.isShared())
            doReleaseShared();
    }
}

private void doReleaseShared() {
    /*
     * Ensure that a release propagates, even if there are other
     * in-progress acquires/releases.  This proceeds in the usual
     * way of trying to unparkSuccessor of head if it needs
     * signal. But if it does not, status is set to PROPAGATE to
     * ensure that upon release, propagation continues.
     * Additionally, we must loop in case a new node is added
     * while we are doing this. Also, unlike other uses of
     * unparkSuccessor, we need to know if CAS to reset status
     * fails, if so rechecking.
     */
    for (;;) {
        Node h = head;
        if (h != null && h != tail) {
            int ws = h.waitStatus;
             //自旋解锁，如果头结点的状态是SIGNAL，因为刚才将当前节点置为了Head，实际就是当前节点，同时一般ws也是SINGAL
            if (ws == Node.SIGNAL) {
                //cas成功后将头结点的线程解锁
                if (!compareAndSetWaitStatus(h, Node.SIGNAL, 0))
                    continue;            // loop to recheck cases
                unparkSuccessor(h);
            }else if (ws == 0 &&
                     !compareAndSetWaitStatus(h, 0, Node.PROPAGATE))
                    //刚才的setHead()方法已经将Thread=null所以，如果修改成功了会跳出循环
                continue;                // loop on failed CAS
        }

        //上面修改成功了node是
        if (h == head)                   // loop if head changed
            break;
    }
}

protected boolean tryReleaseShared(int releases) {
    // Decrement count; signal when transition to zero
    for (;;) {
        int c = getState();
        if (c == 0)
            return false;
        int nextc = c-1;
        if (compareAndSetState(c, nextc))
            return nextc == 0;
    }
}

public final boolean releaseShared(int arg) {
    if (tryReleaseShared(arg)) {
        doReleaseShared();
        return true;
    }
    return false;
}

/**
独占锁的获取锁方法
调用tryAcquire获取锁，如果失败将当前的节点和state加入到队列中，开始竞争/排队获取锁。
*/
public final void acquire(int arg) {
    if (!tryAcquire(arg) &&
        acquireQueued(addWaiter(Node.EXCLUSIVE), arg))
        selfInterrupt();
}

//将当前线程封装成node(mode是独占模式)插入到队列末尾
private Node addWaiter(Node mode) {
    Node node = new Node(Thread.currentThread(), mode);
    // 快速插入将node插入到队列的末尾，只插入一次，失败后通过自旋重复插入直到成功
    Node pred = tail;
    if (pred != null) {
        node.prev = pred;
        //compareAndSetTail实现上是通过，修改tail字段的值
        if (compareAndSetTail(pred, node)) {
            pred.next = node;
            return node;
        }
    }
    //失败后用自旋开始插入
    enq(node);
    return node;
}

//自旋插入队列尾
private Node enq(final Node node) {
    for (;;) {
        Node t = tail;
        if (t == null) {
            //如果tail为空。初始化队列（tail=head=new Node）,然后下次循环将node插入到到tail后面，这时候 new Node()->node
            if (compareAndSetHead(new Node()))
                tail = head;
        } else {
            node.prev = t;
            if (compareAndSetTail(t, node)) {
                t.next = node;
                return t;
            }
        }
    }
}


final boolean acquireQueued(final Node node, int arg) {
    boolean failed = true;
    try {
        boolean interrupted = false;
        //自旋找到头结点，获取锁
        for (;;) {
            //获取前置节点
            final Node p = node.predecessor();
            //如果前置节点是头节点，tryAcquire获取锁
            if (p == head && tryAcquire(arg)) {
                //成功后将当前节点设置为头节点，interrupted=false
                setHead(node);
                p.next = null; // help GC
                failed = false;
                return interrupted;
            }

            //如果失败判断状态，并且挂起当前线程
            if (shouldParkAfterFailedAcquire(p, node) &&
                parkAndCheckInterrupt())
                interrupted = true;
        }
    } finally {
        if (failed)
            cancelAcquire(node);
    }
}

//判断是否该挂起线程
private static boolean shouldParkAfterFailedAcquire(Node pred, Node node) {
    //前置节点的类型
    int ws = pred.waitStatus;
    if (ws == Node.SIGNAL)
        //如果前置节点状态为SIGNAL，当前节点可以挂起
        return true;
    if (ws > 0) {//代码为什么不ws==Node.CANCELLED???
        //循环链表，跳过所有前置节点watiStatus是CANCELLED的节点，直到第一个不是CANCELLED的节点。
        do {
            node.prev = pred = pred.prev;
        } while (pred.waitStatus > 0);
        pred.next = node;
    } else {
        //等待状态是是Propagated状态，CAS改为SIGNAL
        compareAndSetWaitStatus(pred, ws, Node.SIGNAL);
    }
    return false;
}

//
private final boolean parkAndCheckInterrupt() {
    //具体实现见下面LockSupport.park详解
    LockSupport.park(this);
    return Thread.interrupted();
}

public final boolean release(int arg) {
    if (tryRelease(arg)) {
        Node h = head;
        if (h != null && h.waitStatus != 0)
            unparkSuccessor(h);
        return true;
    }
    return false;
}

//唤醒后置节点
private void unparkSuccessor(Node node) {
    //获取头节点状态，置为0防止重复唤醒
    int ws = node.waitStatus;
    if (ws < 0)
        compareAndSetWaitStatus(node, ws, 0);

    //找到后置节点，如果后置节点为空，或者状态是CANCELLED,从尾部开始遍历链表，找到离头节点最近的一个不是CANCELLED状态的节点
    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);
}

public static void park(Object blocker) {
    //获取当前线程，并且设置parkBlocker对象，jstack查看时候就可以知道线程是在哪个对象上阻塞，这里是AQS的实现类
    Thread t = Thread.currentThread();
    setBlocker(t, blocker);
    //阻塞线程，绝对时间，并且无限制的阻塞
    UNSAFE.park(false, 0L);
    //阻塞停止，将parkBlocker置为空。
    setBlocker(t, null);
}

static final class Node {
        /** Marker to indicate a node is waiting in shared mode */
        static final Node SHARED = new Node();
        /** Marker to indicate a node is waiting in exclusive mode */
        static final Node EXCLUSIVE = null;

        /** waitStatus value to indicate thread has cancelled */
        static final int CANCELLED =  1;
        /** waitStatus value to indicate successor's thread needs unparking */
        static final int SIGNAL    = -1;
        /** waitStatus value to indicate thread is waiting on condition */
        static final int CONDITION = -2;
        static final int PROPAGATE = -3;

        volatile int waitStatus;
        volatile Node prev;
        volatile Node next;
        volatile Thread thread;
        Node nextWaiter;

}

public void doWafDataArchiveByDate(LocalDateTime localDateTime) {
    String dateFormater = localDateTime.format(YYYYMMDD);

    //分页
    Long total = statsRedis.domainSize();
    Long pageNo = (total - 1L) / ZSET_PAGE_SIZE + 1;
    Long curPage = 1L;

    do {
        //从redis获取数据
        Long startIdx = (curPage - 1L) * ZSET_PAGE_SIZE;
        Long endIdx = startIdx + ZSET_PAGE_SIZE - 1L;
        Set<String> set = statsRedis.queryRange(startIdx, endIdx);

        curPage++;
        for (String domainUuid : set) {
            try {
                //锁1天
                if (taskLock.lock(domainUuid, dateFormater)) {
                    try {
                        //do logic
                    } catch (Exception e) {
                        e.printStackTrace();
                    }
                } else {
                }
            } finally {
                taskLock.unlock(domainUuid,dateFormater);
            }
        }
    } while (curPage <= pageNo);
}