前言
大家好，我是小郭，通过前面的学习，知道了线程的利与弊，正确的使用多线程，会尽最大的可能去压榨我们系统的资源，从而提高效率，但是如果不合理使用线程，可能会造成副作用，给系统带来更大的压力，进一步的思考，如何才能防止多线程操作一个资源？
有什么办法防止多线程同时操作一个资源
- volatile
- 原子类
- Synchronized和Lock
- Semaphore
我们主要来学习下 Semaphore
Semaphore是什么？
Semaphore被翻译为计数信号量，通常使用进行并发线程数量的限制，保证多个线程能够合理的使用资源。用大白话理解就是理解为红路灯。
官方的翻译：
计数信号量。 从概念上讲，信号量维护一组许可证。 如有必要，每个acquire块都将阻塞直到获得许可为止，然后再获取它。 每个release添加一个许可证，从而有可能释放阻塞的获取者。 但是，没有使用实际的许可对象。 Semaphore只是保持可用数量的计数并采取相应措施。
信号灯模型
- 计数器，初始化计数器数量
- 等待队列，用于排队，未拿到许可证的线程放入队列，拿到许可证，则移除队列
主要方法和核心参数
核心参数
- int permits 允许并发线程数量
- boolean fair 是否公平锁
构造方法
//默认是非公平锁，只要传入并发线程数量 public Semaphore(int permits) { sync = new NonfairSync(permits); } public Semaphore(int permits, boolean fair) { sync = fair ? new FairSync(permits) : new NonfairSync(permits); }
构造方法中，主要做的就是初始化计数器，设置公平锁，第一个构造方法是默认的非公平锁，因为当等待队列中的线程有了空位，设置为非公平锁才能都来争抢提高效率，但是可能会造成某些线程饿死。 所以根据场景，我们进行公平锁的选择。
核心方法
从此信号量获取许可，先休眠，直到获得可用线程或者被中断
void acquire() throws InterruptedException
中断继续
void acquireUninterruptibly()
从此信号量获取设定线程数许可，先休眠，直到获得可用线程或者被中断
void acquire(int permits)
尝试获取许可，如果能够获取成功则立即返回true，否则，则返回false
boolean tryAcquire()
和上面一样，设置了等待最长时间
boolean tryAcquire(long timeout, TimeUnit unit)
释放许可
void release()
返回当前可用的许可证数
int availablePermits()
等待许可证数
int getQueueLength()
返回正在等待线程的合集
Collection<Thread> getQueuedThreads()
原理
实现的核心还是AQS的共享模式
Sync extends AbstractQueuedSynchronizer
acquire()
//以共享模式获取，如果中断则中止. public final void acquireSharedInterruptibly(int arg) throws InterruptedException { //1. 通过首先检查中断状态，中断返回异常 if (Thread.interrupted()) throw new InterruptedException(); // 2. 以共享模式获取，获取到了锁，接下去，执行，没有就排队 if (tryAcquireShared(arg) < 0) doAcquireSharedInterruptibly(arg); } //非公平模式共享锁获取 protected int tryAcquireShared(int acquires) { for (;;) { //判断当前节点在同步队列中是否有前驱节点的判断，获取不到返回-1 if (hasQueuedPredecessors()) return -1; //Semaphore用AQS的state变量的值代表可用许可数 int available = getState(); int remaining = available - acquires; //如果剩余许可数小于0或者CAS将当前可用许可数设置为剩余许可数成功，则返回成功许可数 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 { //自旋过程中的退出条件是是当前节点的前驱节点是头结点并且tryAcquireShared(arg) //返回值大于等于0即能成功获得同步状态 for (;;) { //获取前驱节点 final Node p = node.predecessor(); if (p == head) { //争夺锁 int r = tryAcquireShared(arg); if (r >= 0) { //先把 head 给占了，然后唤醒队列中其他的线程 setHeadAndPropagate(node, r); p.next = null; // help GC failed = false; return; } } if (shouldParkAfterFailedAcquire(p, node) && parkAndCheckInterrupt()) throw new InterruptedException(); } } finally { //失败的话，取消状态清除该节点 if (failed) cancelAcquire(node); } } //设置head的值，完成初始化工作 private Node enq(final Node node) { for (;;) { Node t = tail; if (t == null) { // Must initialize //利用CAS修改头部数据 if (compareAndSetHead(new Node())) tail = head; } else { node.prev = t; if (compareAndSetTail(t, node)) { t.next = node; return t; } } } }
实践
没有进行控制的代码
public static void main(String[] args) { ExecutorService executorService = Executors.newFixedThreadPool(5); IntStream.range(0,5).forEach(i -> executorService.submit(() ->{ try { System.out.println(Thread.currentThread().getName() + "gogogo"); Thread.sleep(1000); System.out.println(Thread.currentThread().getName() + "正在操作"); } catch (InterruptedException e) { e.printStackTrace(); } System.out.println(Thread.currentThread().getName() + "释放操作"); })); executorService.shutdown(); }
输出结果
pool-1-thread-1gogogo pool-1-thread-3gogogo pool-1-thread-4gogogo pool-1-thread-2gogogo pool-1-thread-5gogogo pool-1-thread-1正在操作 pool-1-thread-5正在操作 pool-1-thread-5释放操作 pool-1-thread-2正在操作 pool-1-thread-2释放操作 pool-1-thread-3正在操作 pool-1-thread-3释放操作 pool-1-thread-4正在操作 pool-1-thread-4释放操作 pool-1-thread-1释放操作
进行改造使用我们的并发工具Semaphore
输出结果
private static Semaphore semaphore = new Semaphore(1,false); public static void main(String[] args) { ExecutorService executorService = Executors.newFixedThreadPool(5); IntStream.range(0,5).forEach(i -> executorService.submit(() ->{ try { semaphore.acquire(); System.out.println(Thread.currentThread().getName() + "gogogo"); Thread.sleep(1000); System.out.println(Thread.currentThread().getName() + "正在操作"); } catch (InterruptedException e) { e.printStackTrace(); } semaphore.release(); System.out.println(Thread.currentThread().getName() + "释放操作"); })); executorService.shutdown(); }
输出结果
pool-1-thread-1gogogo pool-1-thread-1正在操作 pool-1-thread-1释放操作 pool-1-thread-2gogogo pool-1-thread-2正在操作 pool-1-thread-2释放操作 pool-1-thread-3gogogo pool-1-thread-3正在操作 pool-1-thread-3释放操作 pool-1-thread-4gogogo pool-1-thread-4正在操作 pool-1-thread-4释放操作 pool-1-thread-5gogogo pool-1-thread-5正在操作 pool-1-thread-5释放操作
我们可以很明显的看出，Semaphore进行了并发线程的控制，每次只一个线程，等待获取许可才能够往下执行。
在流量控制的场景，是我们的优先选择。
总结
- 初始化Semaphore的令牌数量
- 调用acquire()方法启动Semaphore，对进来的线程，进行令牌颁发，线程会阻塞在acquire()上
- 执行完回调函数后，执行release()，释放令牌
- 如果计数器的值小于等于0，就会自动唤醒等待的线程 Semaphore主要利用了队列和计数器，来完成对线程的控制，从而防止过多线程同时操作一个资源，在实战的项目中，我们也可以利用计数器的思路，进行各种池化资源的设计。
