1.等待多线程完成的CountDownLatch
简介
CountDownLatch 允许一个或多个线程等待其他线程完成操作。
应用场景
假如有这样一个需求,当我们需要解析一个Excel里多个sheet的数据时,可以考虑使用多线程,每个线程解析一个sheet里的数据,等到所有的
sheet都解析完之后,程序需要提示解析完成。在这个需求中,要实现主线程等待所有线程完成sheet的解析操作,最简单的做法是使用join。代码如下:
01 |
public class JoinCountDownLatchTest { |
02 |
03 |
public static void main(String[] args) throws InterruptedException { |
04 |
Thread parser1 = new Thread( new Runnable() { |
05 |
@Override |
06 |
public void run() { |
07 |
} |
08 |
}); |
09 |
10 |
Thread parser2 = new Thread( new Runnable() { |
11 |
@Override |
12 |
public void run() { |
13 |
System.out.println( "parser2 finish" ); |
14 |
} |
15 |
}); |
16 |
17 |
parser1.start(); |
18 |
parser2.start(); |
19 |
parser1.join(); |
20 |
parser2.join(); |
21 |
System.out.println( "all parser finish" ); |
22 |
} |
23 |
24 |
} |
join用于让当前执行线程等待join线程执行结束。其实现原理是不停检查join线程是否存活,如果join线程存活则让当前线程永远wait,代码片段如下,
wait(0)表示永远等待下去。
1 |
while (isAlive()) { |
2 |
wait( 0 ); |
3 |
} |
直到join线程中止后,线程的this.notifyAll会被调用,调用notifyAll是在JVM里实现的,所以JDK里看不到,有兴趣的同学可以看看JVM源码。
JDK不推荐在线程实例上使用wait,notify和notifyAll方法。
而在JDK1.5之后的并发包中提供的CountDownLatch也可以实现join的这个功能,并且比join的功能更多。
01 |
public class CountDownLatchTest { |
02 |
03 |
static CountDownLatch c = new CountDownLatch( 2 ); |
04 |
05 |
public static void main(String[] args) throws InterruptedException { |
06 |
new Thread( new Runnable() { |
07 |
@Override |
08 |
public void run() { |
09 |
System.out.println( 1 ); |
10 |
c.countDown(); |
11 |
System.out.println( 2 ); |
12 |
c.countDown(); |
13 |
} |
14 |
}).start(); |
15 |
16 |
c.await(); |
17 |
System.out.println( "3" ); |
18 |
} |
19 |
20 |
} |
CountDownLatch的构造函数接收一个int类型的参数作为计数器,如果你想等待N个点完成,这里就传入N。
当我们调用一次CountDownLatch的countDown方法时,N就会减1,CountDownLatch的await会阻塞当前线程,直到N变成零。由于countDown方法
可以用在任何地方,所以这里说的N个点,可以是N个线程,也可以是1个线程里的N个执行步骤。用在多个线程时,你只需要把这个CountDownLatch的
引用传递到线程里。
其他方法
如果有某个解析sheet的线程处理的比较慢,我们不可能让主线程一直等待,所以我们可以使用另外一个带指定时间的await方法,
await(long time, TimeUnit unit): 这个方法等待特定时间后,就会不再阻塞当前线程。join也有类似的方法。
注意:计数器必须大于等于0,只是等于0时候,计数器就是零,调用await方法时不会阻塞当前线程。CountDownLatch不可能重新初始化或者
修改CountDownLatch对象的内部计数器的值。一个线程调用countDown方法 happen-before 另外一个线程调用await方法。、
2.同步屏障CyclicBarrier
简介
CyclicBarrier 的字面意思是可循环使用(Cyclic)的屏障(Barrier)。它要做的事情是,让一组线程到达一个屏障(也可以叫同步点)时被阻塞,直到最后一个线程到达屏障时,屏障才会开门,所有被屏障拦截的线程才会继续干活。CyclicBarrier默认的构造方法是CyclicBarrier(int parties),其参数表示屏障拦截的线程数量,每个线程调用await方法告诉CyclicBarrier我已经到达了屏障,然后当前线程被阻塞。
实例代码如下:
01 |
public class CyclicBarrierTest { |
02 |
03 |
static CyclicBarrier c = new CyclicBarrier( 2 ); |
04 |
05 |
public static void main(String[] args) { |
06 |
new Thread( new Runnable() { |
07 |
08 |
@Override |
09 |
public void run() { |
10 |
try { |
11 |
c.await(); |
12 |
} catch (Exception e) { |
13 |
14 |
} |
15 |
System.out.println( 1 ); |
16 |
} |
17 |
}).start(); |
18 |
19 |
try { |
20 |
c.await(); |
21 |
} catch (Exception e) { |
22 |
23 |
} |
24 |
System.out.println( 2 ); |
25 |
} |
26 |
} |
输出
1 |
2 |
2 |
1 |
或者输出
1 |
1 |
2 |
2 |
如果把new CyclicBarrier(2)修改成new CyclicBarrier(3)则主线程和子线程会永远等待,因为没有第三个线程执行await方法,即没有第三个线程到达屏障,所以之前到达屏障的两个线程都不会继续执行。
CyclicBarrier还提供一个更高级的构造函数CyclicBarrier(int parties, Runnable barrierAction),用于在线程到达屏障时,优先执行barrierAction,方便处理更复杂的业务场景。代码如下:
01 |
public class CyclicBarrierTest2 { |
02 |
03 |
static CyclicBarrier c = new CyclicBarrier( 2 , new A()); |
04 |
05 |
public static void main(String[] args) { |
06 |
new Thread( new Runnable() { |
07 |
08 |
@Override |
09 |
public void run() { |
10 |
try { |
11 |
c.await(); |
12 |
} catch (Exception e) { |
13 |
14 |
} |
15 |
System.out.println( 1 ); |
16 |
} |
17 |
}).start(); |
18 |
19 |
try { |
20 |
c.await(); |
21 |
} catch (Exception e) { |
22 |
23 |
} |
24 |
System.out.println( 2 ); |
25 |
} |
26 |
27 |
static class A implements Runnable { |
28 |
29 |
@Override |
30 |
public void run() { |
31 |
System.out.println( 3 ); |
32 |
} |
33 |
34 |
} |
35 |
36 |
} |
输出
1 |
3 |
2 |
1 |
3 |
2 |
CyclicBarrier的应用场景
CyclicBarrier可以用于多线程计算数据,最后合并计算结果的应用场景。比如我们用一个Excel保存了用户所有银行流水,每个Sheet保存一个帐户近一年的每笔银行流水,现在需要统计用户的日均银行流水,先用多线程处理每个sheet里的银行流水,都执行完之后,得到每个sheet的日均银行流水,最后,再用barrierAction用这些线程的计算结果,计算出整个Excel的日均银行流水。
CyclicBarrier和CountDownLatch的区别
- CountDownLatch的计数器只能使用一次。而CyclicBarrier的计数器可以使用reset() 方法重置。所以CyclicBarrier能处理更为复杂的业务场景,比如如果计算发生错误,可以重置计数器,并让线程们重新执行一次。
- CyclicBarrier还提供其他有用的方法,比如getNumberWaiting方法可以获得CyclicBarrier阻塞的线程数量。isBroken方法用来知道阻塞的线程是否被中断。比如以下代码执行完之后会返回true。
isBroken的使用代码如下:
01 |
import java.util.concurrent.BrokenBarrierException; |
02 |
import java.util.concurrent.CyclicBarrier; |
03 |
04 |
public class CyclicBarrierTest3 { |
05 |
06 |
static CyclicBarrier c = new CyclicBarrier( 2 ); |
07 |
08 |
public static void main(String[] args) throws InterruptedException, BrokenBarrierException { |
09 |
Thread thread = new Thread( new Runnable() { |
10 |
11 |
@Override |
12 |
public void run() { |
13 |
try { |
14 |
c.await(); |
15 |
} catch (Exception e) { |
16 |
} |
17 |
} |
18 |
}); |
19 |
thread.start(); |
20 |
thread.interrupt(); |
21 |
try { |
22 |
c.await(); |
23 |
} catch (Exception e) { |
24 |
System.out.println(c.isBroken()); |
25 |
} |
26 |
} |
27 |
} |
输出
true3.控制并发线程数的Semaphore
简介
Semaphore(信号量)是用来控制同时访问特定资源的线程数量,它通过协调各个线程,以保证合理的使用公共资源。很多年以来,我都觉得从字面上很难理解Semaphore所表达的含义,只能把它比作是控制流量的红绿灯,比如XX马路要限制流量,只允许同时有一百辆车在这条路上行使,其他的都必须在路口等待,所以前一百辆车会看到绿灯,可以开进这条马路,后面的车会看到红灯,不能驶入XX马路,但是如果前一百辆中有五辆车已经离开了XX马路,那么后面就允许有5辆车驶入马路,这个例子里说的车就是线程,驶入马路就表示线程在执行,离开马路就表示线程执行完成,看见红灯就表示线程被阻塞,不能执行。
应用场景
Semaphore可以用于做流量控制,特别公用资源有限的应用场景,比如数据库连接。假如有一个需求,要读取几万个文件的数据,因为都是IO密集型任务,我们可以启动几十个线程并发的读取,但是如果读到内存后,还需要存储到数据库中,而数据库的连接数只有10个,这时我们必须控制只有十个线程同时获取数据库连接保存数据,否则会报错无法获取数据库连接。这个时候,我们就可以使用Semaphore来做流控,代码如下:
01 |
public class SemaphoreTest { |
02 |
03 |
private static final int THREAD_COUNT = 30 ; |
04 |
05 |
private static ExecutorService threadPool = Executors |
06 |
.newFixedThreadPool(THREAD_COUNT); |
07 |
08 |
private static Semaphore s = new Semaphore( 10 ); |
09 |
10 |
public static void main(String[] args) { |
11 |
for ( int i = 0 ; i < THREAD_COUNT; i++) { |
12 |
threadPool.execute( new Runnable() { |
13 |
@Override |
14 |
public void run() { |
15 |
try { |
16 |
s.acquire(); |
17 |
System.out.println( "save data" ); |
18 |
s.release(); |
19 |
} catch (InterruptedException e) { |
20 |
} |
21 |
} |
22 |
}); |
23 |
} |
24 |
25 |
threadPool.shutdown(); |
26 |
} |
27 |
} |
在代码中,虽然有30个线程在执行,但是只允许10个并发的执行。Semaphore的构造方法Semaphore(int permits) 接受一个整型的数字,表示可用的许可证数量。Semaphore(10)表示允许10个线程获取许可证,也就是最大并发数是10。Semaphore的用法也很简单,首先线程使用Semaphore的acquire()获取一个许可证,使用完之后调用release()归还许可证。还可以用tryAcquire()方法尝试获取许可证。
其他方法
Semaphore还提供一些其他方法:
- int availablePermits() :返回此信号量中当前可用的许可证数。
- int getQueueLength():返回正在等待获取许可证的线程数。
- boolean hasQueuedThreads() :是否有线程正在等待获取许可证。
- void reducePermits(int reduction) :减少reduction个许可证。是个protected方法。
- Collection getQueuedThreads() :返回所有等待获取许可证的线程集合。是个protected方法。
简介
Exchanger(交换者)是一个用于线程间协作的工具类。Exchanger用于进行线程间的数据交换。它提供一个同步点,在这个同步点两个线程可以交换彼此的数据。这两个线程通过exchange方法交换数据, 如果第一个线程先执行exchange方法,它会一直等待第二个线程也执行exchange,当两个线程都到达同步点时,这两个线程就可以交换数据,将本线程生产出来的数据传递给对方。
Exchanger的应用场景
Exchanger可以用于遗传算法,遗传算法里需要选出两个人作为交配对象,这时候会交换两人的数据,并使用交叉规则得出2个交配结果。
Exchanger也可以用于校对工作。比如我们需要将纸制银流通过人工的方式录入成电子银行流水,为了避免错误,采用AB岗两人进行录入,录入到Excel之后,系统需要加载这两个Excel,并对这两个Excel数据进行校对,看看是否录入的一致。代码如下:
01 |
public class ExchangerTest { |
02 |
03 |
private static final Exchanger<String> exgr = new Exchanger<String>(); |
04 |
05 |
private static ExecutorService threadPool = Executors.newFixedThreadPool( 2 ); |
06 |
07 |
public static void main(String[] args) { |
08 |
09 |
threadPool.execute( new Runnable() { |
10 |
@Override |
11 |
public void run() { |
12 |
try { |
13 |
String A = "银行流水A" ; // A录入银行流水数据 |
14 |
exgr.exchange(A); |
15 |
} catch (InterruptedException e) { |
16 |
} |
17 |
} |
18 |
}); |
19 |
20 |
threadPool.execute( new Runnable() { |
21 |
@Override |
22 |
public void run() { |
23 |
try { |
24 |
String B = "银行流水B" ; // B录入银行流水数据 |
25 |
String A = exgr.exchange( "B" ); |
26 |
System.out.println( "A和B数据是否一致:" + A.equals(B) + ",A录入的是:" |
27 |
+ A + ",B录入是:" + B); |
28 |
} catch (InterruptedException e) { |
29 |
} |
30 |
} |
31 |
}); |
32 |
33 |
threadPool.shutdown(); |
34 |
35 |
} |
36 |
} |
其他方法
如果两个线程有一个没有到达exchange方法,则会一直等待,如果担心有特殊情况发生,避免一直等待,可以使用exchange(V x, long timeout, TimeUnit unit)设置最大等待时长。
转载: 并发编程网 – ifeve.com