Java多线程06——JUC并发包02

太过爱你忘了你带给我的痛 2024-03-27 15:57 146阅读 0赞

## 1 线程的同步工具类 `CountDownLatch` ##

`CountDownLatch` 同步辅助类，在完成一组正在其他线程中执行的操作之前，它允许一个或多个线程一直等待。

*  `CountDownLatch` 类是一个同步计数器，构造时传入int参数，该参数就是计数器的初始值，每调用一次 `countDown()` 方法，计数器减 1，当计数器大于0时，`await()` 方法会阻塞当前线程继续执行。
 *  由于调用了 `countDown()` 方法，所以在当前计数到达零之前，`await()` 方法会一直受阻塞。之后，会释放所有处于等待的线程，`await()` 方法之后的所有后续调用都将立即返回，这种现象只出现一次，计数无法被重置。一个线程或者多个，等待另外N个线程完成某个事情之后才能执行。

**创建线程类**

import java.util.concurrent.CountDownLatch;
    
    public class UserThread1 extends Thread {
        private int sum1 = 0;
        private CountDownLatch cd;
    
        public UserThread1(CountDownLatch cd){
            this.cd = cd;
        }
    
        @Override
        public void run() {
            for(int i=0;i<=50;i++){
                sum1 += i;
            }
            cd.countDown();
        }
    
        public int getSum1(){
            return sum1;
        }
    }

import java.util.concurrent.CountDownLatch;
    
    public class UserThread2 extends Thread {
        private int sum2 = 0;
        private CountDownLatch cd;
    
        public UserThread2(CountDownLatch cd){
            this.cd = cd;
        }
    
        @Override
        public void run() {
            for(int i=51;i<=100;i++){
                sum2 += i;
            }
            cd.countDown();
        }
    
        public int getSum2(){
            return sum2;
        }
    }

线程类中使用了 `countDown()` 方法，用于更新计数器。

**创建测试类**

import java.util.concurrent.CountDownLatch;
    
    public class Test {
        public static void main(String[] args) {
            CountDownLatch cd = new CountDownLatch(2);
            UserThread1 thread1 = new UserThread1(cd);
            UserThread2 thread2 = new UserThread2(cd);
            thread1.start();
            thread2.start();
    
            try {
                cd.await();
                int sum = thread1.getSum1() + thread2.getSum2();
                System.out.println("1~100 的和是：" + sum);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
    }

输出结果为：

> 1~100 的和是：5050

在测试类中调用了 await() 方法，用于阻塞当前线程，直到 CountDownLatch 初始化时的计数器变为0为止。

![6a9af2a5cf319f137316545b9ccaa7dd.png][]

**注意：**

*  CountDownLatch 的初始值，应当与所有线程当中的 countDown() 数量相等，
 *  如果上述测试类初始化为1，初始值小于 countDown() 数量，即 `CountDownLatch cd = new CountDownLatch(1);`运行程序后，输出结果可能为

> 1~100 的和是：1275

*  如果上述测试类初始化为3，初始值大于 countDown() 数量，即 `CountDownLatch cd = new CountDownLatch(3);`运行程序后，程序将被挂起，无法退出阻塞状态。

## 2 线程的同步工具类 `CyclicBarrier` ##

`CyclicBarrier` 是一个同步辅助类，允许一组线程互相等待，直到到达某个公共屏障点（common barrier point）。类似于集合点。

![8cc37af530f74f7098ea37df616d827f.png][]

因为该barrier在释放等待线程后可以重用，所以称它为循环的 barrier。

### 2.1 创建被线程调用的资源类 ###

public class TimeCount {
        private int count1;
        private int count2;
        private int sum;
    
        public int getCount1() {
            return count1;
        }
    
        public void setCount1(int count1) {
            this.count1 = count1;
        }
    
        public int getCount2() {
            return count2;
        }
    
        public void setCount2(int count2) {
            this.count2 = count2;
        }
    
        public int getSum() {
            return this.count1 + this.count2;
        }
    
        public void setSum(int sum) {
            this.sum = sum;
        }
    }

### 2.2 创建线程类 ###

在任务执行完成后，使用 `cyclicBarrier.await();` 添加阻塞点。

import java.util.concurrent.CyclicBarrier;
    
    public class UserRunn implements Runnable{
        private TimeCount tc;
        private String name;
        private CyclicBarrier cyclicBarrier;
    
        public UserRunn(TimeCount tc, String name, CyclicBarrier cyclicBarrier){
            this.tc = tc;
            this.name = name;
            this.cyclicBarrier = cyclicBarrier;
        }
    
        @Override
        public void run() {
            if(name.equals("看书")){
                try {
                    Thread.sleep(4000);
                    tc.setCount1(4000);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
            else if(name.equals("写字")){
                try {
                    Thread.sleep(2000);
                    tc.setCount2(2000);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
    
            //阻塞点，等所有线程运行完毕，自动解锁
            try {
                cyclicBarrier.await();
                System.out.println("------------" + name + " end-----------");
            } catch (Exception e) {
                e.printStackTrace();
            }
    
        }
    }

### 2.3 创建测试类 ###

对 CyclicBarrier 进行初始化，因为我们需要两个线程同时到达阻塞点，故初始化中 `new CyclicBarrier(2, new Runnable())` 第一个参数 `parties` 设置为 2。

import java.util.concurrent.CyclicBarrier;
    
    public class Test {
        public static void main(String[] args) {
            TimeCount tc = new TimeCount();
    
            CyclicBarrier cyclicBarrier = new CyclicBarrier(2, new Runnable() {
                @Override
                public void run() {
                    int sum = tc.getSum();
                    System.out.println("功能总耗时：" + sum);
                }
            });
            new Thread(new UserRunn(tc, "看书", cyclicBarrier)).start();
            new Thread(new UserRunn(tc, "写字", cyclicBarrier)).start();
        }
    }

#### 2.4 输出结果 ####

> 功能总耗时：6000
> 
> \------------看书 end-----------
> 
> \------------写字 end-----------

“看书”线程需要执行4秒，而“写字”线程需要执行2秒，在到达阻塞点时，执行耗时短的线程会等待执行耗时长的线程，当到达阻塞点的线程数和 2.3 部分创建的 CyclicBarrier 对象初始化的 parties 数值时，所有线程被同时释放。

![88f4c179b4d2d1b2e331ae31d1c0df64.png][]

**注意：**

创建的 CyclicBarrier 对象初始化的 parties 数值，应与我们想要实现到达屏障点的线程数相同。

如果 parties 小于期望值，达不到汇聚的效果；

如果 parties 大于期望值，线程将一直处于阻塞状态。

## 3 线程的同步工具类 `Semaphore` ##

`Semaphore` 是一个计数信号量，它的本质是一个共享锁，是基于AQS实现的，通过state变量来实现共享。  
通过调用 acquire 方法，对 state 值减去一，当调用 release 的时候，对 state 值加一。  
当 state 变量小于0时，在AQS队列中阻塞等待。

### 3.1 创建餐厅类 ###

**声明对象**

在餐厅类中，声明 `Semaphore` 信号量对象，及最大可接受的线程数 `new Semaphore(num)`。

**加锁**

在需要进行限流的操作前加锁 `acquire();`，当进入该位置的线程数达到最大可接受值时，其他线程将进入等待状态。

**释放锁**

在限流操作结束后释放锁 `release();`，后续线程才会进入到 `acquire();`语句中。

import java.util.concurrent.Semaphore;
    
    public class Restaurant {
        //通过引入Semaphore，实现餐厅限流
        private final Semaphore semaphore;
    
        public Restaurant(int num){
            //设置最大可用的并行信号量
            this.semaphore = new Semaphore(num);
        }
    
        public void eatConsumers(){
            try {
                //加锁
                semaphore.acquire();
    
                System.out.println(Thread.currentThread().getName() + "进入餐厅");
    
                Thread.sleep(3000);
                System.out.println(Thread.currentThread().getName() + "离开餐厅");
                //释放锁
                semaphore.release();
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
    }

### 3.2 创建消费者类 ###

public class Consumer extends Thread {
        private Restaurant restaurant;
    
        public Consumer(Restaurant restaurant){
            this.restaurant = restaurant;
        }
    
        @Override
        public void run() {
            restaurant.eatConsumers();
        }
    }

#### 3.3 创建测试类 ####

设置限流的最大线程数为2

public class Test {
        public static void main(String[] args) {
            Restaurant restaurant = new Restaurant(2);
    
            for(int i=0;i<5;i++) {
                new Consumer(restaurant).start();
            }
        }
    }

#### 3.4 输出结果 ####

> Thread-1进入餐厅
> 
> Thread-4进入餐厅
> 
> Thread-1离开餐厅
> 
> Thread-4离开餐厅
> 
> Thread-2进入餐厅
> 
> Thread-3进入餐厅
> 
> Thread-2离开餐厅
> 
> Thread-0进入餐厅
> 
> Thread-3离开餐厅
> 
> Thread-0离开餐厅

![923ab1ce7d36e5a77902dc312227c851.png][]

## 4 线程的交换类 `Exchanger` ##

Exchanger（交换者）是一个用于线程间协作的工具类，Exchanger 用于进行线程间的数据交换。  
它提供一个同步点，在这个同步点，两个线程可以交换彼此的数据。  
这两个线程通过 exchange() 方法交换数据。

如果第一个线程先执行 exchange() 方法，它会一直等待第二个线程也执行 exchange() 方法，当两个线程都到达同步点时，这两个线程就可以交换数据，将本线程产生出来的数据传递给对方。

import java.util.concurrent.Exchanger;
    import java.util.concurrent.ExecutorService;
    import java.util.concurrent.Executors;
    
    public class Test {
        private static final Exchanger<String> exchanger = new Exchanger<>();
        private static ExecutorService executorService = Executors.newFixedThreadPool(2);
    
        public static void main(String[] args) {
            executorService.execute(new Runnable() {
                @Override
                public void run() {
                    String a = " A 银行转入";
                    System.out.println(Thread.currentThread().getName() + a);
                    try {
                        String b = exchanger.exchange(a);
                        System.out.println(Thread.currentThread().getName() + b);
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                }
            });
    
            executorService.execute(new Runnable() {
                @Override
                public void run() {
                    String a = " B 银行转出";
                    System.out.println(Thread.currentThread().getName() + a);
                    try {
                        String b = exchanger.exchange(a);
                        System.out.println(Thread.currentThread().getName() + b);
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                }
            });
    
            executorService.shutdown();
        }
    }

**执行输出**

> pool-1-thread-1 A 银行转入
> 
> pool-1-thread-2 B 银行转出
> 
> pool-1-thread-2 A 银行转入
> 
> pool-1-thread-1 B 银行转出

**需要注意的是：**

如果参与数据交换的线程数量为**偶数**，可以实现两两数据交换；

如果参与数据交换的线程数量为**奇数**，程序将一直处于阻塞状态。

## 5 线程的 `Fork`\-`Join` 机制 ##

Fork/Join 框架是 JAVA7 提供的一个用于并行执行任务的框架，  
是一个把大任务分割成若干个小任务，最终汇总每个小任务结果后得到大任务结果的框架。

**分治法：**

> 把一个规模大的问题划分为规模较小的子问题，然后分而治之，最后合并子问题的解，得到原问题的解。

import java.util.concurrent.ExecutionException;
    import java.util.concurrent.ForkJoinPool;
    import java.util.concurrent.ForkJoinTask;
    import java.util.concurrent.RecursiveTask;
    
    public class ContTask extends RecursiveTask<Integer> {
        private int start;
        private int end;
    
        //计算任务量的阈值，最小任务量范围
        private static final int TASKSIZE = 30;
        private static int count = 0;
    
        public ContTask(int start, int end){
            this.start = start;
            this.end = end;
        }
    
        //重写compute方法，任务执行的主要计算
        @Override
        protected Integer compute() {
            int sum = 0;
    
            System.out.println("开启线程进行计算" + count++);
            boolean state = (end - start) <= TASKSIZE;
            //如果小于等于任务的阈值
            if(state){
                //无需拆分任务计算
                for(int i=start;i<=end;i++){
                    sum += i;
                }
            }else{
                //进行拆分任务计算
                System.out.printf("start-%d, end-%d这个任务需要进行拆分任务计算。。。%s%n", start, end, Thread.currentThread().getName());
                //分割成两个任务
                int middle = (end + start) / 2;
                ContTask contTask1 = new ContTask(start, middle);
                ContTask contTask2 = new ContTask(middle+1, end);
                //开启线程计算分布式任务
                invokeAll(contTask1, contTask2);
                //阻塞，直到任务完成或取消
                Integer tasksum1 = contTask1.join();
                Integer tasksum2 = contTask2.join();
                //结果合并
                sum = tasksum1 + tasksum2;
            }
    
            return sum;
        }
    
        public static void main(String[] args) {
            //分布式计算池
            ForkJoinPool forkJoinPool = new ForkJoinPool();
            //初始化设置任务
            ContTask contTask = new ContTask(1, 100);
            //分布式计算任务，提交任务
            ForkJoinTask forkJoinTask = forkJoinPool.submit(contTask);
            //得到最终计算结果
            try {
                System.out.println("最终计算结果为：" + forkJoinTask.get());
            } catch (InterruptedException e) {
                e.printStackTrace();
            } catch (ExecutionException e) {
                e.printStackTrace();
            }
        }
    }

**执行结果：**

> 开启线程进行计算0
> 
> start-1, end-100这个任务需要进行拆分任务计算。。。ForkJoinPool-1-worker-1
> 
> 开启线程进行计算1
> 
> start-1, end-50这个任务需要进行拆分任务计算。。。ForkJoinPool-1-worker-1
> 
> 开启线程进行计算2
> 
> start-51, end-100这个任务需要进行拆分任务计算。。。ForkJoinPool-1-worker-2
> 
> 开启线程进行计算4
> 
> 开启线程进行计算3
> 
> 开启线程进行计算6
> 
> 开启线程进行计算5
> 
> 最终计算结果为：5050

![2256c2a4e64c6f568b967688097e8597.png][]

[6a9af2a5cf319f137316545b9ccaa7dd.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/27/be9247bd1e8f4f918ce37fefb9a85be7.png
[8cc37af530f74f7098ea37df616d827f.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/27/f39f218a3992475092e092f7b38a42e8.png
[88f4c179b4d2d1b2e331ae31d1c0df64.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/27/3071b8da0108447bb8230c797544345d.png
[923ab1ce7d36e5a77902dc312227c851.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/27/615953e46664460c995499ca847d6d0c.png
[2256c2a4e64c6f568b967688097e8597.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/27/c5e9872963f94e10a13897ec7cd48737.png