欢迎您访问程序员文章站本站旨在为大家提供分享程序员计算机编程知识!
您现在的位置是: 首页

jave信号量源码解读(Semaphore)

程序员文章站 2022-06-14 10:25:08
...

一、什么是信号量?

一个通俗的例子,今天部门经理带我们30人去西部世界,总共只有5位接待员,一次只能服务5位,剩下的人只能等待空闲的接待员(前面的人接待完毕)。下面是某易的例子。

public class SemaphoreDemo {
    public static void main(String[] args) {
        SemaphoreDemo semaphoreDemo=new SemaphoreDemo();
        //客人数量
        int n=8; 
        //接待员数量
       Semaphore semaphore =new Semaphore(5);
        for(int i=0;i<n;i++){
            String vipNo="vip-00" + i;
            new Thread(() -> {
                try{
                   //获取令牌
                    semaphore.acquire(); 
                    //模拟被接待
                    semaphoreDemo.service(vipNo);
                    //模拟释放
                    semaphore.release();
                }catch (InterruptedException e){
                    e.printStackTrace();
                }
            }).start();
        }
    }
    public  void service(String vipNo) throws InterruptedException {
        System.out.println("vip消费开始  "+vipNo+"");
        Thread.sleep(new Random().nextInt(3000));
        System.out.println("vip 消费结束 "+vipNo+"");
    }
}

运行如图:
jave信号量源码解读(Semaphore)

二、上代码(核心代码不足100行)


public class Semaphore implements java.io.Serializable {
    private final Sync sync;
    //继承抽象队列同步器
    abstract static class Sync extends AbstractQueuedSynchronizer {
        private static final long serialVersionUID = 1192457210091910933L;
      //令牌个数
        Sync(int permits) {
            setState(permits);
        }

        final int getPermits() {
            return getState();
        }
      //非公平获取共享资源
        final int nonfairTryAcquireShared(int acquires) {
        //这里使用循环+cas方式获取
            for (;;) {
                int available = getState();
                int remaining = available - acquires;
                //<0或者cas成功 返回
                if (remaining < 0 ||
                    compareAndSetState(available, remaining))
                    return remaining;
            }
        }
   //尝试释放 依旧cas+循环 cas成功 返回
        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;
            }
        }
//依旧cas+循环
        final void reducePermits(int reductions) {
            for (;;) {
                int current = getState();
                int next = current - reductions;
                if (next > current) // underflow
                    throw new Error("Permit count underflow");
                if (compareAndSetState(current, next))
                    return;
            }
        }
//依旧cas+循环
        final int drainPermits() {
            for (;;) {
                int current = getState();
                if (current == 0 || compareAndSetState(current, 0))
                    return current;
            }
        }
    }

   
    static final class NonfairSync extends Sync {
        private static final long serialVersionUID = -2694183684443567898L;

        NonfairSync(int permits) {
            super(permits);
        }

        protected int tryAcquireShared(int acquires) {
            return nonfairTryAcquireShared(acquires);
        }
    }

    /**
     * Fair version
     */
    static final class FairSync extends Sync {
        private static final long serialVersionUID = 2014338818796000944L;

        FairSync(int permits) {
            super(permits);
        }
   
        protected int tryAcquireShared(int acquires) {
            for (;;) {
            //是cas队列的头结点或者队列为空 令牌-1
                if (hasQueuedPredecessors())
                    return -1;
                int available = getState();
                int remaining = available - acquires;
                //令牌不足 或cas成功
                if (remaining < 0 ||
                    compareAndSetState(available, remaining))
                    return remaining;
            }
        }
    }
// 非公平实现
    public Semaphore(int permits) {
        sync = new NonfairSync(permits);
    }


    public Semaphore(int permits, boolean fair) {
        sync = fair ? new FairSync(permits) : new NonfairSync(permits);
    }
//会调用上面实现AQS的tryAcquireShared方法来获取
    public void acquire() throws InterruptedException {
        sync.acquireSharedInterruptibly(1);
    }

    public void acquireUninterruptibly() {
        sync.acquireShared(1);
    }


    public boolean tryAcquire() {
        return sync.nonfairTryAcquireShared(1) >= 0;
    }

    public boolean tryAcquire(long timeout, TimeUnit unit)
        throws InterruptedException {
        return sync.tryAcquireSharedNanos(1, unit.toNanos(timeout));
    }

 //调用上面的tryReleaseShared方法
    public void release() {
        sync.releaseShared(1);
    }

    public void acquire(int permits) throws InterruptedException {
        if (permits < 0) throw new IllegalArgumentException();
        sync.acquireSharedInterruptibly(permits);
    }


    public void acquireUninterruptibly(int permits) {
        if (permits < 0) throw new IllegalArgumentException();
        sync.acquireShared(permits);
    }


    public boolean tryAcquire(int permits) {
        if (permits < 0) throw new IllegalArgumentException();
        return sync.nonfairTryAcquireShared(permits) >= 0;
    }

    public boolean tryAcquire(int permits, long timeout, TimeUnit unit)
        throws InterruptedException {
        if (permits < 0) throw new IllegalArgumentException();
        return sync.tryAcquireSharedNanos(permits, unit.toNanos(timeout));
    }

  
    public void release(int permits) {
        if (permits < 0) throw new IllegalArgumentException();
        sync.releaseShared(permits);
    }


    public int availablePermits() {
        return sync.getPermits();
    }

    public int drainPermits() {
        return sync.drainPermits();
    }

    protected void reducePermits(int reduction) {
        if (reduction < 0) throw new IllegalArgumentException();
        sync.reducePermits(reduction);
    }

    public boolean isFair() {
        return sync instanceof FairSync;
    }

}

三、总结

信号量对于资源的获取和释放,除去部分涉及AQS,其实本质依旧是使用了Cas+循环的操作,多了资源的判定个数操作。

相关标签: 源码