多线程之美3一Java并发工具类

Posted 2020-11-26 flydashpig

目录结构

一、简介
      1.1、Semaphore
      1.2、CountDownLatch
      1.3、CyclicBarrier 
二、信号量Semaphore
      2.1、构造方法
      2.2、主要方法
      2.3、示例代码
三、计数器CountDownLatch
      3.1、主要方法
      3.2、示例代码1
      3.3、示例代码2
四、栅栏类CyclicBarrier
      4.1、构造方法
      4.2、示例代码

一、简介

1.1、 Semaphore

信号量，见文知义，常用于并发控制中的限流作用，我理解是限定数量的共享锁机制。该共享资源最多同时可让n个线程访问,超过n个线程就阻塞等待，如有资源空闲, 唤醒其他等待线程（唤醒又分公平与非公平，默认非公平）比如一条四车道大桥，每次仅能并发通过4辆汽车，而在高峰期时100辆车涌入，这次需要一个信号灯来限制车辆，每次最多放行4辆车，在车辆通过后再放行。在并发环境下，每辆车就是如一个线程，4车道大桥就如有限的资源，需要控制线程的数量，在这种业务场景下，靠锁同步的机制（如synchronized)力有不逮，java并发包中提供Semaphore类可以帮助解决此类场景。

应用场景： 
1、资源控制：控制数据库连接数，如有多个IO操作，每个需要操作数据库写入，数据库连接数有限，控制连接数据库数量。
例:100个线程执行IO,只有10个mysql连接，最多同时可以有10个线程获取到连接，否则会报错无法获取连接，这时可用信号量控制。

2、可当同步锁使用，设置信号量通道等于1。

1.2、CountDownLatch

允许一个或多个线程等待其他线程完成操作后再执行。其内部维护一个计数器，设置初始值给state，每调用 countDown()方法一次,state数量减1,调用await()方法的线程被阻塞，需要等待state减少为0时才可被唤醒继续执行。

应用场景： 
1、一个任务要统计公司一星期的财务流水总额，每次需要读取5张Excel表统计流水汇总，如何快速地统计出来?
可以使用CountDownLatch,先开始5个线程并发地分别统计每张表的流水额度总和，当5个线程统计结束，再汇总总额。

2、开发对外接口,要求响应快，而该接口内部逻辑复杂，涉及多个服务的调用，并依赖这些独立服务响应结果进行下一步操作,这时可以考虑CountDownLatch或者 CyclicBarrier,并发调用多个服务，获取这些结果后才进行下一步，缩短处理时间。

1.3、CyclicBarrier

循环屏障，栅栏类; CyclicBarrier 可让多个线程相互等待,当所有线程都达到后再唤醒所有线程继续执行。如导游设定目标点，所有游客到这集合，先到的游客等待其他游客，当所有的游客都到了后，大家再一起出发。

CyclicBarrier与CountDownLatch区别：
1)  CyclicBarrier 在应用场景中，多个线程之间相互等待，线程之间在业务上可能会更有依赖性 ; CountDownLatch是每个执行 coutDown方法的线程之间可以没有依赖性，而是执行await方法的线程更依赖这些执行coutDown的线程。

2）CountDownLatch计数器只能使用一次，Semaphore可以多次使用，可以重置使用。

3）CyclicBarrier 可以处理更复杂的业务场景，如线程都达到屏障后，可以在构造函数中 CyclicBarrier(int parties, Runnable barrierAction) 传入线程barrierAction，当达到屏障触发条件时，可以比其他等待线程优先执行，处理业务。

4）CyclicBarrier还有很多方法，如查看当前到达屏障被阻塞的线程数量 getNumberWaiting()。

5）CountDownLatch 是每个线程执行完减1操作，当计数器为0时，才唤醒等待线程，阻塞线程只有1个或者多个；CyclicBarrier是让所有线程到达屏障处就被阻塞了，当所有线程都到达时，唤醒所有被阻塞线程继续执行，阻塞线程有多个，线程之间是相互等待。

二、信号量Semaphore

2.1、构造方法

Semaphore(int permits) //默认非公平
//是否公平尝试获取许可证  
Semaphore(int permits, boolean fair)

2.2、主要方法

acquire() //从信号量那去获取一个许可，如果没有剩余的就被阻塞
acquire(int permits)//也可一次获取多个许可
release() //释放一个许可，将其返还给信号量，给其他线程使用。 
release(int permits) //一次释放多个许可

2.3、示例代码

package Semaphore;
import java.text.SimpleDateFormat;
import java.util.Date;
import java.util.concurrent.Semaphore;
import java.util.concurrent.TimeUnit;
/**
 * @author zdd
 * 2019/11/27 6:43
 * Description: 信号量测试
 */
public class SemaphoreService {
    private  static SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");

    private Semaphore semaphore = new Semaphore(4);

    public void doSomeThing() throws InterruptedException {
        //1,获取一个许可
        semaphore.acquire();

        System.out.println(Thread.currentThread().getName() + "--start--" +getFormatDate());
        // 停顿1s
        TimeUnit.SECONDS.sleep(1);
        System.out.println(Thread.currentThread().getName() + "--end--" +getFormatDate());
        //2,用完释放一个许可，可供其他线程使用
        semaphore.release();
    }
    public static void main(String[] args) {
        SemaphoreService semaphoreService  = new SemaphoreService();
        for (int i = 0; i < 10; i++) {
            WorkThread workThread = new WorkThread(i+"",semaphoreService);
            workThread.start();
        }
    }

    public String getFormatDate() {
      return sdf.format(new Date());
    }
   // 工作线程类 - 静态内部类
   static  class WorkThread extends Thread {

        SemaphoreService semaphoreService;
       //构造参数传入线程名称及SemaphoreService
        public WorkThread(String name, SemaphoreService semaphoreService) {
            this.semaphoreService = semaphoreService;
            setName(name);
        }
        @Override
        public void run() {
            try {
                semaphoreService.doSomeThing();
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
    }
}

1.线程最多4个同时执行

//1,设置信号量数为4，执行结果如下图
private Semaphore semaphore = new Semaphore(4);

2.线程依次执行，实现同步

//1,设置信号量数为1，执行结果如下图
private Semaphore semaphore = new Semaphore(1);

三、计数器CountDownLatch

3.1、主要方法

countDown() 计数器减1
await() 阻塞等待，直到计数器为0唤醒继续执行
await(long timeout, TimeUnit unit) 阻塞等待，在等待设定时间计时器还没到减为0，也不会再继续等待了。

3.2、示例代码1

package countDownLatch;

import javax.swing.plaf.IconUIResource;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

/**
 * @author zdd
 * 2019/11/29 7:19 下午
 * Description: 班长召集士兵集合拉练案例
 */
public class CountDownLatchSoldierTest {

    private static final Integer THREAD_COUNT = 10;
    //等待士兵10人
    private static  CountDownLatch countDownLatch = new CountDownLatch(10);

    public static void main(String[] args) throws InterruptedException {
        //1,开10个线程模拟士兵签到,采用线程池创建
        ExecutorService executorService = Executors.newFixedThreadPool(THREAD_COUNT);
        for (int i = 0; i < 10; i++) {
            //2，提交10个线程执行
            executorService.submit(()->{
                try{
                    System.out.println(Thread.currentThread().getName()+ ": 士兵报告");
                }finally {
                    //3,计数器减1
                    countDownLatch.countDown();
                }
            });
        }
        //3,主线程做班长，负责等待所有士兵到齐，开始拉练
        countDownLatch.await();
        System.out.println("班长： 集合完毕，开始5公里越野！ ");

        //4，关闭线程池
        executorService.shutdown();
    }
}

3.3、示例代码2

计算器未减到0，主线程持续等待。

package countDownLatch;

import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;

/**
 * zdd
 * 2019/11/5 10:50 
 * Description:测试计数器使用
 */
public class CountDownLatchTest {

    static  int number = 2;
    //1，设置计数器的值为 2
    static CountDownLatch countDownLatch = new CountDownLatch(number);


    public static void  main(String[] args) throws InterruptedException {
        //2，开启一个线程，传入计数器
        new Thread(()-> {
             try {
                //睡2s，模拟执行业务
                TimeUnit.SECONDS.sleep(2);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }finally {
                System.out.println("调用countDown前， 当前count 值为："+countDownLatch.getCount());
                //3，调用递减 1次
                countDownLatch.countDown();
                System.out.println("调用countDown后， 当前count 值为："+countDownLatch.getCount());
            }
        },"worker").start();

        //情况1、主线程开始执行 ,等待1s之后，如果还没到条件，也不等了哦
       // countDownLatch.await(1,TimeUnit.SECONDS);
       //情况2、阻塞等待，如果计数未到0,一直阻塞等待
        countDownLatch.await();
        System.out.println("主线程继续执行");
    }
}

主线程阻塞等待，执行结果如下：

四、栅栏类CyclicBarrier

4.1、构造方法

CyclicBarrier(int parties) //parties为需要等待线程的数量
//barrierAction,在所有其他线程到达后，优先执行的线程。可根据业务添加
CyclicBarrier(int parties, Runnable barrierAction) 

4.2、示例代码

package CyclicBarrier;

import java.util.Map;
import java.util.concurrent.*;
/**
 * @author zdd
 * 2019/11/28 9:20
 * Description: 开启4个线程分别统计sheet表，使用栅栏类实现同步统计计算，最后计算总和；此处也可用  
 * CountDownLatch实现
 */
public class BankAccountService {
    private ConcurrentHashMap<String,Integer> concurrentHashMap = new ConcurrentHashMap<>();
  
    // 线程数
    private final static int THREAD_COUNT = 4;
    private CyclicBarrier cyclicBarrier = new CyclicBarrier(4, new Runnable() {
        @Override
        public void run() {
            Integer sumcount = 0;
            //汇总每个sheet计算结果
            for (Map.Entry<String, Integer> entry: concurrentHashMap.entrySet()) {
                sumcount+=entry.getValue();
            }
            System.out.println("优先执行,  求和计算完毕，总和为："+ sumcount);
        }
    });

    public void  count() {
        ExecutorService executor = Executors.newFixedThreadPool(4);

        for (int i = 0; i < THREAD_COUNT; i++) {
            executor.execute(new Runnable() {
                @Override
                public void run() {
                    System.out.println(Thread.currentThread().getName()+"到达屏障！");
                    concurrentHashMap.put(Thread.currentThread().getName(),1);

                    try {
                        cyclicBarrier.await();
                        System.out.println(Thread.currentThread().getName()+"继续执行！");
                    } catch (Exception e) {
                        e.printStackTrace();
                    }
                }
            });
        }
        executor.shutdown();
    }

    public static void main(String[] args) {
        BankAccountService bankAccountService = new BankAccountService();
        bankAccountService.count();
    }
}

运行结果如下：

// 将线程数改为2,阻塞等待其他线程
 private final static int THREAD_COUNT = 2;

---

参考资料：
Java并发编程的艺术-方腾飞