Semaphore源码解读

旧城等待， 2021-06-22 15:39 484阅读 0赞

#### Semaphore ####

Semaphore，俗称信号量，它是操作系统中PV操作的原语在java的实现，它也是基于AbstractQueuedSynchronizer的。

#### Semaphore的强大功能 ####

Semaphore的功能非常强大，大小为1的信号量就类似于互斥锁，通过同时只能有一个线程获取信号量实现。大小为n（n>0）的信号量可以实现限流的功能，它可以实现只能有n个线程同时获取信号量。  
光是以上的两个特性就能解决很多现实问题。

#### 代码示例 ####

import org.junit.Test;
    import java.util.concurrent.LinkedBlockingDeque;
    import java.util.concurrent.Semaphore;
    import java.util.concurrent.ThreadPoolExecutor;
    import java.util.concurrent.TimeUnit;
    
    /** * <pre>类名: SemaphoneTest</pre> * <pre>描述: 信号量测试</pre> * <pre>版权: web_chen@163.com</pre> * <pre>日期: 2020/11/6 8:55</pre> * <pre>作者: chenwb</pre> */
    public class SemaphoneTest { 
    
    	/** * 实现一个同时只能处理5个请求的限流器 */
    	private Semaphore semaphore = new Semaphore(5);
    
    	/** * 定义一个线程池 */
    	private ThreadPoolExecutor executor = new ThreadPoolExecutor(20, 50, 60, TimeUnit.SECONDS, new LinkedBlockingDeque<>(2000));
    
    	/** * 模拟执行方法 */
    	public void exec() { 
    		try { 
    			semaphore.acquire(1);
    			// 模拟真实方法执行
    			System.out.println("执行exec方法,时间：" + System.currentTimeMillis());
    			Thread.sleep(1000);
    		} catch (Exception e) { 
    			e.printStackTrace();
    		} finally { 
    			semaphore.release(1);
    		}
    	}
    
    	@Test
    	public void test() throws InterruptedException { 
    		for (; ; ) { 
    			Thread.sleep(100);
    			// 模拟请求以10个/s的速度
    			executor.execute(() -> exec());
    		}
    	}
    }

上面示例代码模拟了一个限流器，它限制了同时只能处理5个请求。

#### 源码分析 ####

我们主要看一下semaphore的acquire与release方法，先看acquire：

public void acquire(int permits) throws InterruptedException { 
        	// 信号量非负
            if (permits < 0) throw new IllegalArgumentException();
            sync.acquireSharedInterruptibly(permits);
        }

接着就调用AbstractQueuedSynchronizer的acquireSharedInterruptibly，获取共享模式下的状态量state

public final void acquireSharedInterruptibly(int arg)
                throws InterruptedException { 
    		...省略
    		// 获取信号量
            if (tryAcquireShared(arg) < 0)
            	// 如果信号量小于0，没有获取到，则需要阻塞了
                doAcquireSharedInterruptibly(arg);
        }

###### 如何获取信号量tryAcquireShared(arg)，它是AbstractQueuedSynchronizer的方法，在Semaphore中重写，有FairSync与NonfairSync两种实现： ######

先看下NonfairSync是如何获取信号量：

final int nonfairTryAcquireShared(int acquires) { 
                for (;;) { 
                    int available = getState();
                    int remaining = available - acquires;
                    if (remaining < 0 ||
                        compareAndSetState(available, remaining))
                        return remaining;
                }
            }

自旋，如果remaining <0,直接返回remaining ，否则尝试将remaining 替换原来的信号量available，返回remaining ，比较简单。  
再看FairSync实现：

protected int tryAcquireShared(int acquires) { 
                for (;;) { 
                	// 仅多了一个判断，如果前面已经有排队的线程，则直接获取不到
                    if (hasQueuedPredecessors())
                        return -1;
                    int available = getState();
                    int remaining = available - acquires;
                    if (remaining < 0 ||
                        compareAndSetState(available, remaining))
                        return remaining;
                }
            }

###### 获取不到信号量，则排队doAcquireSharedInterruptibly ######

private void doAcquireSharedInterruptibly(int arg) throws InterruptedException { 
           	// 加入排队节点
            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就可以出队了
                            setHeadAndPropagate(node, r);
                            p.next = null; // help GC
                            failed = false;
                            return;
                        }
                    }
                    // 如果不是队头的线程就要考虑挂起线程了
                    if (shouldParkAfterFailedAcquire(p, node) &&
                        parkAndCheckInterrupt())
                        throw new InterruptedException();
                }
            } finally { 
                if (failed)
                    cancelAcquire(node);
            }
        }

**以上的获取信息量可以归纳为：**  
1、先cas获取信号量，获取到，结束。  
2、如果信号量<0,加入阻塞队列。如果排在队头，再尝试获取信息量，获取成功，出队，结束。  
3、考虑挂起线程，等待其他线程释放信息量。

再下看release方法：

public void release(int permits) { 
            if (permits < 0) throw new IllegalArgumentException();
            sync.releaseShared(permits);
        }
        
        public final boolean releaseShared(int arg) { 
            if (tryReleaseShared(arg)) { 
                doReleaseShared();
                return true;
            }
            return false;
        }

tryReleaseShared在Semaphore中重写：

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");
                    if (compareAndSetState(current, next))
                        return true;
                }
            }

信号量state增加之后，决定唤醒哪个阻塞线程：

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;
            }
        }

**以上的释放信息量可以归纳为：**  
1、释放信号量  
2、唤醒一个阻塞线程