Java并发编程-Semaphore

基于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

【推广】 免费学中医，健康全家人
原文地址：https://www.cnblogs.com/lgjava/p/11268283.html