第8章Java中的并发工具类

　　在JDK的并发包里提供了几个非常有用的并发工具类。CountDownLatch、CyclicBarrier和Semaphore工具类提供了一种并发流程控制的手段，Exchanger工具类则提供了在线程间交换数据的一种手段。

　　8.1等待多线程完成的CountDownLatch

　　　　CountDownLatch允许一个或多个线程等待其他线程完成操作。

　　　　假如有这样一个需求：我们需要解析一个Excel中的多个sheet数据，此时考虑多线程，每个线程解析一个sheet数据，所有sheet都解析完成后，程序需要提示解析完成。在这个需求中，需要实现主线程等待所有线程完成sheet解析操作，最简单的方法是使用join()方法，代码如下所示：

　　

package com.example.demo.test;

public class JoinCountDownLatchTest {
    public static void main(String[] args) throws Exception {
        Thread thread1 = new Thread(new Runnable() {
            
            @Override
            public void run() {
                // TODO Auto-generated method stub
                
            }
        });
        Thread thread2 = new Thread(new Runnable() {
            
            @Override
            public void run() {
                // TODO Auto-generated method stub
                
            }
        });
        thread1.start();
        thread2.start();
        thread1.join();
        thread2.join();
        System.out.println("all finish");
    }
}

　　CountDownLatch也可以实现join的功能，并且功能更多，代码如下

package com.example.demo.test;

import java.util.concurrent.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");
    }

}

　　8.2 同步屏障CyclicBarrier

　　　　字面意思是可以循环使用的屏障，它要做的事情就是，让一组线程到达一个屏障之后被阻塞，直到最后一个线程到达屏障时，所有线程继续执行。

public class CyclicBarrierTest {
    static CyclicBarrier cyclicBarrier = new CyclicBarrier(2);
    
    public static void main(String[] args) throws Exception, BrokenBarrierException {
        new Thread(new Runnable() {
            
            @Override
            public void run() {
                try {
                    cyclicBarrier.wait();
                } catch (Exception e) {
                    // TODO: handle exception
                }
                System.out.println(1);
            }
        }).start();
        cyclicBarrier.await();
        System.out.println(2);
    }
}

package com.example.demo.test;

import java.security.KeyStore.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{
    private CyclicBarrier c = new CyclicBarrier(4,this);
    private Executor executor = Executors.newFixedThreadPool(4);
    private ConcurrentHashMap<String, Integer> sheetBankWaterCount = new ConcurrentHashMap<>();
    
    private void counter() {
        for(int i=0;i<4;i++) {
            executor.execute(new Runnable() {    
                @Override
                public void run() {
                    sheetBankWaterCount.put(Thread.currentThread().getName(), 1);
                    try {
                        c.await();
                    } catch (InterruptedException e) {
                        // TODO Auto-generated catch block
                        e.printStackTrace();
                    } catch (BrokenBarrierException e) {
                        // TODO Auto-generated catch block
                        e.printStackTrace();
                    }
                }
            });
        }
    }

    @Override
    public void run() {
        System.out.println(c.getNumberWaiting());
        int result =0;
        for(java.util.Map.Entry<String, Integer> sheet : sheetBankWaterCount.entrySet()) {
            result +=sheet.getValue();
        }
        sheetBankWaterCount.put("result", result);
        System.out.println(result);
    }

    public static void main(String[] args) {
        BankWaterService bankWaterService = new BankWaterService();
        bankWaterService.counter();
    }
}

　　8.3控制并发线程数的Semaphore

public class SemphoreTest {
    private static final int THREAD_COUNT = 30;
    private static ExecutorService threadPool = Executors.newFixedThreadPool(THREAD_COUNT);
    private static Semaphore semaphore = 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 {
                        semaphore.acquire();
                    } catch (InterruptedException e) {
                        // TODO Auto-generated catch block
                        e.printStackTrace();
                    }
                    System.out.println("save data");
                    semaphore.release();
                }
            });
        }
        threadPool.shutdown();
    }
}

　　8.4 线程之间交换数据的Exchanger

　　　　Exchanger（交换者）是一个用于线程间协作的工具类。Exchanger用于进行线程间的数据交换。他提供一个同步点，在这个同步点，两个线程可以彼此交换数据，如果第一个线程先执行exchange()方法，它会一直等待第二个线程也执行exchange()方法，当两个线程都到达同步点时，这两个线程就可以交换数据，将本线程生产出来的数据传递给对方。

　　　　其可用于遗传算法。遗传算法需要选出两个人作为交配对象，这时候会交换两个人的数据，并用交叉规则得到两个交配结果。Exhanger也可以用于校对工作，比如我们需要将银行流水通过人工方式录入电子银行流水，为了避免错误，采用AB岗两人进行录入，录入到Excel后，系统加载这两个Excel，进行比对查看是否一致。

　　

package com.example.demo.test;

import java.util.concurrent.Exchanger;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

public class ExchangerTest {
    private static final Exchanger<String> EXCHANGER = new Exchanger<>();
    private static ExecutorService threadPool = Executors.newFixedThreadPool(2);
    
    public static void main(String[] args) {
        threadPool.execute(new Runnable() {
            
            @Override
            public void run() {
                String A = "A流水";
                try {
                    String B = EXCHANGER.exchange(A);
                    System.out.println("是否一致："+B.equals(A));
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
        });
        threadPool.execute(new Runnable() {
            
            @Override
            public void run() {
                try {
                    String B = "B流水";
                    String A = EXCHANGER.exchange(B);
                    System.out.println("是否一致："+B.equals(A));
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
        });
        threadPool.shutdown();
    }
}