并发工具类

1.CountDownLatch

  CountDownLatch是一个同步计数器,初始化的时候传入需要计数的线程等待数,可以是需要等待执行完成的线程数,或者大于;

  作用:用来协调多个线程之间的同步,或者说起到线程之间的通信(而不是用作互斥的作用)。是一组线程等待其他的线程完成工作以后在执行,相当于加强版join;

  await():阻塞当前线程,等待其他线程执行完成,直达计数器计数值减到0;

  countDown():负责计算器的减一;

package com.wn.Test01;

import java.util.concurrent.CountDownLatch;

public class CountDownLathTest01 {
    public static void main(String[] args) throws InterruptedException {
        System.out.println("等待子线程执行完毕......");
        final CountDownLatch countDownLatch=new CountDownLatch(2);
        new Thread(new Runnable() {
            @Override
            public void run() {
                System.out.println("子线程:"+Thread.currentThread().getName()+"开始执行......");
                countDownLatch.countDown(); //每次减去1
                System.out.println("子线程:"+Thread.currentThread().getName()+"结束执行......");
            }
        }).start();
        new Thread(new Runnable() {
            @Override
            public void run() {
                System.out.println("子线程:"+Thread.currentThread().getName()+"开始执行......");
                countDownLatch.countDown();;
                System.out.println("子线程:"+ Thread.currentThread().getName()+"结束执行......");
            }
        }).start();
        countDownLatch.await(); //调用当前方法主线程阻塞   countDown结果为0,阻塞变为运行状态
        System.out.println("两个子线程执行完毕......");
        System.out.println("继续主线程执行...");

    }
}

    

2.CyclicBarrier

  CyclicBarrier字面意思是栅栏,是多线程中一个重要的类,主要用于线程内部之间的线程的相互等待问题,初始化的时候传入需要等待的线程数;

  作用:让一组线程达到某一个屏障被阻塞,一直到组内最后一个线程达到屏障时,屏障开放,所有被阻塞的线程才会继续运行;

  CyclicBarrier(int parties):初始化定义需要等待的线程数parties;

  CyclicBarrier(int parties,Runnable barrierAction):当屏障开放的时候,线程barrierAction的任务会执行;

package com.wn.Test01;

import java.util.concurrent.BrokenBarrierException;
import java.util.concurrent.CyclicBarrier;

public class CyclicBarrierTest01 extends Thread {
    private CyclicBarrier cyclicBarrier;
    public CyclicBarrierTest01(CyclicBarrier cyclicBarrier){
        this.cyclicBarrier=cyclicBarrier;
    }

    public void run(){
        System.out.println("线程:"+Thread.currentThread().getName()+",正在写入数据......");
        try {
            Thread.sleep(3000); //休眠
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        System.out.println("线程"+Thread.currentThread().getName()+",写入数据成功......");

        try {
            cyclicBarrier.await();      //等待
        } catch (InterruptedException e) {
            e.printStackTrace();
        } catch (BrokenBarrierException e) {
            e.printStackTrace();
        }
        System.out.println("所有线程执行完毕......");
    }

    public static void main(String[] args) {
        //设置等待线程数量,当线程数量达到指定数量时,统一向下执行
        CyclicBarrier cyclicBarrier=new CyclicBarrier(5);
        for (int i=0;i<5;i++){
            CyclicBarrierTest01 test01=new CyclicBarrierTest01(cyclicBarrier);
            test01.start(); //开始

        }
    }
}

    

  CountDownLatch和CyclicBarrier的区别:

    1.CountDownLatch放行由第三者控制,CyclicBarrier放行由一组线程本身控制;

    2.CountDownLatch放行条件>=线程数,CyclicBarrier放行条件=线程数;

    3.CountDownLatch会阻塞主线程,CyclicBarrier不会阻塞主线程,只会阻塞子线程;

    4.CountDownLatch的计数器只能使用一次,而CyclicBarrier的计数器可以使用reset()方法重置,所以CyclicBarrier能处理更为复杂的业务场景,比如如果计算发生错误,可以重置计数器,并让我们重新执行一次;

3.Semaphore

  Semaphore有名信号量,是操作系统中的一个概念,在java并发编程中,信号量控制的是线程并发的数量;

  作用:Semaphore管理一系列许可证。每个acquire方法阻塞,直到有一个许可证可以获得然后拿走一个许可证;每个release方法增加一个许可证,这可能会释放一个阻塞的acquire方法。然而,其实并没有实际的许可证这个对象,Semaphore只是维持一个可获得许可证的数量,主要控制同时访问某个特定资源的线程数量,多用在流量控制;

  注意:其他Semaphore的底层显示就是基于AQS的共享锁实现的

  如果一个线程要访问共享资源,必须先获得信号量,如果信号量的计数器值大于1,意味着有共享资源可以访问,则使其计数器值减去1,在访问共享资源。如果计数器值为0,线程进入休眠。当某个线程使用完共享资源后,释放信号量,并将信号量内部的计数器加1,之间进入休眠的线程将被唤醒并再次试图获取信号量;

package com.wn.conclass;

import java.util.Random;
import java.util.concurrent.Semaphore;

public class SemaphoreDemo implements Runnable {
    private String name;
    private Semaphore wc;

    public SemaphoreDemo(String name, Semaphore wc) {
        this.name = name;
        this.wc = wc;
    }

    @Override
    public void run() {
        try {
            //剩下的资源
            int availablePermits = wc.availablePermits();
            if (availablePermits>0){
                System.out.println(name+"有了有了......");
            }else{
                System.out.println(name+"没了没了......");
            }
            //申请,如果资源达到3次,就等待
            wc.acquire();
            System.out.println(name+"到了到了......");
            Thread.sleep(new Random().nextInt(1000));   //模拟过程的时间
            System.out.println(name+"结束了......");
            wc.release();   //释放资源
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }

    public static void main(String[] args) {
        Semaphore semaphore=new Semaphore(3);
        for (int i=1;i<=10;i++){
            SemaphoreDemo parent=new SemaphoreDemo(""+i+"个人",semaphore);
            new Thread(parent).start();
        }
    }
}

    

4.Exchanger

  Exchanger类似于一个交换器,可以对元素进行配对和交换的线程的同步点,用于两个线程间的数据交换;

  具体来说,Exchanger类允许在两个线程之间定义同步点。当两个线程都到达同步点时,它们交换数据结构,因此第一个线程的数据结构进行第二个线程中,第二个线程的数据结构进入到第一个线程中;

  就像两个线程各个交换自己的数据;

package com.wn.conclass;

import java.util.concurrent.Exchanger;

public class ExchangerTest01 {
    private static String str1="资源1";
    private static String str2="资源2";
    //构建资源交换对象
    public static Exchanger<String> stringExchanger=new Exchanger<>();
    public static void main(String[] args){
        //第一个线程
        new Thread(new Runnable() {
            @Override
            public void run() {
                System.out.println(Thread.currentThread().getName()+"初始占用资源:"+str1);
                //资源交换,将资源交给其他线程和获取到其他线程交换过来的资源
                try {
                    String newStr = stringExchanger.exchange(str1);
                    System.out.println(Thread.currentThread().getName()+"交换资源:"+newStr);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
        }).start();
        //第二个线程
        new Thread(new Runnable() {
            @Override
            public void run() {
                System.out.println(Thread.currentThread().getName()+"初始占用资源:"+str2);
                //资源交换,将资源交给其他线程和获取到其他线程交换过来的资源
                try {
                    String newStr = stringExchanger.exchange(str2);
                    System.out.println(Thread.currentThread().getName()+"交换资源:"+newStr);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
        }).start();

    }
}

    

 

原文地址:https://www.cnblogs.com/wnwn/p/12522875.html