2、线程的并发工具类

作者: 咸鱼Jay | 来源:发表于2022-03-01 15:44 被阅读0次

Fork-Join

java下多线程的开发可以我们自己启用多线程，线程池，还可以使用forkjoin， forkjoin 可以让我们不去了解诸如 Thread,Runnable 等相关的知识，只要遵循 forkjoin 的开发模式，就可以写出很好的多线程并发程序

分而治之

同时 forkjoin 在处理某一类问题时非常的有用，哪一类问题？分而治之的问题。十大计算机经典算法：快速排序、堆排序、归并排序、二分查找、线性查找、深度优先、广度优先、Dijkstra、动态规划、朴素贝叶斯分类，有几个属于分而治之？3 个，快速排序、归并排序、二分查找，还有大数据中 M/R 都是。

分治法的设计思想是：将一个难以直接解决的大问题，分割成一些规模较小的相同问题，以便各个击破，分而治之。

分治策略是：对于一个规模为n的问题，若该问题可以容易地解决（比如说规模n较小）则直接解决，否则将其分解为k个规模较小的子问题，这些子问题互相独立且与原问题形式相同(子问题相互之间有联系就会变为动态规范算法)，递归地解这些子问题，然后将各子问题的解合并得到原问题的解。这种算法设计策略叫做分治法。

归并排序

归并排序是建立在归并操作上的一种有效的排序算法。该算法是采用分治法的一个非常典型的应用。将已有序的子序列合并，得到完全有序的序列；即先使每个子序列有序，再使子序列段间有序。

若将两个有序表合并成一个有序表，称为 2-路归并，与之对应的还有多路归并。

对于给定的一组数据，利用递归与分治技术将数据序列划分成为越来越小的半子表，在对半子表排序后，再用递归方法将排好序的半子表合并成为越来越大的有序序列。

为了提升性能，有时我们在半子表的个数小于某个数（比如 15）的情况下，对半子表的排序采用其他排序算法，比如插入排序。

归并排序（降序）示例

先將数组划分为左右两个子表
然后继续左右两个子表拆分
对最后的拆分的子表，两两进行排序

对有序的子表进行排序和比较合并
对合并后的子表继续比较合并

Fork-Join 原理

Fork/Join框架：就是在必要的情况下，将一个大任务，进行拆分(fork)成若干个小任务(拆到不可再拆时)，再将一个个的小任务运算的结果进行join汇总。

工作密取

即当前线程的Task已经全被执行完毕，则自动取到其他线程的 Task 池中取出 Task 继续执行。

ForkJoinPool 中维护着多个线程（一般为CPU核数）在不断地执行Task，每个线程除了执行自己职务内的Task之外，还会根据自己工作线程的闲置情况去获取其他繁忙的工作线程的 Task，如此一来就能能够减少线程阻塞或是闲置的时间，提高 CPU 利用率。

Fork/Join 实战

Fork/Join 使用的标准范式

我们要使用ForkJoin框架，必须首先创建一个ForkJoin任务。它提供在任务中执fork和join的操作机制，通常我们不直接继承ForkjoinTask类，只需要直接继承其子类。

RecursiveAction，用于没有返回结果的任务
RecursiveTask，用于有返回值的任务

task要通过ForkJoinPool来执行，使用submit或invoke提交，两者的区别是：invoke是同步执行，调用之后需要等待任务完成，才能执行后面的代码；submit是异步执行。

join()和get方法当任务完成的时候返回计算结果。

在我们自己实现的compute方法里，首先需要判断任务是否足够小，如果足够小就直接执行任务。如果不足够小，就必须分割成两个子任务，每个子任务在调用invokeAll方法时，又会进入compute方法，看看当前子任务是否需要继续分割成孙任务，如果不需要继续分割，则执行当前子任务并返回结果。使用join方法会等待子任务执行完并得到其结果。

Fork/Join 的同步用法和异步用法

1. Fork/Join的同步用法同时演示返回结果值：统计整形数组中所有元素的和

public class MakeArray {
    //数组长度
    public static final int ARRAY_LENGTH  = 4000;

    public static int[] makeArray() {

        //new一个随机数发生器
        Random r = new Random();
        int[] result = new int[ARRAY_LENGTH];
        for(int i=0;i<ARRAY_LENGTH;i++){
            //用随机数填充数组
            result[i] =  r.nextInt(ARRAY_LENGTH*3);
        }
        return result;

    }
}

定义一个4000长度数组的类MakeArray，数组里面的数是随机填充的。

下面先使用单线程进行累加：

public class SumNormal {
    
    public static void main(String[] args) {
        int count = 0;
        int[] src = MakeArray.makeArray();

        long start = System.currentTimeMillis();
        for(int i= 0;i<src.length;i++){
//          SleepTools.ms(1);// ①
            count = count + src[i];
        }
        System.out.println("The count is "+count +" spend time:"+(System.currentTimeMillis()-start)+"ms");      
    }
}

运行结果：

The count is 23492056 spend time:0ms

下面使用Fork/Join来实现

public class SumArray {
    private static class SumTask extends RecursiveTask<Integer>{

        private final static int THRESHOLD = MakeArray.ARRAY_LENGTH/10;
        private int[] src;
        private int fromIndex;
        private int toIndex;

        public SumTask(int[] src, int fromIndex, int toIndex) {
            this.src = src;
            this.fromIndex = fromIndex;
            this.toIndex = toIndex;
        }

        @Override
        protected Integer compute() {
            /*任务的大小是否合适*/
            if (toIndex - fromIndex < THRESHOLD){
//                System.out.println(" from index = "+fromIndex+" toIndex="+toIndex);
                int count = 0;
                for(int i= fromIndex;i<=toIndex;i++){
                    //SleepTools.ms(1);// ②
                     count = count + src[i];
                }
                return count;
            }else{
                //fromIndex....mid.....toIndex
                int mid = (fromIndex+toIndex)/2;
                SumTask left = new SumTask(src,fromIndex,mid);
                SumTask right = new SumTask(src,mid+1,toIndex);
                invokeAll(left,right);
                return left.join()+right.join();
            }
        }
    }

    public static void main(String[] args) {
        int[] src = MakeArray.makeArray();
        /*new出池的实例*/
        ForkJoinPool pool = new ForkJoinPool();
        /*new出Task的实例*/
        SumTask innerFind = new SumTask(src,0,src.length-1);

        long start = System.currentTimeMillis();

        pool.invoke(innerFind);// 同步执行
        System.out.println("Task is Running.....");

        System.out.println("The count is "+innerFind.join() +" spend time:"+(System.currentTimeMillis()-start)+"ms");
    }
}

运行结果：

Task is Running.....
The count is 24428568 spend time:2067ms

从上面看到单线程运行和Fork/Join运行的结果看，好像Fork/Join不如单线程运行的，这是因为Fork/Join的compute()方法其实就是递归调用，递归调用就会有方法的出栈和入栈，还有一个就是这个是多线程运行，就会有上下文切换问题，所以这里Fork/Join比单线程慢。

现在打开①和②处的代码，在运行

// 单线程运行
The count is 24071071 spend time:7368ms
// Fork/Join运行
Task is Running.....
The count is 24050929 spend time:2375ms

2. Fork/Join的异步用法同时演示不要求返回值：遍历指定目录（含子目录）寻找指定类型文件

public class FindDirsFiles extends RecursiveAction {

    private File path;

    public FindDirsFiles(File path) {
        this.path = path;
    }

    @Override
    protected void compute() {
        List<FindDirsFiles> subTasks = new ArrayList<FindDirsFiles>();

        File[] files = path.listFiles();
        if (files!=null){
            for (File file : files) {
                if (file.isDirectory()) {
                    // 对每个子目录都新建一个子任务。
                    subTasks.add(new FindDirsFiles(file));
                } else {
                    // 遇到文件，检查。
                    if (file.getAbsolutePath().endsWith("txt")){
                        System.out.println("文件:" + file.getAbsolutePath());
                    }
                }
            }
            if (!subTasks.isEmpty()) {
                // 在当前的 ForkJoinPool 上调度所有的子任务。
                for (FindDirsFiles subTask : invokeAll(subTasks)) {
                    subTask.join();
                }
            }
        }
    }

    public static void main(String [] args){
        try {
            // 用一个 ForkJoinPool 实例调度总任务
            ForkJoinPool pool = new ForkJoinPool();
            FindDirsFiles task = new FindDirsFiles(new File("F:/"));

            pool.execute(task);// 异步提交
            
            /*主线程做自己的业务工作*/
            System.out.println("Task is Running......");
            Thread.sleep(1);
            int otherWork = 0;
            for(int i=0;i<100;i++){
                otherWork = otherWork+i;
            }
            System.out.println("Main Thread done sth......,otherWork="+otherWork);
            task.join();// 阻塞方法
            System.out.println("Task end");
        } catch (Exception e) {
            e.printStackTrace();
        }
    }
}

CountDownLatch

闭锁，CountDownLatch这个类能够使一个线程等待其他线程完成各自的工作后再执行。例如，应用程序的主线程希望在负责启动框架服务的线程已经启动所有的框架服务之后再执行。

CountDownLatch是通过一个计数器来实现的，计数器的初始值为初始任务的数量。每当完成了一个任务后，计数器的值就会减1（CountDownLatch.countDown()方法）。当计数器值到达0时，它表示所有的已经完成了任务，然后在闭锁上等待CountDownLatch.await()方法的线程就可以恢复执行任务。

应用场景：

实现最大的并行性：有时我们想同时启动多个线程，实现最大程度的并行性。例如，我们想测试一个单例类。如果我们创建一个初始计数为1的CountDownLatch，并让所有线程都在这个锁上等待，那么我们可以很轻松地完成测试。我们只需调用一次countDown()方法就可以让所有的等待线程同时恢复执行。

开始执行前等待n个线程完成各自任务：例如应用程序启动类要确保在处理用户请求前，所有N个外部系统已经启动和运行了，例如处理excel中多个表单。

public class UseCountDownLatch {
    
    static CountDownLatch latch = new CountDownLatch(6);

    /*初始化线程*/
    private static class InitThread implements Runnable{

        public void run() {
            System.out.println("Thread_"+Thread.currentThread().getId() +" ready init work......");
            latch.countDown();
            for(int i =0;i<2;i++) {
                System.out.println("Thread_"+Thread.currentThread().getId() +" ........continue do its work");
            }
        }
    }

    /*业务线程等待latch的计数器为0完成*/
    private static class BusiThread implements Runnable{

        public void run() {
            try {
                latch.await();
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            for(int i =0;i<3;i++) {
                System.out.println("BusiThread_"+Thread.currentThread().getId() +" do business-----");
            }
        }
    }

    public static void main(String[] args) throws InterruptedException {
        new Thread(new Runnable() {
            public void run() {
                SleepTools.ms(1);
                System.out.println("Thread_"+Thread.currentThread().getId() +" ready init work step 1st......");
                latch.countDown();
                System.out.println("begin step 2nd.......");
                SleepTools.ms(1);
                System.out.println("Thread_"+Thread.currentThread().getId() +" ready init work step 2nd......");
                latch.countDown();
            }
        }).start();
        new Thread(new BusiThread()).start();
        for(int i=0;i<=3;i++){
            Thread thread = new Thread(new InitThread());
            thread.start();
        }

        latch.await();
        System.out.println("Main do ites work........");
    }
}

CyclicBarrier

CyclicBarrier 是什么？

从字面上的意思可以知道，这个类的中文意思是“循环栅栏”。大概的意思就是一个可循环利用的屏障。

它的作用就是会让所有线程都等待完成后才会继续下一步行动。

举个例子，就像生活中我们会约朋友们到某个餐厅一起吃饭，有些朋友可能会早到，有些朋友可能会晚到，但是这个餐厅规定必须等到所有人到齐之后才会让我们进去。这里的朋友们就是各个线程，餐厅就是CyclicBarrier。

构造方法

public CyclicBarrier(int parties)
public CyclicBarrier(int parties, Runnable barrierAction)

parties是参与线程的个数
第二个构造方法有一个Runnable参数，这个参数的意思是当所有的线程到达那个屏障后，先执行这个barrierAction，最后在执行所有线程。
最后一个到达线程要做的任务

重要方法

public int await() throws InterruptedException, BrokenBarrierException
public int await(long timeout, TimeUnit unit) throws InterruptedException, BrokenBarrierException, TimeoutException

线程调用await()表示自己已经到达栅栏
BrokenBarrierException表示栅栏已经被破坏，破坏的原因可能是其中一个线程await()时被中断或者超时

基本使用

无`barrierAction`的用法：

/**
 *共4个子线程，他们全部完成工作后，交出自己结果，再被统一释放去做自己的事情
 */
public class UseCyclicBarrier {
    private static CyclicBarrier barrier = new CyclicBarrier(4);

    public static void main(String[] args) {
        for(int i=0;i<4;i++){
            Thread thread = new Thread(new SubThread());
            thread.start();
        }
    }

    /**相互等待的子线程*/
    private static class SubThread implements Runnable{
        @Override
        public void run() {
            long id = Thread.currentThread().getId();
            try {
                Thread.sleep(1000+id);
                System.out.println("Thread_"+id+" ....do something ");
                barrier.await();
                Thread.sleep(1000+id);
                System.out.println("Thread_"+id+" ....do its business ");
            } catch (Exception e) {
                e.printStackTrace();
            }
        }
    }
}

Thread_20 ....do something 
Thread_21 ....do something 
Thread_22 ....do something 
Thread_23 ....do something 
Thread_20 ....do its business 
Thread_21 ....do its business 
Thread_22 ....do its business 
Thread_23 ....do its business

有`barrierAction`的用法

/**
 *共4个子线程，他们全部完成工作后，交出自己结果，
 *再被统一释放去做自己的事情，而交出的结果被另外的线程拿来拼接字符串
 */
public class UseCyclicBarrier {
    private static CyclicBarrier barrier = new CyclicBarrier(4,new CollectThread());
    private static ConcurrentHashMap<String,Long> resultMap = new ConcurrentHashMap<>();//存放子线程工作结果的容器

    public static void main(String[] args) {
        for(int i=0;i<4;i++){
            Thread thread = new Thread(new SubThread());
            thread.start();
        }
    }

    /**汇总任务*/
    private static class CollectThread implements Runnable{
        @Override
        public void run() {
            StringBuilder result = new StringBuilder();
            for(Map.Entry<String,Long> workResult:resultMap.entrySet()){
                result.append("["+workResult.getValue()+"]");
            }
            System.out.println(" the result = "+ result);
            System.out.println("do other business........");
        }
    }

    /**相互等待的子线程*/
    private static class SubThread implements Runnable{
        @Override
        public void run() {
            long id = Thread.currentThread().getId();
            resultMap.put(Thread.currentThread().getId()+"",id);
            try {
                Thread.sleep(1000+id);
                System.out.println("Thread_"+id+" ....do something ");
                barrier.await();
                Thread.sleep(1000+id);
                System.out.println("Thread_"+id+" ....do its business ");
            } catch (Exception e) {
                e.printStackTrace();
            }
        }
    }
}

Thread_20 ....do something 
Thread_21 ....do something 
Thread_22 ....do something 
Thread_23 ....do something 
 the result = [22][23][20][21]
do other business........
Thread_20 ....do its business 
Thread_21 ....do its business 
Thread_22 ....do its business 
Thread_23 ....do its business

CyclicBarrier是可循环的，那么他的可循环是怎么体现呢？

上面示例中的barrier.await();方法是可以反复调用,在上面SubThread类里在添加一个barrier.await();方法，可以看到打印结果就知道循环的意义了。

private static class SubThread implements Runnable{
    @Override
    public void run() {
        long id = Thread.currentThread().getId();
        resultMap.put(Thread.currentThread().getId()+"",id);
        try {
            Thread.sleep(1000+id);
            System.out.println("Thread_"+id+" ....do something ");
            barrier.await();
            Thread.sleep(1000+id);
            System.out.println("Thread_"+id+" ....do its business ");
            barrier.await();
        } catch (Exception e) {
            e.printStackTrace();
        }
    }
}

Thread_21 ....do something 
Thread_22 ....do something 
Thread_23 ....do something 
Thread_20 ....do something 
 the result = [22][23][20][21]
do other business........
Thread_21 ....do its business 
Thread_22 ....do its business 
Thread_23 ....do its business 
Thread_20 ....do its business 
 the result = [22][23][20][21]
do other business........

CyclicBarrier 使用场景

可以用于多线程计算数据，最后合并计算结果的场景。

CyclicBarrier 与 CountDownLatch 区别

CountDownLatch是一次性的，CyclicBarrier是可循环利用的
在协调线程之间同时运行这件事上面，CountDownLatch的工作线程同时执行是由外面的线程协调执行的，而CyclicBarrier是由工作线程本身相互协调执行的。
构造函数上：CountDownLatch的构造函数的参数count和线程数没有关系，而CyclicBarrier的构造函数的参数parties跟线程数密切相关的。
CountDownLatch的几个工作线程同时跑的时候，是不能执行其他的计算结果的，而CyclicBarrier是可以的，只需要传入barrierAction就可以让等待的线程做完自己的工作进行所谓的汇总

Semaphore

Semaphore 是什么？

Semaphore字面意思是信号量的意思，它的作用是控制访问特定资源的线程数目。

怎么使用 Semaphore？

构造方法

public Semaphore(int permits)
public Semaphore(int permits, boolean fair)

permits表示许可线程的数量
fair表示公平性，如果这个设为true的话，下次执行的线程会是等待最久的线程

重要方法

public void acquire() throws InterruptedException
public void release()

acquire()表示阻塞并获取许可
release()表示释放许可

基本使用

多个线程同时执行，但是限制同时执行的线程数量为2个。

public class SemaphoreDemo {

    static class TaskThread extends Thread {

        Semaphore semaphore;

        public TaskThread(Semaphore semaphore) {
            this.semaphore = semaphore;
        }

        @Override
        public void run() {
            try {
                semaphore.acquire();
                System.out.println(getName() + " acquire");
                Thread.sleep(1000);
                semaphore.release();
                System.out.println(getName() + " release ");
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
    }

    public static void main(String[] args) {
        int threadNum = 5;
        Semaphore semaphore = new Semaphore(2);
        for (int i = 0; i < threadNum; i++) {
            new TaskThread(semaphore).start();
        }
    }

}

打印结果：

Thread-1 acquire
Thread-0 acquire
Thread-0 release 
Thread-1 release 
Thread-2 acquire
Thread-3 acquire
Thread-2 release 
Thread-4 acquire
Thread-3 release 
Thread-4 release

从打印结果可以看出，一次只有两个线程执行acquire()，只有线程进行release()方法后才会有别的线程执行acquire()。

需要注意的是Semaphore只是对资源并发访问的线程数进行监控，并不会保证线程安全。

Semaphore 使用场景

可用于流量控制，限制最大的并发访问数。

Callable、Future和FutureTask

之前介绍过创建线程有两种方式:1.继承Thread类，2.实现Runnable接口，这两种都是没有返回值的。

public class UseFuture {

    /*实现Callable接口，允许有返回值*/
    private static class UseCallable implements Callable<Integer> {
        private int sum;

        @Override
        public Integer call() throws Exception {
            System.out.println("Callable子线程开始计算！");
            Thread.sleep(2000);
            for (int i = 0; i < 5000; i++) {
                sum = sum + i;
            }
            System.out.println("Callable子线程计算结束！结果为: " + sum);
            return sum;
        }
    }

    public static void main(String[] args)
            throws InterruptedException, ExecutionException {

        UseCallable useCallable = new UseCallable();
        //包装
        FutureTask<Integer> futureTask = new FutureTask<>(useCallable);
        new Thread(futureTask).start();
        System.out.println(futureTask.get());
    }
}

Callable子线程开始计算！
Callable子线程计算结束！结果为: 12497500
12497500

2、线程的并发工具类

Fork-Join

分而治之

归并排序

归并排序（降序）示例

Fork-Join 原理

工作密取

Fork/Join 实战

Fork/Join 使用的标准范式

Fork/Join 的同步用法和异步用法

CountDownLatch

CyclicBarrier

CyclicBarrier 是什么？

构造方法

重要方法

基本使用

无`barrierAction`的用法：

有`barrierAction`的用法

CyclicBarrier 使用场景

CyclicBarrier 与 CountDownLatch 区别

Semaphore

Semaphore 是什么？

怎么使用 Semaphore？

构造方法

重要方法

基本使用

Semaphore 使用场景

Callable、Future和FutureTask

相关文章

网友评论

延伸阅读

深度阅读

栏目导航

热点阅读

2、线程的并发工具类

Fork-Join

分而治之

归并排序

归并排序（降序）示例

Fork-Join 原理

工作密取

Fork/Join 实战

Fork/Join 使用的标准范式

Fork/Join 的同步用法和异步用法

CountDownLatch

CyclicBarrier

CyclicBarrier 是什么？

构造方法

重要方法

基本使用

无barrierAction的用法：

有barrierAction的用法

CyclicBarrier 使用场景

CyclicBarrier 与 CountDownLatch 区别

Semaphore

Semaphore 是什么？

怎么使用 Semaphore？

构造方法

重要方法

基本使用

Semaphore 使用场景

Callable、Future和FutureTask

相关文章

网友评论

延伸阅读

深度阅读

栏目导航

热点阅读

无`barrierAction`的用法：

有`barrierAction`的用法