JUC是指java.util.concurrent包，里面封装了解决并发编程的一些工具，下面温习一下CountDownLatch、CyclicBarrier、Semaphore、ExChanger的使用。
1、Semaphore
Semaphore，是信号量，作用于控制同时访问某些资源的线程数量，用在流量控制。源码如下
/**
* Creates a {@code Semaphore} with the given number of
* permits and nonfair fairness setting.
*
* @param permits the initial number of permits available.
* This value may be negative, in which case releases
* must occur before any acquires will be granted.
*/
public Semaphore(int permits) {
sync = new NonfairSync(permits);
}
/**
* Creates a {@code Semaphore} with the given number of
* permits and the given fairness setting.
*
* @param permits the initial number of permits available.
* This value may be negative, in which case releases
* must occur before any acquires will be granted.
* @param fair {@code true} if this semaphore will guarantee
* first-in first-out granting of permits under contention,
* else {@code false}
*/
public Semaphore(int permits, boolean fair) {
sync = fair ? new FairSync(permits) : new NonfairSync(permits);
}
在源码中可以看到提供了两种构造方法，为非公平方式和公平方式，在构造方法中，均需要传入许可的数量，这个数量就是资源允许访问的最大线程数。
public class SemaphoreTest {
public static void main(String[] args) {
// 5辆班车
Semaphore semaphore = new Semaphore(5);
// 10组人，一组人一辆车
// for (int i = 0; i < 10; i++) {
// Car car = new Car(i, semaphore);
// car.start();
// }
// 10组人，一组人一辆车
for (int i = 0; i < 10; i++) {
Bus bus = new Bus(i, semaphore);
bus.start();
}
}
static class Car extends Thread {
public int num;
public Semaphore semaphore;
public Car(int num, Semaphore semaphore) {
this.num = num;
this.semaphore = semaphore;
}
@Override
public void run() {
try {
// 堵塞式
semaphore.acquire();
System.out.printf("第%s辆班车出发了\n",num+1);
Thread.sleep(new Random().nextInt(20) * 1000);
semaphore.release();
System.out.printf("第%s辆班车送完员工回来了\n",num+1);
} catch (InterruptedException e) {
e.printStackTrace();
}
}
}
static class Bus extends Thread {
public int num;
public Semaphore semaphore;
public Bus(int num, Semaphore semaphore) {
this.num = num;
this.semaphore = semaphore;
}
@Override
public void run() {
try {
// 非堵塞式
boolean acquire = semaphore.tryAcquire();
if (acquire) {
System.out.printf("第%s辆班车出发了\n",num+1);
Thread.sleep(new Random().nextInt(20) * 1000);
semaphore.release();
System.out.printf("第%s辆班车送完员工回来了\n",num+1);
} else {
System.out.printf("未赶上第%s辆班车\n",num+1);
}
} catch (InterruptedException e) {
e.printStackTrace();
}
}
}
}
通过结果可知，acquire为堵塞式，tryAcquire为非堵塞式，可以通过返回的Boolean值判断是否执行。
2、CyclicBarrier
CyclicBarrier，是栅栏锁，作用于让一组线程达到某个屏障被堵塞，直到组内最后一个线程到达，然后屏障开发，所有线程继续执行。源码如下
/**
* Creates a new {@code CyclicBarrier} that will trip when the
* given number of parties (threads) are waiting upon it, and which
* will execute the given barrier action when the barrier is tripped,
* performed by the last thread entering the barrier.
*
* @param parties the number of threads that must invoke {@link #await}
* before the barrier is tripped
* @param barrierAction the command to execute when the barrier is
* tripped, or {@code null} if there is no action
* @throws IllegalArgumentException if {@code parties} is less than 1
*/
public CyclicBarrier(int parties, Runnable barrierAction) {
if (parties <= 0) throw new IllegalArgumentException();
this.parties = parties;
this.count = parties;
this.barrierCommand = barrierAction;
}
/**
* Creates a new {@code CyclicBarrier} that will trip when the
* given number of parties (threads) are waiting upon it, and
* does not perform a predefined action when the barrier is tripped.
*
* @param parties the number of threads that must invoke {@link #await}
* before the barrier is tripped
* @throws IllegalArgumentException if {@code parties} is less than 1
*/
public CyclicBarrier(int parties) {
this(parties, null);
}
从源码中可以看到有两个构造方法，下面的构造方法实际上是调用了上面的构造方法，均需要传入int类型参数，这个值也是执行线程的数量，另一个参数是barrierAction线程，当屏障开发后需要执行的操作可以放在里面。
public class CyclicBarrierTest {
private static ConcurrentHashMap<String, Long> resultMap = new ConcurrentHashMap<>();
private static CyclicBarrier cyclicBarrier = new CyclicBarrier(5,new CollectThread());
public static void main(String[] args) {
for (int i = 0; i < 5; i++) {
SubTread subTread = new SubTread();
subTread.start();
}
}
static class CollectThread extends Thread {
@Override
public void run() {
StringBuffer result = new StringBuffer();
for (Map.Entry item : resultMap.entrySet()) {
result.append("[" + item.getValue() + "]");
}
System.out.printf("result = %s\n",result);
}
}
static class SubTread extends Thread {
@Override
public void run() {
long id = Thread.currentThread().getId();
resultMap.put(String.valueOf(id),id);
try {
System.out.printf("Thread_%s do business\n",id);
Thread.sleep(new Random().nextInt(20) * 1000);
cyclicBarrier.await();
} catch (InterruptedException e) {
e.printStackTrace();
} catch (Exception e) {
e.printStackTrace();
}
}
}
}
通过结果可知，主线程创建了5个子线程，子线程执行完成后都调用了await，然后他们都在一个屏障前堵塞着，等到最后一个子线程也执行完成，调用await后，屏障开放，开始执行barrierAction线程里的操作。
3、CountDownLatch
CountDownLatch，是闭锁，作用于让一组线程等待其他线程完成之后才执行。源码如下
/**
* Constructs a {@code CountDownLatch} initialized with the given count.
*
* @param count the number of times {@link #countDown} must be invoked
* before threads can pass through {@link #await}
* @throws IllegalArgumentException if {@code count} is negative
*/
public CountDownLatch(int count) {
if (count < 0) throw new IllegalArgumentException("count < 0");
this.sync = new Sync(count);
}
从源码中看到，构造方法中需要传入int类型参数，这个值是被扣减的初始数量。
public class CountDownLatchTest {
private static CountDownLatch countDownLatch = new CountDownLatch(5);
public static void main(String[] args) throws InterruptedException {
// 启动业务线程
BusinessTread businessTread = new BusinessTread();
businessTread.start();
// 启动初始化线程
for (int i = 0; i < 5; i++) {
InitTread initTread = new InitTread();
initTread.start();
}
// 主线程进入等待
countDownLatch.await();
System.out.printf("main thread do ... \n");
}
static class InitTread extends Thread {
@Override
public void run() {
long id = Thread.currentThread().getId();
System.out.printf("Thread_%s init ... \n",id);
try {
Thread.sleep(new Random().nextInt(20) * 1000);
} catch (InterruptedException e) {
e.printStackTrace();
}
countDownLatch.countDown();
}
}
static class BusinessTread extends Thread {
@Override
public void run() {
try {
long id = Thread.currentThread().getId();
System.out.printf("Thread_%s ready do ... \n",id);
countDownLatch.await();
System.out.printf("Thread_%s do business ... \n",id);
} catch (InterruptedException e) {
e.printStackTrace();
}
}
}
}
通过结果可知，业务线程和主线程等到5个初始化线程完成之后才开始执行。
4、ExChanger
ExChanger，是交换器，作用于线程之间交换数据，但比较受限，只能两个线程之间交换数据。源码如下
/**
* Creates a new Exchanger.
*/
public Exchanger() {
participant = new Participant();
}
从源码中看到，构造方法没有入参，只是在创建时指定一下需要交换的数据的泛型。
public class ExchangerTest {
private static final Exchanger<Set<String>> exchanger = new Exchanger<>();
public static void main(String[] args) {
new Thread(){
@Override
public void run() {
Set<String> first = new HashSet<>();
first.add("A");
first.add("B");
first.add("C");
try {
Set<String> exchange = exchanger.exchange(first);
for (String s : exchange) {
System.out.println("first="+s);
}
} catch (InterruptedException e) {
e.printStackTrace();
}
}
}.start();
new Thread(){
@Override
public void run() {
Set<String> second = new HashSet<>();
second.add("1");
second.add("2");
second.add("3");
try {
Set<String> exchange = exchanger.exchange(second);
for (String s : exchange) {
System.out.println("second="+s);
}
} catch (InterruptedException e) {
e.printStackTrace();
}
}
}.start();
}
}
通过结果可知，两个线程中的数据发生了交换。
小结：
CountDownLatch调用方法有await、countDown；
CycleBarrier调用方法有await；
Semaphore调用方法有acquire/tryAcquire、release；
CountDownLatch允许一个或多个线程，等待其他一组线程完成之后在继续执行；
CycleBarrier允许同一组线程之间等待，到达同一个屏障之后在继续执行；
Semaphore控制访问指定资源的线程数量；
Exchanger允许两个线程之间交换数据；
