标签：prot 释放 failed cut log pool 学习 null 线程状态
基于AQS的前世今生，来学习并发工具类Semaphore。本文将从Semaphore的应用场景、源码原理解析来学习这个并发工具类。
1、 应用场景
Semaphore用来控制同时访问某个特定资源的操作数量，或者同时执行某个指定操作的数量。还可以用来实现某种资源池限制，或者对容器施加边界。
1.1 当成锁使用
控制同时访问某个特定资源的操作数量，代码如下：
1 public class SemaphoreLock { 2 public static void main(String[] args) { 3 //1、信号量为1时 相当于普通的锁 信号量大于1时 共享锁 4 Output o = new Output(); 5 for (int i = 0; i < 5; i++) { 6 new Thread(() -> o.output()).start(); 7 } 8 } 9 } 10 class Output { 11 Semaphore semaphore = new Semaphore(1); 12 13 public void output() { 14 try { 15 semaphore.acquire(); 16 System.out.println(Thread.currentThread().getName() + " start at " + System.currentTimeMillis()); 17 Thread.sleep(1000); 18 System.out.println(Thread.currentThread().getName() + " stop at " + System.currentTimeMillis()); 19 }catch(Exception e) { 20 e.printStackTrace(); 21 }finally { 22 semaphore.release(); 23 } 24 } 25 }
1.2 线程通信信号
线程间通信，代码如下：
1 public class SemaphoreCommunication { 2 public static void main(String[] args) { 3 //2、线程间进行通信 4 Semaphore semaphore = new Semaphore(1); 5 new SendingThread(semaphore,"SendingThread"); 6 new ReceivingThread(semaphore,"ReceivingThread"); 7 } 8 } 9 class SendingThread extends Thread { 10 Semaphore semaphore; 11 String name; 12 13 public SendingThread(Semaphore semaphore,String name) { 14 this.semaphore = semaphore; 15 this.name = name; 16 new Thread(this).start(); 17 } 18 19 public void run() { 20 try { 21 semaphore.acquire(); 22 for (int i = 0; i < 5; i++) { 23 System.out.println(name + ":" + i); 24 Thread.sleep(1000); 25 } 26 } catch (Exception e) { 27 e.printStackTrace(); 28 } 29 semaphore.release(); 30 } 31 } 32 33 class ReceivingThread extends Thread { 34 Semaphore semaphore; 35 String name; 36 37 public ReceivingThread(Semaphore semaphore,String name) { 38 this.semaphore = semaphore; 39 this.name = name; 40 new Thread(this).start(); 41 } 42 43 public void run() { 44 try { 45 semaphore.acquire(); 46 for (int i = 0; i < 5; i++) { 47 System.out.println(name + ":" + i); 48 Thread.sleep(1000); 49 } 50 } catch (Exception e) { 51 e.printStackTrace(); 52 } 53 semaphore.release(); 54 } 55 }
1.3 资源池限制
对资源池进行资源限制，代码如下：
1 public class SemaphoreConnect { 2 public static void main(String[] args) throws Exception { 3 //3、模拟连接池数量限制 4 ExecutorService executorService = Executors.newCachedThreadPool(); 5 for (int i = 0; i < 200; i++) { 6 executorService.submit(new Runnable() { 7 @Override 8 public void run() { 9 Connection.getInstance().connect(); 10 } 11 }); 12 } 13 executorService.shutdown(); 14 executorService.awaitTermination(1, TimeUnit.DAYS); 15 } 16 } 17 class Connection { 18 private static Connection instance = new Connection(); 19 private Semaphore semaphores = new Semaphore(10,true); 20 private int connections = 0; 21 22 private Connection() { 23 } 24 25 public static Connection getInstance() { 26 return instance; 27 } 28 29 public void connect() { 30 try { 31 semaphores.acquire(); 32 doConnect(); 33 } catch (InterruptedException e) { 34 e.printStackTrace(); 35 }finally { 36 semaphores.release(); 37 } 38 } 39 40 private void doConnect() { 41 synchronized (this) { 42 connections ++; 43 System.out.println("current get connections is : " + connections); 44 } 45 46 try { 47 Thread.sleep(2000); 48 } catch (InterruptedException e) { 49 e.printStackTrace(); 50 } 51 52 synchronized (this) { 53 connections --; 54 System.out.println("after release current connections is : " + connections); 55 } 56 } 57 }
1.4 容器边界限制
对容器进行边界限制，代码如下：
1 public class SemaphoreBoundedList { 2 public static void main(String[] args) { 3 //4、容器边界限制 4 final BoundedList ba = new BoundedList(5); 5 Runnable runnable1 = new Runnable() { 6 public void run() { 7 try { 8 ba.add("John"); 9 ba.add("Martin"); 10 ba.add("Adam"); 11 ba.add("Prince"); 12 ba.add("Tod"); 13 System.out.println("Available Permits : " + ba.getSemaphore().availablePermits()); 14 ba.add("Tony"); 15 System.out.println("Final list: " + ba.getArrayList()); 16 }catch (InterruptedException ie) { 17 Thread.interrupted(); 18 } 19 } 20 }; 21 Runnable runnable2 = new Runnable() { 22 public void run() { 23 try { 24 System.out.println("Before removing elements: "+ ba.getArrayList()); 25 Thread.sleep(5000); 26 ba.remove("Martin"); 27 ba.remove("Adam"); 28 }catch (InterruptedException ie) { 29 Thread.interrupted(); 30 } 31 } 32 }; 33 Thread thread1 = new Thread(runnable1); 34 Thread thread2 = new Thread(runnable2); 35 thread1.start(); 36 thread2.start(); 37 } 38 } 39 class BoundedList<T> { 40 private final Semaphore semaphore; 41 private List arrayList; 42 43 BoundedList(int limit) { 44 this.arrayList = Collections.synchronizedList(new ArrayList()); 45 this.semaphore = new Semaphore(limit); 46 } 47 48 49 public boolean add(T t) throws InterruptedException { 50 boolean added = false; 51 semaphore.acquire(); 52 try { 53 added = arrayList.add(t); 54 return added; 55 } finally { 56 if (!added) 57 semaphore.release(); 58 } 59 60 } 61 62 63 public boolean remove(T t) { 64 boolean wasRemoved = arrayList.remove(t); 65 if (wasRemoved) 66 semaphore.release(); 67 return wasRemoved; 68 } 69 70 public void remove(int index) { 71 arrayList.remove(index); 72 semaphore.release(); 73 } 74 75 public List getArrayList() { 76 return arrayList; 77 } 78 79 80 public Semaphore getSemaphore() { 81 return semaphore; 82 } 83 }
2、 源码原理解析
2.1 获取信号
获取信号的方法如下:
public void acquire() throws InterruptedException { sync.acquireSharedInterruptibly(1);//共享式获取AQS的同步状态 }
调用的是AQS的acquireSharedInterruptibly方法：
public final void acquireSharedInterruptibly(int arg) throws InterruptedException { if (Thread.interrupted())//线程中断 说明信号量对线程中断敏感 throw new InterruptedException(); if (tryAcquireShared(arg) < 0) //获取信号量失败 线程进入同步队列自旋等待 doAcquireSharedInterruptibly(arg); }
其中tryAcquireShared依赖的是Sync的实现，Sync提供了公平和非公平式的方式，先看非公平式。
protected int tryAcquireShared(int acquires) { return nonfairTryAcquireShared(acquires); } final int nonfairTryAcquireShared(int acquires) { for (;;) { int available = getState();//同步状态 当前的信号量许可数 int remaining = available - acquires;//减去释放的信号量 剩余信号量许可数 if (remaining < 0 ||//剩余信号量小于0 直接返回remaining 不做CAS compareAndSetState(available, remaining))//CAS更新 return remaining; } }
再看下公平式的。
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。
1 private void doAcquireSharedInterruptibly(int arg) 2 throws InterruptedException { 3 final Node node = addWaiter(Node.SHARED);//线程进入同步队列 4 boolean failed = true; 5 try { 6 for (;;) {//自旋 7 final Node p = node.predecessor(); 8 if (p == head) {//当前节点的前置节点是AQS的头节点 即自己是AQS同步队列的第一个节点 9 int r = tryAcquireShared(arg); //再去获取信号量 10 if (r >= 0) {//获取成功 11 setHeadAndPropagate(node, r);//退出自旋 12 p.next = null; // help GC 13 failed = false; 14 return; 15 } 16 } 17 if (shouldParkAfterFailedAcquire(p, node) && 18 parkAndCheckInterrupt()) 19 throw new InterruptedException(); 20 } 21 } finally { 22 if (failed) 23 cancelAcquire(node); //获取失败 就取消获取 24 } 25 }
2.2 释放信号
释放信号的方法如下：
public void release() { sync.releaseShared(1); }
调用的是AQS的releaseShared方法：
public final boolean releaseShared(int arg) { if (tryReleaseShared(arg)) {//释放信号量 doReleaseShared();//唤醒后续的线程节点 return true; } return false; }
tryReleaseShared交由子类Sync实现，代码如下：
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更新当前信号量许可数 return true; } }
释放成功后，则继续调用doReleaseShared，唤醒后续线程节点可以来争取信号量了。
1 private void doReleaseShared() { 2 for (;;) { 3 Node h = head; //头节点 4 if (h != null && h != tail) {//同步队列中存在线程等待 5 int ws = h.waitStatus; //头节点线程状态 6 if (ws == Node.SIGNAL) {//头节点线程状态为SIGNAL 唤醒后续线程节点 7 if (!compareAndSetWaitStatus(h, Node.SIGNAL, 0)) 8 continue; // loop to recheck cases 9 unparkSuccessor(h); //唤醒下个节点 10 } 11 else if (ws == 0 && 12 !compareAndSetWaitStatus(h, 0, Node.PROPAGATE)) 13 continue; // loop on failed CAS 14 } 15 if (h == head) // loop if head changed 16 break; 17 } 18 }
总结：Semaphore使用AQS同步状态来保存信号量的当前计数。它里面定义的acquireSharedInterruptibly方法会减少计数，当计数为非正值时阻塞线程，releaseShared方法会增加计数，在计数不超过信号量限制时要解除线程的阻塞。
参考资料:
https://github.com/lingjiango/ConcurrentProgramPractice
https://www.caveofprogramming.com/java-multithreading/java-multithreading-semaphores-part-12.html
https://java2blog.com/java-semaphore-example/
http://tutorials.jenkov.com/java-util-concurrent/semaphore.html
标签：prot 释放 failed cut log pool 学习 null 线程状态
原文地址：https://www.cnblogs.com/lgjava/p/11268283.html
