Semaphore，动态增减信号量

【信号量】：

• 用于控制对某资源访问的同一时间的并发量。

 

【如何获取】：

• semaphore.tryAcquire()，尝试获取，不阻塞
• semaphore.acquire()，没信号量可用时，将进行阻塞等

 

【如何释放】：

• semaphore.release();
• 线程抛出各种异常，都别忘了在finally中释放信号量；
• 如果释放的比获取的信号量还多，例如获取了2个，释放了5次，那么当前信号量就动态的增加为5了，要注意。

 

【动态增加】：

• 多释放几次，就可以达到信号量动态增加的效果了

 

【动态减小】：

• 信号量本来有这个api的，不过是protected方法，所以我们需要显式继续Semaphore，并重新实现该api，见ResizeableSemaphore类中的reducePermits(int reduction);
• 举例如下：（该表格有个假设前提，不存在多余的release而产生出新的信号量，即release次数<=acquire次数）

当前信号量A (A=A1+A2)	占用信号量A1	可用信号量A2	重新设置信号量B (B=B1+B2)	当前可用的信号量B1	当前待释放的量B2
5	3	2	3	0	0
5	3	2	1	0	-2
5	3	2	9	6	0

 

package com.jd.las.basic.service;

import java.util.concurrent.Semaphore;

/**
 * 
 * Title: 动态信号量<br>
 * 
 * Description: <br>
 * 
 * Company: <a href=www.jd.com>京东</a><br>
 * 
 * @author <a href=mailto:[e-mail]>龙准</a>
 * 
 * @date 2015年5月26日 下午7:29:42
 */
public class JdSemaphore {

	 /**
     * semaphore starts at 0 capacity; must be set by setMaxPermits before use
     */
    private final ResizeableSemaphore semaphore = new ResizeableSemaphore();
 
    /**
     * how many permits are allowed as governed by this semaphore.
     * Access must be synchronized on this object.
     */
    private int maxPermits = 0;
 
    /**
     * New instances should be configured with setMaxPermits().
     */
    public JdSemaphore() {
        // no op
    }
 
    /*
     * Must be synchronized because the underlying int is not thread safe
     */
    /**
     * Set the max number of permits. Must be greater than zero.
     *
     * Note that if there are more than the new max number of permits currently
     * outstanding, any currently blocking threads or any new threads that start
     * to block after the call will wait until enough permits have been released to
     * have the number of outstanding permits fall below the new maximum. In
     * other words, it does what you probably think it should.
     *
     * @param newMax
     */
    public synchronized void setMaxPermits(int newMax) {
        if (newMax < 1) {
            throw new IllegalArgumentException("Semaphore size must be at least 1,"
                + " was " + newMax);
        }
 
        int delta = newMax - this.maxPermits;
 
        if (delta == 0) {
            return;
        } else if (delta > 0) {
            // new max is higher, so release that many permits
            this.semaphore.release(delta);
        } else {
            delta *= -1;
            // delta < 0.
            // reducePermits needs a positive #, though.
            this.semaphore.reducePermits(delta);
        }
 
        this.maxPermits = newMax;
    }
 
    /**
     * Release a permit back to the semaphore. Make sure not to double-release.
     *
     */
    public void release() {
        this.semaphore.release();
    }
 
    /**
     * Get a permit, blocking if necessary.
     *
     * @throws InterruptedException
     *             if interrupted while waiting for a permit
     */
    public void acquire() throws InterruptedException {
        this.semaphore.acquire();
    }
 
    /**
     * A trivial subclass of <code>Semaphore</code> that exposes the reducePermits
     * call to the parent class. Doug Lea says it's ok...
     * http://osdir.com/ml/java.jsr.166-concurrency/2003-10/msg00042.html
     */
    private static final class ResizeableSemaphore extends Semaphore {
        /**
         *
         */
        private static final long serialVersionUID = 1L;
 
        /**
         * Create a new semaphore with 0 permits.
         */
        ResizeableSemaphore() {
            super(0);
        }
 
        @Override
        protected void reducePermits(int reduction) {
            super.reducePermits(reduction);
        }
    }

}

 

这里可参考：http://blog.teamlazerbeez.com/2009/04/20/javas-semaphore-resizing/