AQS的几个同步组件

Posted 2022-11-10 shizhuoping

AQS的几个同步组件

CountDownLatch

一个线程或多个线程一直等待，直到其他线程执行的操作完成才继续执行。

应用：并行计算。

计数器是不能重计的，计数值不能往上加，类似操作只有一次。

使用时首先new一个countDownLatch，构造方法中放入计数，然后在一个线程中调用await方法，这个线程就会进入等待状态，其他线程中调用countDown方法减少这个计数，直至降到0时，await方法阻塞的线程会被唤醒继续工作，为保证顺利执行，countDown方法一定要放在finally块里保证。

countdownLatch的await方法还可以设置超时时间，第一参数是数字，第二个是单位，超过这个时间await的线程就会被唤醒，之前给定的线程还会继续执行完。

Semaphore（信号量）

控制某个资源可以被多少个线程访问

应用场景：资源有限的并发控制，如数据库连接。

使用时首先new一个semaphere，然后参数为可以并发访问的个数，如20。

然后在线程中调用acquire方法和release方法，把要控制并发的执行部分放入这两部分之间，然后就完成了，这样执行时，同时访问的线程最多只有20个，这样在控制并发场景数据库连接时，就不会因为获取连接得到异常。

acquire方法和release方法可以同时获取或释放多个许可，传入许可数当做方法参数即可。

还可以完成这样的场景，如果参与竞争的线程没有获得许可就直接放弃，可以在if语句中放入tryAcquire方法，如果获取许可执行操作，没有获取就可以放弃或者执行对应操作。

tryAcquire方法可以尝试获取多个许可，还可以传入时间参数，表示没有获取可以等一段时间，如果还是没有再返回false。

CyclicBarrier

可以实现多个线程相互等待，当所有线程就绪后，各线程才能接下来执行后面的操作。

应用场景：多线程计算

它和countDownLatch区别：它可以循环使用、它是多个线程相互等待。

当一个线程调用await方法时，计数器的值加1，直至计数器加到对应值了，然后释放所有线程，计数器又变回0，可以重用。

首先new一个CyclicBarrier，把参数传入代表释放的标记值是多少。然后各线程执行时调用await方法进行阻塞，调用await方法一次标记就会加一，直到到要求的值就将全部阻塞线程释放，一起执行。CyclicBarrier构造方法中也可以加一个runnable，其中的任务当到达标记值时优先执行，执行完之后再释放其他线程。

await方法也可以设置时间参数，过了这个时间也会自动唤醒当前线程，但是在执行过程中方法会抛出BrokenBarrierException，要把异常catch住然后继续执行，这样就能实现自动唤醒。

Exchanger

用于两个线程交换值然后继续运行。

新建一个Exchanger对象，它是有泛型的，泛型的类型就是要交换值的类型。然后线程1调用其exchange方法将要交换的值1传入，然后线程1进入阻塞状态等待线程2交换。线程2调用exchange方法将要交换的值2传入，然后线程1获得该值同时线程2也拿到了值1.两个线程继续运行。

exchange方法还可以传时间+时间单位进去，意思就是等待一段时间如果线程2还不交换就直接返回。

Fork/Join框架

这个框架是为了解决并行计算小问题然后汇总的这类问题。采用工作窃取算法，为每个线程分配一个队列，各个线程从各自的队列头部中取任务然后执行，如果某个线程先执行完对应的队列就从其他线程的队列尾端取任务执行，只有当队列中有一个任务时才有可能产生竞争关系，这个算法充分利用了多线程的优势进行并行计算，并减少了线程间的竞争。这个框架使用还有一些限制：

1、只能用fork/join来控制线程，否则会破坏线程机制

2、任务不能执行IO操作

3、任务不能抛出异常

使用时首先建立一个ForkJoinPool对象，然后构造一个计算类a实现RecursiveTask接口，重写compute方法，这个方法返回值就是计算结果，然后new一个a对象，将a对象提交到pool中，得到一个future对象，然后future调用get就能获取计算结果。在compute方法计算中，应用划分问题的思想，对于大问题划分为小问题，小问题计算后将结果直接return，大问题划分时直接新建a对象，然后该对象调用fork方法开始计算，用join方法获取计算结果，将结果汇总返回。

建立一个类ForkJoinTask1继承RecursiveTask接口，重写compute方法，在这个方法内的逻辑就是如果问题大就细化问题，建立小问题的ForkJoinTask1，然后调用fork和join方法得到小问题的结果，最后得到大问题的值。如果是小问题就直接计算不必拆分。

public class ForkJoinTask1 extends RecursiveTask<Integer>

    public static final int num = 2;
    private int start;
    private int end;

    public ForkJoinTask1(int start, int end) 
        this.start = start;
        this.end = end;
    
    
    @Override
    protected Integer compute() 
        // TODO Auto-generated method stub
        int sum = 0;
        boolean canCompute = (end - start) <= num;
        if(canCompute) 
            for(int i = start; i <= end; i++) 
                sum += i;
            
        else 
            int middle = (start + end) / 2;
            ForkJoinTask1 leftTask = new ForkJoinTask1(start, middle);
            ForkJoinTask1 rightTask = new ForkJoinTask1(middle + 1, end);
            
            //执行子任务
            leftTask.fork();
            rightTask.fork();
            
            //等待任务执行结束合并其结果
            int leftResult = leftTask.join();
            int rightResult = rightTask.join();
            
            sum = leftResult + rightResult;
        
        return sum;
    

调用时首先建立ForkJoinPool对象，然后建立任务类ForkJoinTask1，将任务对象提交到pool中得到一个Future，get取到结果。

        ForkJoinPool pool = new ForkJoinPool();
        
        ForkJoinTask1 task = new ForkJoinTask1(1, 100);
        
        Future<Integer> result = pool.submit(task);
        
        System.out.println(result.get());