1.等待多线程完成的CountDownLatch
允许一个或多个线程等待其他线程完成操作。
join用于让当前执行线程等待join线程执行结束。其实现原理是不停检查join线程是否存活，如果join线程存活则让当前线程永远的等待。其中，wait(0)表示永远等待下去，代码片段如下
while(isAlive()){
wait(0);
}
直到线程终止后，线程的this.notifyAll()方法会被调用，调用notifyAll()方法是在JVM里实现。
package 并发工具;
/**
* join用于让当前执行线程等待直到调用join的线程执行结束。
* while(isAlive()){
* wait(0)
* }
* @author Time
*
*/
public class JoinCountDownLatch {
public static void main(String[] args) throws InterruptedException{
// TODO Auto-generated method stub
Thread parser1=new Thread(new Runnable(){
@Override
public void run() {
// TODO Auto-generated method stub
System.out.println("paraser1 finish");
}
});
Thread parser2=new Thread(new Runnable(){
@Override
public void run(){
System.out.println("paraser2 finish");
}
});
parser1.start();
parser2.start();
parser1.join();
parser2.join();
System.out.println("all parser finish");
}
}
CountDownLatch的构造方法接收一个int类型的参数作为计数器，如果你想等待N个点完成，这里就传入N。
当我们调用CountDownLatch的countDown方法时，N就会减一，CountDownLatch的await方法会阻塞当前线程，直到N变成零，由于countDown方法可以用在任何地方，所以这里说的N个点，可以是N个线程，也可以是1个线程里的N个执行步骤。用在多个线程时，只需要把这个CountDownLatch的引用传递到线程里即可。
如果有某个解析的线程处理的比较慢，我们不可能让主线程一直等待，所以可以使用另一个带指定时间的await方法-await(long time,TimeUnit unit)，这个方法等待特定时间后，就会不在阻塞当前线程，join也有类似的方法。
package 并发工具;
import java.util.concurrent.CountDownLatch;
/**
* CountDownLatch的构造方法接收一个int类型的参数作为计数器，
* 如果你想等待N个点完成，这里就传入N。
* 当我们调用CountDownLatch的countDown方法时，N就会减一，
* CountDownLatch的await方法会阻塞当前线程，直到N变成零，
* 由于countDown方法可以用在任何地方，所以这里说的N个点，
* 可以是N个线程，也可以是1个线程里的N个执行步骤。用在多个线程时，
* 只需要把这个CountDownLatch的引用传递到线程里即可。
*
*
* 如果有某个解析的线程处理的比较慢，我们不可能让主线程一直等待，
* 所以可以使用另一个带指定时间的await方法-await(long time,TimeUnit unit)，
* 这个方法等待特定时间后，就会不在阻塞当前线程，join也有类似的方法。
* @author Time
*
*/
public class CountDownLatchTest {
static CountDownLatch c=new CountDownLatch(2);
public static void main(String[] args) throws InterruptedException{
// TODO Auto-generated method stub
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");
}
}
2.
CyclicBarrier的字面意思是可循环使用的屏障。直到所有线程到达屏障时，屏障才会开门。
CyclicBarrier默认的构造方法是CyclicBarrier(int a),其参数表示屏障拦截的线程数量，每个线程调用await方法告诉CyclicBarrier我已经到达了屏障，然后当前线程被阻塞。
因为主线程和子线程的调度是由CPU决定的，两个线程都有可能先执行，所以会产生不同的输出。
import java.util.concurrent.BrokenBarrierException;
import java.util.concurrent.CyclicBarrier;
public class CyclicBarrierTest{
static CyclicBarrier c=new CyclicBarrier(3);
public static void main(String[] args) {
// TODO Auto-generated method stub
new Thread(new Runnable(){
@Override
public void run(){
try {
c.await();
} catch (InterruptedException | BrokenBarrierException e) {
// TODO Auto-generated catch block
e.printStackTrace();
}
System.out.println(1);
}
}).start();
try{
c.await();
}catch(Exception e){
}
System.out.println(2);
}
}
CyclicBarrier还提供了一个更高级的构造方法CyclicBarrier(int a,Runnable barrierAction),用于到达屏障时，优先执行barrierAction，方便处理更复杂的业务场景
import java.util.concurrent.BrokenBarrierException;
import java.util.concurrent.CyclicBarrier;
public class CyclicBarrierTest2 {
static CyclicBarrier c=new CyclicBarrier(2,new A());
public static void main(String[] args) {
// TODO Auto-generated method stub
new Thread(new Runnable(){
@Override
public void run(){
try {
c.await();
} catch (InterruptedException | BrokenBarrierException e) {
// TODO Auto-generated catch block
e.printStackTrace();
}
System.out.println(1);
}
}).start();
try {
c.await();
} catch (InterruptedException | BrokenBarrierException e) {
// TODO Auto-generated catch block
e.printStackTrace();
}
System.out.println(2);
}
static class A implements Runnable{
@Override
public void run(){
System.out.println(3);
}
}
}
CyclicBarrier的应用场景
CyclicBarrier可以用于多线程计算数据，最后合并计算结果的场景。
例子：计算流水账
import java.util.Map.Entry;
import java.util.concurrent.BrokenBarrierException;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.CyclicBarrier;
import java.util.concurrent.Executor;
import java.util.concurrent.Executors;
public class BankWaterService implements Runnable{
/**
* 创建4个屏障处理完之后执行当前类的run方法
*/
private CyclicBarrier c=new CyclicBarrier(4,this);
/**
* 假设只有4个sheet，所以只启动4个线程
*/
private Executor executor=Executors.newFixedThreadPool(4);
/**
* 保存每个sheet计算出的银流结果
*/
private ConcurrentHashMap<String,Integer> sheetbankwatercount=new ConcurrentHashMap<String,Integer>();
private void count(){
for(int i=0;i<4;i++)
{
executor.execute(new Runnable(){
@Override
public void run() {
// TODO Auto-generated method stub
sheetbankwatercount.put(Thread.currentThread().getName(), 1);
try {
c.await();
} catch (InterruptedException | BrokenBarrierException e) {
// TODO Auto-generated catch block
e.printStackTrace();
}
}
});
}
}
public static void main(String[] args) {
// TODO Auto-generated method stub
BankWaterService bwc=new BankWaterService();
bwc.count();
}
@Override
public void run() {
// TODO Auto-generated method stub
int result=0;
for(Entry<String,Integer>sheet:sheetbankwatercount.entrySet()){
result+=sheet.getValue();
}
}
}
3.控制并发线程数的Semaphore
Semaphore(信号量)是用来控制同时访问特定资源的线程数量，它通过协调各个资源的线程数量，以保证合理的使用公共资源。
应用场景
Semaphore可以用作流量控制，特别是公用资源有限的应用场景，比如数据库连接。假如有一个寻求，要读取几万个文件的数据，因为都是IO密集型任务，我们可以启动几十个线程并发地读取，但是如果读到内存后，还需要存储到数据库中，而数据库的连接数只有10个，这时我们必须控制只有10个线程同时获取数据库连接保存数据，否则会报错无法获取数据库连接。这个时候，就可以用Semaphore来做流量控制，代码如下：
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.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) {
// TODO Auto-generated method stub
for(int i=0;i<THREAD_COUNT;i++){
threadPool.execute(new Runnable(){
@Override
public void run(){
try{
s.acquire();
System.out.println("sava data");
s.release();
}catch(InterruptedException e){
}
}
});
}
threadPool.shutdown();
}
}
在代码中虽有30个线程在执行，但是只允许10个并发执行。Semaphore的构造方法Semaphore(int a)接受一个整型的数字，表示可用的许可证数量。Semaphore(10)表示允许10个线程获取许可证，也就是最大的并发数是10.Semaohore的用法也很简单，首先线程使用acquire()方法获取许可，使用之后调用relase()方法归还许可证。还可以用tryAcquire()方法来获取许可证。
| 
转播
