Semaphore (控制并发线程数)【信号量】

主要方法：acquire()拿到许可证，release返回

应用场景:数据库连接，假设有几万个文件需要读取，我们可以启动几十个线程并发的读取。但是，读到内存后还需要存储到数据库，此时数据库连接只有10个，我们可以使用semaphore控制线程持有数据库连接，同时只有十个线程获取数据库连接。

使用demo

package com.w.juc;

import java.util.concurrent.Semaphore;

public class SemaphoreDemo implements Runnable{
    private static Semaphore semaphore = new Semaphore(3);

    @Override
    public void run() {
        System.out.println(Thread.currentThread()+"正在请求一张许可证");
        try {
            semaphore.acquire();
            System.out.println(Thread.currentThread()+"请求到一张");
            Thread.sleep(2000);
            semaphore.release();
            System.out.println(Thread.currentThread()+"释放了一张");
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }

    public static void main(String[] args) {
        new Thread(new SemaphoreDemo(),"一号").start();
        new Thread(new SemaphoreDemo(),"二号").start();
        new Thread(new SemaphoreDemo(),"三号").start();
        new Thread(new SemaphoreDemo(),"四号").start();
        new Thread(new SemaphoreDemo(),"五号").start();
        new Thread(new SemaphoreDemo(),"六号").start();
    }
}

源码分析

从构造函数开始

public Semaphore(int permits) {
    sync = new NonfairSync(permits);//直接new的是内部类，非公平类
}
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);//直接调用父类中的方法
        }
    }

父类Sync中

Sync(int permits) {
    setState(permits);//还是使用同步状态来代表许可证的数量
}

也就是构造函数做了什么？初始化许可证数量

接下来看，我们常用的俩方法，必用的。

acquire()，请求获取一张许可证。还有一个重载的方法acquire(int)，和它的区别在于，增加了参数不能小于0判断，可以一次获取多张大于0的许可证

public void acquire() throws InterruptedException {
    sync.acquireSharedInterruptibly(1);//直接调用Sync中的方法
}
 public final void acquireSharedInterruptibly(int arg)
            throws InterruptedException {
        if (Thread.interrupted())
            throw new InterruptedException();
        if (tryAcquireShared(arg) < 0)//这里感觉好熟悉，好多类都调用了这个方法，countDownLatch，reentrantLock
            doAcquireSharedInterruptibly(arg);//AQS的共享锁模式，CountDownLatch
    }

tryAcquireShared(arg) 是由子类NonfairSync重写的方法，但是子类直接调用了父类Sync的方法
protected int tryAcquireShared(int acquires) {
            return nonfairTryAcquireShared(acquires);
}    
父类的Sync
final int nonfairTryAcquireShared(int acquires) {
            for (;;) {
                int available = getState();
                int remaining = available - acquires;
                //当计算过后的剩余许可证数量大于0，且cas设置新值成功，那么返回大于0的值，可以获取到锁
                if (remaining < 0 ||
                    compareAndSetState(available, remaining))
                    return remaining;
            }
        }

下面是返回许可证的release() 方法

public void release() {
    sync.releaseShared(1);
}
AQS方法
 public final boolean releaseShared(int arg) {
        if (tryReleaseShared(arg)) {//子类Sync重写
            doReleaseShared();//这个也是AQS的共享锁释放方法，和CountDownLatch一样
            //至于这里唤醒全部线程，应该是和Semaphore功能相关。
            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");
                if (compareAndSetState(current, next))//cas设置成功，就返回true。失败自旋一直到成功
                    return true;
            }
        }

问题：

Semaphore是什么？是一个用于控制线程访问共享资源的工具类，信号量，它保存了一系列的许可（permits），每次调用acquire()都将消耗一个许可，每次调用release()都将归还一个许可

Semaphore具有哪些特性？通常用于限制同一时间对共享资源的访问次数上，也就是常说的限流。

Semaphore的许可次数是否可以动态增减？可以，调用release(int)

Semaphore (控制并发线程数)【信号量】

Java相关栏目本月热门文章