CountDownLatch,Cyclicbarrier和Semaphore工具类提供了一种并发流程控制的手段,Exchanger工具类则提供了在线程间交换数据的一种手段
CountDownLatch允许一个或者多个线程等待其他线程完成操作
public class CountDownLatchTest {
static CountDownLatch c = new CountDownLatch(2);
public static void main(String[] args) throws InterruptedException{
new Thread(new Runnable() {
@Override
public void run() {
System.out.println(1);
c.countDown();
System.out.println(2);
c.countDown();
}
}).start();
c.await();
System.out.println("3");
}
}
计数器必须大于0,只是等于0的时候,计数器就是0,调用await()方法时不会阻塞当前线程,
CountDownLatch不可能重新初始化或者修改CountDownLatch对象的内部计数器的值,
一个线程调用CountDown方法happen-before,另外一个线程调用await()方法
CyclicBarrier的字面意思是可循环使用(Cyclic)的屏障(Barrier),它的功能是,让一组线程到达一个屏障(也可以叫做同步点)时被阻塞,直到最后一个线程到达屏障时,屏障才会开门,所有被屏障拦截的线程才会继续运行。
CyclicBarrier(int parties)
public class CyclicBarrierTest {
static CyclicBarrier c = new CyclicBarrier(2);
public static void main(String[] args) {
new Thread(new Runnable() {
@Override
public void run() {
try {
c.await();
} catch (Exception e) {
e.printStackTrace();
}
System.out.println(1);
}
}).start();
try {
c.await();
} catch (Exception e) {
e.printStackTrace();
}
System.out.println(2);
}
}
//可能输出1,2
//也可能输出2,1
因为主线程和子线程调度是由CPU决定,两个线程都有可能先执行
public class CyclicBarrierTest {
static CyclicBarrier c = new CyclicBarrier(2,new A());
public static void main(String[] args) {
new Thread(new Runnable() {
@Override
public void run() {
try {
c.await();
} catch (Exception e) {
e.printStackTrace();
}
System.out.println(1);
}
}).start();
try {
c.await();
} catch (Exception e) {
e.printStackTrace();
}
System.out.println(2);
}
static class A implements Runnable{
@Override
public void run(){
System.out.println(3);
}
}
}
//结果为3,1,2
Semaphore(信号量)是用来控制同时访问特定资源的线程数量,它通过协调各个线程,以保证合理的使用公共资源
应用场景
public class SemaphoreTest {
private static final int THREAD_COUNT = 30;
private static ExecutorService threadPool = Executors.newFixedThreadPool(THREAD_COUNT);
private static Semaphore s = new Semaphore(10);
public static void main(String[] args) {
for (int i = 0; i < THREAD_COUNT; i++) {
threadPool.execute(new Runnable() {
@Override
public void run() {
try {
s.acquire();
System.out.println("save data");
s.release();
} catch (InterruptedException e) {
e.printStackTrace();
}
}
});
}
threadPool.shutdown();
}
}
在代码中,虽然有30个线程在执行,但是只允许10个并发执行
Semaphore的构造方法Semaphore(int permit)接受一个整型的数字,表示可用的许可证数量,Semaphore(10)表示10个线程获得了许可证,也就是最大并发数为10,
Semaphore的用法很简单,就是先Semaphore的acquire()方法获取一个许可证,使用完后调用release()方法归还许可证
Semaphore的其他方法
Exchanger是线程中协作的工具类。Exchanger用于线程间的数据交换,它提供一个同步点,这个同步点,两个线程可以交换彼此的数据。两个线程通过exchanger()来切换数据
public class ExchangerTest {
private static final Exchanger <String> exgr = new Exchanger<>();
private static ExecutorService threadPool = Executors.newFixedThreadPool(2);
public static void main(String[] args) {
threadPool.execute(new Runnable() {
@Override
public void run() {
try {
String A = "银行流水A"; //A录入银行流水数据
exgr.exchange(A);
} catch (InterruptedException e) {
e.printStackTrace();
}
}
});
threadPool.execute(new Runnable() {
@Override
public void run() {
try {
String B = "银行流水B";//B录入银行
String A = exgr.exchange("B");
System.out.println("A和B数据是否一致:"+A.equals(B) +",A录入的是:"+A+"B录入的是"+B);
} catch (InterruptedException e) {
e.printStackTrace();
}
}
});
threadPool.shutdown();
}
}
//A和B数据是否一致:false,A录入的是:银行流水A B录入的是银行流水B
如果两个线程有一个没有执行exchange()方法,则会一直等待,如果担心有特殊情况发生,避免一直等待,可以使用exchange(V x,longtimeout,TimeUnit unit)设置最大等待时长
原文：https://www.cnblogs.com/sgw1018/p/Java-concurrent-tool.html