java并发编程之一--Semaphore的使用

1、介绍

　　Semaphore 中文的含义信号，信号系统，此类的只要作用就是限制线程的并发的数量。

　　Semaphore内部主要通过AQS（AbstractQueuedSynchronizer）实现线程的管理。Semaphore有两个构造函数，参数permits表示许可数，它最后传递给了AQS的state值。线程在运行时首先acquire（获取）许可，如果成功，许可数就减1，线程运行，当线程运行结束就release（释放）许可，许可数就加1。如果许可数为0，则获取失败，线程位于AQS的等待队列中，它会被其它释放许可的线程唤醒。在创建Semaphore对象的时候还可以指定它的（fair）公平性。一般常用非公平的信号量，非公平信号量是指在获取许可时先尝试获取许可，而不必关心是否已有需要获取许可的线程位于等待队列中，如果获取失败，才会入列。而公平的信号量在获取许可时首先要查看等待队列中是否已有线程，如果有则入列。

2、方法

构造函数

1 //参数permits代表同一时间内，最多允许多少个线程执行
2 public Semaphore(int permits) {  
3     sync = new NonfairSync(permits);  
4 }  
5 //参数fair表示是否公平信号量
6 public Semaphore(int permits, boolean fair) {  
7     sync = (fair)? new FairSync(permits) : new NonfairSync(permits);  
8 }

acquire()、release()方法

1 public void acquire() throws InterruptedException {  }     //获取一个许可  
2 public void acquire(int permits) throws InterruptedException { }    //获取permits个许可  
3 public void release() { }          //释放一个许可  
4 public void release(int permits) { }    //释放permits个许可

其他的方法

 1 //返回可用的许可，并将可用许可数清零
 2 public int drainPermits() 
 3 //尝试获取permits个许可，若获取成功，则立即返回true，若获取失败，则立即返回false  
 4 public boolean tryAcquire(int permits) { };   
 5 //尝试获取permits个许可，若在指定的时间内获取成功，则立即返回true  
 6 public boolean tryAcquire(int permits, long timeout, TimeUnit unit) throws InterruptedException { };
 7 //在等待进入acquire()方法的线程 不允许被中断
 8 public void acquireUninterruptibly(int permits) 
 9 //取得等待许可线程个数
10 public final int getQueueLength() 
11 //是否有线程在等待这个许可
12 public final boolean hasQueuedThreads() 
13 //得到当前可用的许可数目  
14 public int availablePermits();

3、测试

详见本人github：https://github.com/BrokenColor/java-demo 下的 semaphoredemo-包中的测试

另送：Exchanger类

　　Exchanger（交换者）是一个用于线程间协作的工具类。Exchanger用于进行线程间的数据交换。它提供一个同步点，在这个同步点两个线程可以交换彼此的数据。这两个线程通过exchange方法交换数据，如果第一个线程先执行exchange方法，它会一直等待第二个线程也执行exchange，当两个线程都到达同步点时，这两个线程就可以交换数据，将本线程生产出来的数据传递给对方。因此使用Exchanger的重点是成对的线程使用exchange()方法，当有一对线程达到了同步点，就会进行交换数据。因此该工具类的线程对象是成对的。Exchanger类提供了两个方法，String exchange(V x):用于交换，启动交换并等待另一个线程调用exchange；String exchange(V x,long timeout,TimeUnit unit)：用于交换，启动交换并等待另一个线程调用exchange，并且设置最大等待时间，当等待时间超过timeout便停止等待。

 1 /**
 2  * 
 3  */
 4 package concurrenttest.semaphoredemo.pojo;
 5 
 6 import java.util.concurrent.Exchanger;
 7 import java.util.concurrent.TimeUnit;
 8 
 9 /**
10  * 线程 Exchanger A
11  * @author bc
12  * @data 2018年9月12日
13  */
14 public class Thread_ExchangerA extends Thread{
15 
16     private Exchanger<String> exchanger;
17     
18     public Thread_ExchangerA(Exchanger<String> exchanger) {
19         super();
20         this.exchanger = exchanger;
21     }
22     @Override
23     public void run() {
24         try {
25             //exchange()具有阻塞的特点，如果没有其他线程，则一直阻塞等待
26             System.out.println("线程A中得到线程B的值="+exchanger.exchange("中国人A"));
27             //设置超时参数
28 //            System.out.println("线程A中得到线程B的值="+exchanger.exchange("中国人A",5,TimeUnit.SECONDS));
29         } catch (Exception e) {
30             e.getStackTrace();
31         }
32     }
33     
34 }

Thread_ExchangerA

 1 package concurrenttest.semaphoredemo.pojo;
 2 
 3 import java.util.concurrent.Exchanger;
 4 import java.util.concurrent.TimeUnit;
 5 
 6 /**
 7  * 线程 Exchanger B
 8  * @author bc
 9  * @data 2018年9月12日
10  */
11 public class Thread_ExchangerB extends Thread{
12 
13     private Exchanger<String> exchanger;
14     
15     public Thread_ExchangerB(Exchanger<String> exchanger) {
16         super();
17         this.exchanger = exchanger;
18     }
19     @Override
20     public void run() {
21         try {
22             //exchange()具有阻塞的特点，如果没有其他线程，则一直阻塞等待
23             System.out.println("线程B中得到线程A的值="+exchanger.exchange("中国人B"));
24             //设置超时参数
25 //            System.out.println("线程B中得到线程A的值="+exchanger.exchange("中国人B",5,TimeUnit.SECONDS));
26         } catch (Exception e) {
27             e.getStackTrace();
28         }
29     }
30     
31 }

Thread_ExchangerB

 1 package concurrenttest.semaphoredemo.runtest;
 2 
 3 import java.util.concurrent.Exchanger;
 4 
 5 import concurrenttest.semaphoredemo.pojo.Thread_ExchangerA;
 6 import concurrenttest.semaphoredemo.pojo.Thread_ExchangerB;
 7 import concurrenttest.semaphoredemo.pojo.Thread_ExchangerC;
 8 
 9 
10 /**
11  * 线程 Exchanger test
12  * exchange()具有阻塞的特点，如果没有其他线程，则一直阻塞等待
13  * 
14  * Exchanger的典型应用场景是：一个任务在创建对象，而这些对象的生产代价很高，另一个任务在消费这些对象。
15  * 通过这种方式，可以有更多的对象在被创建的同时被消费。
16  * 
17  * @author bc
18  * @data 2018年9月12日
19  */
20 public class RunTest_Exchanger {
21 
22     public static void main(String[] args) {
23         Exchanger<String> exchanger = new Exchanger<String>();
24         
25         Thread_ExchangerA threadA = new Thread_ExchangerA(exchanger);
26         threadA.start();
27         Thread_ExchangerB threadB = new Thread_ExchangerB(exchanger);
28         threadB.start();
29 //        Thread_ExchangerC threadC = new Thread_ExchangerC(exchanger);
30 //        threadC.start();
31         
32         System.out.println("main end");
33     }
34 
35 }

RunTest_Exchanger

运行结果：

Exchanger的典型应用场景是：　　

　　一个任务在创建对象，而这些对象的生产代价很高，另一个任务在消费这些对象。 通过这种方式，可以有更多的对象在被创建的同时被消费。

详见本人github：https://github.com/BrokenColor/java-demo 下的 semaphoredemo-包中的测试