同步工具类Semaphore

阳光穿透心脏的1/2处 2022-09-09 14:42 201阅读 0赞

Semaphore提供了资源数量的并发访问控制，也就是线程的并发量  
代码使用

//一开始有5份共享资源,即5个线程并发量。第二个参数表示是否是公平
    Semaphore semaphore = new Semaphore(5,true);
    
    // 工作线程每获取一份资源，就在该对象上记下来
    // 在获取的时候是按照公平的方式还是非公平的方式，要看上面代码的第二个参数是否公平。 
    // 一般非公平抢占效率较高。
    semaphore.acquire();
    // 工作线程每归还一份资源，就在该对象上记下来 
    // 此时资源可以被其他线程使用 
    semaphore.release();
    
    
    //释放指定数目的许可，并将它们归还给Semaphore。 
    //可用许可数加上该指定数目。 
    //如果线程需要获取N个许可，在有N个许可可用之前，该线程阻塞。
    semaphore.release(2);

构造方法

//FairSync和NonfairSync是两个继承Sync的内部类,Sync又是继承AbstractQueuedSynchronizer
    public Semaphore(int permits, boolean fair) {
        
        sync = fair ? new FairSync(permits) : new NonfairSync(permits);
    }

获得资源

public void acquire() throws InterruptedException {
        
    	//可中断的方式获得资源
        sync.acquireSharedInterruptibly(1);
    }
    
    public final void acquireSharedInterruptibly(int arg) throws InterruptedException {
        
    	//抛出被中断异常
        if (Thread.interrupted())
            throw new InterruptedException();
        //尝试获得资源
        if (tryAcquireShared(arg) < 0)
        	//获得资源失败，正经的do可中断的获得资源
            doAcquireSharedInterruptibly(arg);
    }
    
    //FairSync中的tryAcquireShared方法
    protected int tryAcquireShared(int acquires) {
        
        for (;;) {
        
        	//如果排队获取资源的Node链表的第一个Node不是当前线程返回-1,即当前线程不是第一个排队获取资源的线程
        	//第一个node是head.next
            if (hasQueuedPredecessors())
                return -1;
            //获得aqs的status，status是Semaphore的资源
            int available = getState();
            //检查剩余的资源是否足够
            int remaining = available - acquires;
            //cas获取资源失败自旋，资源不足或者cas成功return
            if (remaining < 0 || compareAndSetState(available, remaining))
                return remaining;
        }
    }
    
    private void doAcquireSharedInterruptibly(int arg) throws InterruptedException {
        
    	//把当前线程加入等待获取资源的队列
        final Node node = addWaiter(Node.SHARED);
        boolean failed = true;
        try {
        
            for (;;) {
        
            	//获得排在当前线程的前面一个线程
                final Node p = node.predecessor();
                //如果它前一个节点是head，说明当前线程排在第一位
                if (p == head) {
        
                	//尝试获取资源
                    int r = tryAcquireShared(arg);
                    //获取到了资源
                    if (r >= 0) {
        
                    	//把当前线程设置为head
                        setHeadAndPropagate(node, r);
                        p.next = null; // help GC
                        failed = false;
                        return;
                    }
                }
                //没获取到资源，挂起当前线程，等待唤醒，唤醒后回到for循环，把自己设置成head
                //设置前一个节点（p）的waitStatus为Node.SIGNAL
                if (shouldParkAfterFailedAcquire(p, node) &&
                    parkAndCheckInterrupt())
                    throw new InterruptedException();
            }
        } finally {
        
            if (failed)
                cancelAcquire(node);
        }
    }

释放资源

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 (;;) {
        
      		//获得资源数量（state）
            int current = getState();
            int next = current + releases;
            //溢出
            if (next < current) // overflow
                throw new Error("Maximum permit count exceeded");
            //cas设置资源数量
            if (compareAndSetState(current, next))
                return true;
        }
    }
    private void doReleaseShared() {
        
        for (;;) {
        
            Node h = head;
            if (h != null && h != tail) {
        
                int ws = h.waitStatus;
                //前面获得资源时shouldParkAfterFailedAcquire设置了阻塞线程的前一个节点为Node.SIGNAL
                if (ws == Node.SIGNAL) {
        
                	//设置head状态,失败自旋
                    if (!compareAndSetWaitStatus(h, Node.SIGNAL, 0))
                        continue;            // loop to recheck cases
                    //唤醒head的next线程,唤醒后doAcquireSharedInterruptibly方法会重新设置新的head,然后下面的判断h == head是否结束唤醒
                    unparkSuccessor(h);
                }
                //jdk1.6修复一个bug
                else if (ws == 0 &&
                         !compareAndSetWaitStatus(h, 0, Node.PROPAGATE))
                    continue;                // loop on failed CAS
            }
            //唤醒结束
            if (h == head)                   // loop if head changed
                break;
        }
    }