java中并发常用工具类及示例代码

Posted 2022-12-16 零

前言:在你无聊的时候,想想比你优秀还努力的人,也许就不觉的无聊了
今日记录：四个并发中可能会用到的工具类,分别是：
CountDownLatch
CyclicBarrier
Semaphore
Exchanger

CountDownLatch

是一组线程等待其他的线程完成工作以后在执行，加强版join

区别在于：

• 调用thread.join() 方法必须等thread 执行完毕，当前线程才能继续往下执行
• 而CountDownLatch通过计数器提供了更灵活的控制，只要检测到计数器为0当前线程就可以往下执行而不用管相应的thread是否执行完毕

我们可以把它理解为倒计时锁

举个栗子：

1.我们在玩LOL英雄联盟时会出现十个人不同加载状态，但是最后一个人由于各种原因始终加载不了100%，于是游戏系统自动等待所有玩家的状态都准备好，才展现游戏

示例代码

public class LoadCountDownLatch 
    static CountDownLatch countDownLatch = new CountDownLatch(10);

    public static void main(String[] args) throws InterruptedException 
        for (int i = 1; i <= 10; i++) 
            Thread.sleep(1000);
            new Thread(() -> 
                System.out.println(Thread.currentThread().getName() + "加入游戏");
                countDownLatch.countDown();
            , "玩家" + i).start();
        
        countDownLatch.await();
        System.out.println("游戏启动");
    

2.每位乘客(线程)上车后，可用座位减1，直到为0，老司机就开始发车了

示例代码

/*
    *类说明：演示CountDownLatch，有5个初始化的线程，6个扣除点，
    *扣除完毕以后，主线程和业务线程才能继续自己的工作
    */
    public class UseCountDownLatch 

        static CountDownLatch latch = new CountDownLatch(6);

        // 初始化线程(只有一步，有4个)
        private static class InitThread implements Runnable 

            @Override
            public void run() 
                System.out.println("初始化线程_" + Thread.currentThread().getId() + " 准备好初始化工作......");
                latch.countDown(); //初始化线程完成工作了，countDown方法只扣减一次；
                for (int i = 0; i < 2; i++) 
                    System.out.println("初始化线程_" + Thread.currentThread().getId() + " 继续其他工作........");
                
            
        

        // 业务线程
        private static class BusiThread implements Runnable 

            @Override
            public void run() 
                try 
                    latch.await();
                 catch (InterruptedException e) 
                    e.printStackTrace();
                
                for (int i = 0; i < 3; i++) 
                    System.out.println("业务线程_" + Thread.currentThread().getId() + " 执行业务-----");
                
            
        

        public static void main(String[] args) throws InterruptedException 
            //单独的初始化线程,初始化分为2步，需要扣减两次
            new Thread(() -> 
                System.out.println("其他准备线程_" + Thread.currentThread().getId()
                                   + " 准备好初始化工作步骤 1......");
                latch.countDown();//每完成一步初始化工作，扣减一次
                System.out.println("开始第2步.......");

                System.out.println("其他准备线程_" + Thread.currentThread().getId()
                                   + " 准备好初始化工作步骤 2......");
                latch.countDown();//每完成一步初始化工作，扣减一次
            ).start();

            // 执行业务线程
            new Thread(new BusiThread()).start();

            // 初始化线程创建4个，执行
            for (int i = 0; i <= 3; i++) 
                new Thread(new InitThread()).start();
            
            latch.await();
            System.out.println("主线程执行........");
        
    

CyclicBarrier

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

可看成是个障碍，所有的线程必须到齐后才能一起通过这个障碍

举个栗子：

以前公司组织户外拓展活动，帮助团队建设，其中最重要一个项目就是全体员工（包括女同事，BOSS）在完成其他项目时，到达一个高达四米的高墙没有任何抓点，要求所有人，一个不能少的越过高墙，才能继续进行其他项目

示例代码

public class UseCyclicBarrier 

    private static CyclicBarrier barrier = new CyclicBarrier(5, () -> 
        System.out.println("所有员工翻过高墙，开始干饭");
    );

    // 1.翻过高墙
    private static class SubThread implements Runnable 

        @Override
        public void run() 
            try 
                m1();
             catch (Exception e) 
                e.printStackTrace();
            
        

        public void m1() throws Exception 
            Random r = new Random();//随机决定工作线程的是否睡眠
            if (r.nextBoolean()) 
                Thread.sleep(1000);
                System.out.println(Thread.currentThread().getName() + "翻过高墙！！！");
                barrier.await();
             else 
                System.out.println(Thread.currentThread().getName() + "没有过");
                m1();
            
        
    

    public static void main(String[] args) throws BrokenBarrierException, InterruptedException 
        for (int i = 0; i <= 4; i++) 
            Thread thread = new Thread(new SubThread(), "员工" + i);
            thread.start();
        
        System.out.println("主线程");
    

Semaphore

Semaphore 通常我们叫它信号量， 可以用来控制同时访问特定资源的线程数量，通过协调各个线程，以保证合理的使用资源

通常用于那些资源有明确访问数量限制的场景，常用于限流

比如：数据库连接池，同时进行连接的线程有数量限制，连接不能超过一定的数量，当连接达到了限制数量后，后面的线程只能排队等前面的线程释放了数据库连接才能获得数据库连接。

比如：停车场场景，车位数量有限，同时只能容纳多少台车，车位满了之后只有等里面的车离开停车场外面的车才可以进入。

常用方法说明

acquire()  
获取一个令牌，在获取到令牌、或者被其他线程调用中断之前线程一直处于阻塞状态。

acquire(int permits)  
获取一个令牌，在获取到令牌、或者被其他线程调用中断、或超时之前线程一直处于阻塞状态。
    
acquireUninterruptibly() 
获取一个令牌，在获取到令牌之前线程一直处于阻塞状态（忽略中断）。
    
tryAcquire()
尝试获得令牌，返回获取令牌成功或失败，不阻塞线程。

tryAcquire(long timeout, TimeUnit unit)
尝试获得令牌，在超时时间内循环尝试获取，直到尝试获取成功或超时返回，不阻塞线程。

release()
释放一个令牌，唤醒一个获取令牌不成功的阻塞线程。

hasQueuedThreads()
等待队列里是否还存在等待线程。

getQueueLength()
获取等待队列里阻塞的线程数。

drainPermits()
清空令牌把可用令牌数置为0，返回清空令牌的数量。

availablePermits()
返回可用的令牌数量。

示例代码

public class TestCar 
    //停车场同时容纳的车辆10
    private static Semaphore semaphore = new Semaphore(10);

    public static void main(String[] args) 

        //模拟100辆车进入停车场
        for (int i = 0; i < 50; i++) 

            Thread thread = new Thread(() -> 
                try 
                    System.out.println("====" + Thread.currentThread().getName() + "来到停车场");
                    if (semaphore.availablePermits() == 0) 
                        System.out.println("车位不足，请耐心等待");
                    
                    // 尝试获取令牌
                    semaphore.acquire();//获取令牌尝试进入停车场
                    System.out.println(Thread.currentThread().getName() + "成功进入停车场");
                    Thread.sleep(new Random().nextInt(1000));//模拟车辆在停车场停留的时间
                    System.out.println(Thread.currentThread().getName() + "驶出停车场");
                    semaphore.release();//释放令牌，腾出停车场车位
                 catch (InterruptedException e) 
                    e.printStackTrace();
                
            , i + "号车");
            thread.start();
        
    

Exchanger

Exchanger用于线程间的通信和数据交换；它提供了一个exchange方法，两个线程调用exchange方法时，线程1先调用该方法会进入阻塞状态，直到线程2调用该方法，然后安全交换数据，之后两个线程继续运行

Exchanger的原理是使用了ThreadLocal来实现的，至于线程阻塞的部分是使用java中的Unsafe类park()/unpark()来实现的

应用场景之一：游戏中玩家之间直接交易物品

示例代码

public class UseExchanger 
    private static final Exchanger<Set<String>> exchanger = new Exchanger<Set<String>>();
    private static Set<String> setA = new HashSet<String>();//存放数据的容器
    private static Set<String> setB = new HashSet<String>();//存放数据的容器
    public static void main(String[] args) 
        //第一个线程
        new Thread(new Runnable() 
            @Override
            public void run() 
                try 
                    /*添加数据
                     * set.add(.....)
                     * */
                    setA.add("屠龙");
                    System.out.println("玩家A交换之前:" + setA);
                    setA = exchanger.exchange(setA);//交换set
                    System.out.println("玩家A交换之后:" + setA);
                 catch (InterruptedException e) 
                    e.printStackTrace();
                
            
        ).start();

        //第二个线程
        new Thread(new Runnable() 
            @Override
            public void run() 
                try 
                    /*添加数据
                     * set.add(.....)
                     * set.add(.....)
                     * */
                    setB.add("1W");
                    System.out.println("玩家B交换之前:" + setB);
                    setB = exchanger.exchange(setB);//交换set
                    /*处理交换后的数据*/
                    System.out.println("玩家B交换之后:" + setB);
                 catch (InterruptedException e) 
                    e.printStackTrace();
                
            
        ).start();