Java中的并发工具类

Posted 2021-01-06 helloduo

CountDownLatch

相当于一个计数器，使用countDown()方法可以对计数器进行减一，如果计数器的值没有减到0，那么await方法后面的的都不执行。

 1 static CountDownLatch c = new CountDownLatch(2);
 2     
 3     public static void main(String[] args) throws InterruptedException {
 4         
 5         new Thread(()->{
 6             
 7             System.out.println(1);
 8             c.countDown();
 9             System.out.println(2);
10             c.countDown();
11             
12         }).start();
13         c.await();
14         System.out.println(3);
15     }

View Code

同步屏障 CyclicBarrier

让一组线程到达屏障的时候被阻塞，直到最后一个到达屏障的时候，所有的线程才会继续执行

 1 static CyclicBarrier c = new CyclicBarrier(2);
 2     
 3     public static void main(String[] args) {
 4         new Thread(()->{
 5             try {
 6                 c.await();
 7             } catch (InterruptedException | BrokenBarrierException e) {
 8                 e.printStackTrace();
 9             }
10             System.out.println(1);
11         }).start();
12         try {
13             c.await();
14         } catch (InterruptedException | BrokenBarrierException e) {
15             // TODO Auto-generated catch block
16             e.printStackTrace();
17         }
18         System.out.println(2);
19     }

View Code

CyclicBarrier方法功能大致一样，但是CyclicBarrier可以进行reset  可以多次使用，适用于复杂的业务场景

Semaphore 信号量

Semaphore可用作流量控制     两个重要的方法，acquire()  和release方法   acquire方法相当于获取一个许可证，如果超过了信号量的大小，则阻塞，release方法用来归还许可证  

 1 private static final int THREAD_COUNT = 30;
 2     private static ExecutorService threadPool = Executors.newFixedThreadPool(THREAD_COUNT);
 3     private static Semaphore s = new Semaphore(10);
 4     public static void main(String[] args) {
 5         for(int i=0; i<THREAD_COUNT; i++) {
 6             threadPool.execute(()->{
 7                 try {
 8                     s.acquire();
 9                     System.out.println("save data");
10                     s.release();
11                 } catch (InterruptedException e) {
12                     // TODO Auto-generated catch block
13                     e.printStackTrace();
14                 }
15                 
16             });
17         }
18         threadPool.shutdown();
19     }

View Code

Exchanger

用于线程间的数据交换

在两个同步点   有一个执行exchange方法，会等待第二个线程也执行exchange方法，进行数据交换

 1 private static final Exchanger<String> change = new Exchanger<String>();
 2     private static ExecutorService threadPool = Executors.newFixedThreadPool(2);
 3     
 4     public static void main(String[] args) {
 5         
 6         threadPool.execute(()->{    
 7             try {
 8                 String a = "dataA";
 9                 change.exchange(a);
10             } catch (InterruptedException e) {
11                 // TODO Auto-generated catch block
12                 e.printStackTrace();
13             }
14         });
15         
16         threadPool.execute(()->{
17             
18             try {
19                 String b = "dataB";
20                 String exA = change.exchange(b);
21                 System.out.println(exA);
22             } catch (InterruptedException e) {
23                 e.printStackTrace();
24             }
25         });
26         threadPool.shutdown();
27     }

View Code