本文共 7057 字，大约阅读时间需要 23 分钟。

java并发工具类

1. 今天主要介绍四个java并发常见的工具类
   • CountDownLatch
   • CyclicBarrier
   • Semaphore
   • Exchanger

CountDownLatch

• 用过Thread的join()都知道，join让“主线程”等待“子线程”结束之后才能继续运行。jdk1.5之后提供了CountDownLatch提供了比join更加丰富的功能，是一种共享锁
• 常见api
  • 构造函数 接收一个int值作为计数器，想n个线程等待n个点就传入N
    public CountDownLatch(int count) {
    if (count < 0) throw new IllegalArgumentException("count < 0");
    this.sync = new Sync(count);
    }
  • .countDown() n - -; 释放共享锁
  • await() n==0 线程执行; CAS

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);    }}

CyclicBarrier

• 表面意思理解为可循环使用的屏障，作用是让一组线程在到达一个屏障时被阻塞，等到最后一个线程到达屏障点，才会运行被拦截的线程继续运行
  • 构造函数 CyclicBarrier(int parties) 屏障拦截的线程数量
  • await() 调用该方法时表示线程已经到达屏障，随即阻塞

public class CyclicBarrierTest {    static CyclicBarrier c = new CyclicBarrier(3);    public static void main(String[] args) {        new Thread(new Runnable() {            @Override            public void run() {                try {                    c.await();                } catch (InterruptedException e) {                    e.printStackTrace();                } catch (BrokenBarrierException e) {                    e.printStackTrace();                }                System.out.println(1);            }        }).start();        try {            c.await();        } catch (InterruptedException e) {            e.printStackTrace();        } catch (BrokenBarrierException e) {            e.printStackTrace();        }        System.out.println(2);    }}

• 高级构造函数 CyclicBarrier(int,Runnable) 优先执行

public class CyclicBarrierTest2 {    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 (InterruptedException e) {                    e.printStackTrace();                } catch (BrokenBarrierException e) {                    e.printStackTrace();                }                System.out.println(1);            }        }).start();        try {            c.await();        } catch (InterruptedException e) {            e.printStackTrace();        } catch (BrokenBarrierException e) {            e.printStackTrace();        }        System.out.println(2);    }    static class A implements Runnable{        @Override        public void run() {            System.out.println(3);        }    }}

• 应用场景
  • 银行流水处理服务，模拟Excel统计多个sheet汇总

public class BankWaterService implements Runnable {
       private CyclicBarrier c = new CyclicBarrier(4,this);    private Executor executor = Executors.newFixedThreadPool(4);    private ConcurrentHashMap
   
     sheetBankWaterCount = new ConcurrentHashMap<>();    private void count(){        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) {                        e.printStackTrace();                    } catch (BrokenBarrierException e) {                        e.printStackTrace();                    }                }            });        }    }    @Override    public void run() {        int result = 0;        for (Map.Entry
    
      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.count();    }}
    
   

3.两者的区别

public class CyclicBarrierTest3 {    static CyclicBarrier c = new CyclicBarrier(2);    public static void main(String[] args) {        Thread thread = new Thread(new Runnable() {            @Override            public void run() {                try {                    c.await();                } catch (InterruptedException e) {                    e.printStackTrace();                } catch (BrokenBarrierException e) {                    e.printStackTrace();                }            }        });        thread.start();        thread.interrupt();        try {            c.await();        } catch (InterruptedException e) {            e.printStackTrace();        } catch (BrokenBarrierException e) {           //e.printStackTrace();            System.out.println(c.isBroken());        }    }}

Semaphore

• 用来控制同时访问特定资源的线程数量，通过协调保证合理的使用公共资源
• 理解 比作控制车流的红绿灯，如马路要控制流量，只限制100辆车通行，其他必须在路口处等待，不能行驶在马路上，当其中有5辆离开马路，那么允许后面5辆进入马路。例子中的车就是线程，驶入马路表示线程正在执行，离开马路表示线程执行完成，看见红灯表示被阻塞。
• 代码

public class SemaphoreTest {
       private static final int THREAD_COUNT = 30;    private static ExecutorService pool = 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++) {            pool.execute(new Runnable() {                @Override                public void run() {                    try {                        s.acquire();                    } catch (InterruptedException e) {                        e.printStackTrace();                    }                }            });        }        pool.shutdown();    }}

• 其他常见api
  • int availablePermits() 返回此信号量中当前可用的许可证数
  • int getQueueLength() 正在等待获取许可证的线程数
  • boolean hasQueuedThreads() 是否有线程正在等待获取许可证
  • void reducePermits() 减少许可证
  • Collection getQueuedThreads() 返回所有等待获取许可证的线程集合

Exchanger

• 线程之间的数据交换，提供一个同步点，两个线程可以交换彼此数据，通过exchange()交换数据，当两个线程都执行到exchange()方法，可以将本线程的数据传递给对方。
• 应用场景
  • 遗传算法
  • 校对工作
• 示例代码

public static final Exchanger
   
     exgr = new Exchanger<>();    public static ExecutorService pool = Executors.newFixedThreadPool(2);    public static void main(String[] args) {        pool.execute(new Runnable() {            @Override            public void run() {                String A = "B";                try {                    exgr.exchange(A);                } catch (InterruptedException e) {                    e.printStackTrace();                }            }        });        pool.execute(new Runnable() {            @Override            public void run() {                try {                String B ="B";                String A = exgr.exchange("B");                    System.out.println("是否一致"+A.equals(B)+"A:::"+A+"B:::"+B);                } catch (InterruptedException e) {                    e.printStackTrace();                }            }        });        pool.shutdown();    }} //output 是否一致trueA:::BB:::B
   

转载地址：http://eqiii.baihongyu.com/