起源于定时任务优化的限流思考

Java
Class TimerTask {
    public void execute() {
        items.forEach(item -> {
            try {
                semaphore.acquire();
                executor.run(item, semaphore);
            } catch (InterruptedException e) {
                log.error(e.toString());
            }
        });
    }
}
Class Executor { 
@Async
    public void run(Item item, Semaphore semaphore) {
        try {   
 callRemoteAPI(item);
        } catch (Exception e) {
            log.error(e);
        }
        semaphore.release();
    }
}

Java
public void acquire() throws InterruptedException {
    // 公平锁获取锁的时候，获取锁的个数是1
    sync.acquireSharedInterruptibly(1);
}
// 获取锁
public final void acquireSharedInterruptibly(int arg) throws InterruptedException {
    if (Thread.interrupted())
        throw new InterruptedException();
    // 如果剩余锁数量小于0则锁获取失败，需要等待。否则就直接返回，说明获取锁成功。
    if (tryAcquireShared(arg) < 0)
        doAcquireSharedInterruptibly(arg);
}
// 尝试获取共享锁
protected int tryAcquireShared(int acquires) {
    for (;;) {
        if (hasQueuedPredecessors())
            return -1;
        // 获取AQS中State的数值
        int available = getState();
        // 将剩余锁的数量acquires（公平锁则-1）
        int remaining = available - acquires;
        // 如果剩余锁数量小于0，则将available置为原有值并返回剩余数量
        // 如果剩余锁数量大于0，则将available设为remaining。如果设置失败，则说明有其他线程已经在内存中修改过值了，需要重新循环获取锁
        if (remaining < 0 ||
            compareAndSetState(available, remaining))
            return remaining;
    }
}
// 如果锁暂时获取失败，则进入等待队列
private void doAcquireSharedInterruptibly(int arg)
    throws InterruptedException {
    // 这里等待的实现用的是AQS的队列，将元素添加到队列尾部
    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) {
                    // node获取锁后出队，设置新head，并将共享性传播给后继节点
                    setHeadAndPropagate(node, r);
                    p.next = null; // help GC
                    failed = false;
                    return;
                }
            }
            // 对节点进行检查和更新状态，如果线程应该阻塞，返回 true。
            // 线程获取锁失败后入队列并不会立刻阻塞，而是判断是否应该阻塞。只有当头结点的waitStatus是SIGNAL时shouldParkAfterFailedAcquire会返回true继而阻塞
            if (shouldParkAfterFailedAcquire(p, node) &&
                parkAndCheckInterrupt())
                throw new InterruptedException();
        }
    } finally {
        if (failed)
            cancelAcquire(node);
    }
}

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


// 释放共享锁入口
public final boolean releaseShared(int arg) {
    if (tryReleaseShared(arg)) {
        doReleaseShared();
        return true;
    }
    return false;
}


// 尝试释放共享锁
protected final boolean tryReleaseShared(int releases) {
    for (;;) {
        int current = getState();
        int next = current + releases;
        if (next < current) // overflow
            throw new Error("Maximum permit count exceeded");
        // 将current + 1，如果失败说明内存中值已经变化了，重新循环
        if (compareAndSetState(current, next))
            return true;
    }
}
// 释放共享锁逻辑
private void doReleaseShared() {
    for (;;) {
        Node h = head;
        if (h != null && h != tail) {
            int ws = h.waitStatus;
            if (ws == Node.SIGNAL) {
                if (!compareAndSetWaitStatus(h, Node.SIGNAL, 0))
                    continue;            // loop to recheck cases
                unparkSuccessor(h);
            }
            else if (ws == 0 &&
                     !compareAndSetWaitStatus(h, 0, Node.PROPAGATE))
                continue;                // loop on failed CAS
        }
        if (h == head)                   // loop if head changed
            break;
    }
}