JUC 并发包中常用的工具类的使用

水深无声 2023-09-30 20:53 1阅读 0赞

java.util.concurrent工具包的简称就是JUC，这是一个处理线程的工具包 。AQS 被认为是 J.U.C 的核⼼

**目录**

CountDownLatch类 计数器

Semaphore类 信号量

CyclicBarrier类 循环屏障

J.U.C - 其它组件

FutureTask

BlockingQueue 阻塞队列

常用API

BlockingQueue队列下的生产者和消费者

--------------------

## CountDownLatch类 计数器 ##

如果某个线程的执行需要等待其它线程执行完，CountDownLatch能够阻塞当前线程，当其它线程倒数完成后，再执行，相当于大号的join()

创建时要定义倒计时的数字：

new CountDownLatch(int 倒计时数) 实际上就是要等待的线程数

主要API：

*  await() 阻塞当前线程
 *  countDown() 倒计时1次 -1
 *  getCount() 获得倒计时数

**
     * 使用CountDownLatch
     */
    public class CountDownLatchDemo {
    
        public static void main(String[] args) {
            //定义CountDownLatch对象
            CountDownLatch latch = new CountDownLatch(3);
            Thread thread = new Thread(() ->{
                System.out.println(Thread.currentThread().getName()+"准备好了！"+latch.getCount());
                try {
                    //阻塞当前线程，直到有3个线程调用countdown方法，将计数器减到0才会继续执行后面的内容
                    latch.await();
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
                System.out.println(Thread.currentThread().getName()+"冲刺了，到达终点！");
            });
            thread.setName("闪电");
            thread.start();
    
            for (int i = 0; i < 3; i++) {
                new Thread(() -> {
                    //倒计时一次   将次数-1,
                    latch.countDown();
                    System.out.println(Thread.currentThread().getName()+"倒计时完成！"+latch.getCount());
                }).start();
                try {
                    Thread.sleep(3000);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
        }
    }

![8efcadefa79647f6b00d68cf5d59bc8a.png][]

## Semaphore类 信号量 ##

Semaphore 类似于操作系统中的信号量，可以控制对互斥资源的访问线程数

信号量[Semaphore][] 就是一个计数器，表示当前可用资源的个数。可以控制线程的启动数量，从而**控制并发量**

它允许n个任务同时访问某个资源，可以将信号量看做是在向外分发使用资源的许可证，**只有成功获取许可证，才能使用资源**。

可以把它简单的理解成我们停车场入口立着的那个显示屏，每有一辆车进入停车场显示屏就会显示剩余车位减1，每有一辆车从停车场出去，显示屏上显示的剩余车辆就会加1，当显示屏上的剩余车位为0时，停车场入口的栏杆就不会再打开，车辆就无法进入停车场了，直到有一辆车从停车场出去为止。

创建时设置信号量：

new Semaphone(int 可用资源) //创建信号量 参数1 信号量值 参数2 是否公平锁（公平锁，先等待获得锁几率更大，非公平锁，都一样）

核心操作：PV：P-申请资源操作 V-释放资源操作

Semaphore的PV加减操作都是原子性的，在多线程场景下可以直接使用

常用API：

*  acquire() 申请一个信号量 P操作，每次申请一个资源，如果可用资源为空，则进入阻塞状态，直到有其他线程释放资源
    
     *  acquire(int n) 每次申请n个资源
 *  release() 释放信号量 V操作，每次释放一个资源
    
     *  release(int n) 每次释放n个资源
 *  availablePermits() 获取当前可用信号量数

//以下代码模拟了对某个服务的并发请求，每次只能有 3 个客户端同时访问，请求总数100
    public class SemaphoreExample {
        public static void main(String[] args) {
            final int clientCount = 3;
            final int totalRequestCount = 10;
            Semaphore semaphore = new Semaphore(clientCount);
            ExecutorService executorService = Executors.newCachedThreadPool();
            for (int i = 0; i < totalRequestCount; i++) {
                executorService.execute(()->{
                    try {
                            System.out.print(semaphore.availablePermits() + " ");
                            if(semaphore.availablePermits()==0){
                                System.out.println("资源不足，请等待。。。");
                            }
                            semaphore.acquire();
                            Thread.sleep(new Random().nextInt(10000));
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    } finally {
                        semaphore.release();
                    }
                });
            }
            executorService.shutdown();
        }
    }

## CyclicBarrier类 循环屏障 ##

⽤来控制多个线程互相等待，只有当多个线程都到达时，这些线程才会继续执⾏。

和 CountdownLatch 相似，都是通过维护计数器来实现的。线程执⾏ await() ⽅法之后计数器会减 1， 并进⾏等待，直到计数器为 0，所有调⽤ await() ⽅法⽽在等待的线程才能继续执⾏

CyclicBarrier 和 CountdownLatch 的⼀个区别是，**CyclicBarrier 的计数器通过调⽤ reset() ⽅法可以循 环使⽤，所以它才叫做循环屏障**

CountdownLatch中子线程并不会因为调用countDown()而阻塞,会继续进行该做的工作**,只是通知计数器-1,**只需要在所有进程都进行到某一节点后才会执行被阻塞的进程.如果我们想要多个线程在同一时间进行就要用到CyclicBarrier了 ,调用await()**会阻塞当前线程，并通知计数器-1**，归0后会同一时间执行之前阻塞的线程

![72ba77a7c365498f97e78b301a9ba980.png][]

CyclicBarrier 有两个构造函数，其中 parties 指示计数器的初始值，barrierAction 在所有线程都到达屏障的时候会执⾏⼀次

public class CyclicBarrierExample {
        public static void main(String[] args) {
            CyclicBarrier cyclicBarrier = new CyclicBarrier(10);
            ExecutorService executorService = Executors.newCachedThreadPool();
            for (int i = 0; i < totalThread; i++) {
                executorService.execute(() -> {
                    System.out.print("before..");
                    try {
                        cyclicBarrier.await();
                    } catch (InterruptedException | BrokenBarrierException e) {
                        e.printStackTrace();
                    }
                    System.out.print("after..");
                });
            }
            executorService.shutdown();
        }
    }
    
    //输出结果
    //before..before..before..before..before..before..before..before..before..before..after..after..after..after..after..after..after..after..after..after..

## J.U.C - 其它组件 ##

### FutureTask ###

在介绍 Callable 时我们知道它可以有返回值，返回值通过 Future 进⾏封装。FutureTask 实现了 RunnableFuture 接⼝，该接⼝继承⾃ Runnable 和 Future 接⼝，这使得 FutureTask 既可以当做⼀ 个任务执⾏，也可以有返回值

作用：FutureTask 可⽤于异步获取执⾏结果或取消执⾏任务的场景。当⼀个计算任务需要执⾏很⻓时间，那么 就可以⽤ FutureTask 来封装这个任务，主线程在完成⾃⼰的任务之后再去获取结果。

public static void main(String[] args) throws ExecutionException, InterruptedException {
            FutureTask<Integer> futureTask = new FutureTask<Integer>(
                    new Callable<Integer>() {
                        @Override
                        public Integer call() throws Exception {
                            int result = 0;
                            for (int i = 0; i < 100; i++) {
                                Thread.sleep(10);
                                result += i;
                            }
                            return result;
                        }
            });
            Thread computeThread = new Thread(futureTask);
            computeThread.start();
            Thread otherThread = new Thread(() -> {
                System.out.println("other task is running...");
                try {
                    Thread.sleep(1000);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            });
            otherThread.start();
            System.out.println(futureTask.get());//other task is running...      4950
        }

### BlockingQueue 阻塞队列 ###

阻塞队列是一种集合，当数据达到临界值时自动让生产者线程等待，数据为空时自动让消费者线程等待，也能根据条件自动唤醒线程

特点：线程安全，无法扩容，采用了生产者消费者模式

![d25650ee07294cb0a6e56a09d32d052e.png][]

**阻塞添加**：当队列已满时，往队列中添加元素会阻塞等待

**阻塞删除**：当队列为空时，往队列中删除或获取元素将被阻塞

常用实现类有：

*  ArrayBlockingQueue **有界队列** 基于数组结构的一个有界阻塞队列，先进先出，存取互相排斥（存取是同一把锁，操作的是同一数组对象）
 *  **LinkedBlockingQueue** **无界队列** 基于链表结构的一个无界阻塞队列，先进先出，存取互不干扰（锁分离，存取不同的Node对象）可以指定容量
 *  DelayQueue 带延迟的阻塞队列 基于优先级队列实现的无界阻塞队列，为空时阻塞
 *  SynchronousQueue **直接提交** 不存储元素的阻塞队列，是一个没有缓存的Blocking，不会为队列中元素维护存储空间，它只是多个线程之间 数据交换的媒介
 *  **PriorityBlockingQueue** **优先级队列**

**线程池用的是LinkedBlockingQueue** **线程池的排队策略与BlockingQueue有关 （可查看线程池 `newScheduledThreadPool` 的创建方式）。**

#### 常用API ####

<table> 
 <thead> 
  <tr> 
   <th>方法</th> 
   <th>抛出异常</th> 
   <th>不会抛出异常，有返回值</th> 
   <th>阻塞等待</th> 
   <th>超时等待</th> 
  </tr> 
 </thead> 
 <tbody> 
  <tr> 
   <td>添加</td> 
   <td>add() 到临界值会抛异常</td> 
   <td>offer()</td> 
   <td>put() 到临界值自动阻塞</td> 
   <td>offer(e,timeout,TimeUnit)</td> 
  </tr> 
  <tr> 
   <td>移除</td> 
   <td>remove() 为空抛出异常</td> 
   <td>poll()</td> 
   <td>take() 队列为空自动阻塞</td> 
   <td>poll(e,timeout,TimeUnit)</td> 
  </tr> 
  <tr> 
   <td>获取队首元素</td> 
   <td>element()</td> 
   <td>peek()</td> 
   <td></td> 
   <td></td> 
  </tr> 
 </tbody> 
</table>

说明：

抛出异常：add()队列到临界值抛的异常是IllegalStateException ；remove()队列为空是抛的异常是NoSuchException

有返回值：offer()插入结果返回布尔值，poll()移除结果返回移除的元素

阻塞：put()队列到临界值会一直阻塞，直到put成功或者出现中断退出；take()队列为空会一直阻塞，直到队列可用

超时等待：当队列满时，会阻塞生产者线程一段时间，超过时间后生产者线程会自动退出

#### BlockingQueue队列下的生产者和消费者 ####

class MyResource {
        // 默认开启，进行生产消费
        // 这里用到了volatile是为了保持数据的可见性，也就是当TLAG修改时，要马上通知其它线程进行修改
        private volatile boolean FLAG = true;
    
        // 使用原子包装类，而不用number++
        private AtomicInteger atomicInteger = new AtomicInteger();
    
        // 这里不能为了满足条件，而实例化一个具体的SynchronousBlockingQueue
        BlockingQueue<String> blockingQueue = null;
    
        // 而应该采用依赖注入里面的，构造注入方法传入
        public MyResource(BlockingQueue<String> blockingQueue) {
            this.blockingQueue = blockingQueue;
            // 查询出传入的class是什么
            System.out.println(blockingQueue.getClass().getName());
        }
    
    
        public void myProducer() throws Exception{
            String data = null;
            boolean retValue;
            // 多线程环境的判断，一定要使用while进行，防止出现虚假唤醒
            // 当FLAG为true的时候，开始生产
            while(FLAG) {
                data = atomicInteger.incrementAndGet() + "";
    
                // 2秒存入1个data
                retValue = blockingQueue.offer(data, 2L, TimeUnit.SECONDS);
                if(retValue) {
                    System.out.println(Thread.currentThread().getName() + "\t 插入队列:" + data  + "成功" );
                } else {
                    System.out.println(Thread.currentThread().getName() + "\t 插入队列:" + data  + "失败" );
                }
    
                try {
                    TimeUnit.SECONDS.sleep(1);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
    
            System.out.println(Thread.currentThread().getName() + "\t 停止生产，表示FLAG=false，生产介绍");
        }
    
    
        public void myConsumer() throws Exception{
            String retValue;
            // 多线程环境的判断，一定要使用while进行，防止出现虚假唤醒
            // 当FLAG为true的时候，开始生产
            while(FLAG) {
                // 2秒移除1个data
                retValue = blockingQueue.poll(2L, TimeUnit.SECONDS);
                if(retValue != null && retValue != "") {
                    System.out.println(Thread.currentThread().getName() + "\t 消费队列:" + retValue  + "成功" );
                } else {
                    FLAG = false;
                    System.out.println(Thread.currentThread().getName() + "\t 消费失败，队列中已为空，退出" );
    
                    // 退出消费队列
                    return;
                }
            }
        }
    
        /**
         * 停止生产的判断
         */
        public void stop() {
            this.FLAG = false;
        }
    
    }
    
    public class BlockingQueueProducerConsumer {
    
        public static void main(String[] args) {
            // 传入具体的实现类， ArrayBlockingQueue
            MyResource myResource = new MyResource(new ArrayBlockingQueue<String>(10));
    
            new Thread(() -> {
                System.out.println(Thread.currentThread().getName() + "\t 生产线程启动\n\n");
    
                try {
                    myResource.myProducer();
                    System.out.println("\n");
    
                } catch (Exception e) {
                    e.printStackTrace();
                }
            }, "producer").start();
    
    
            new Thread(() -> {
                System.out.println(Thread.currentThread().getName() + "\t 消费线程启动");
    
                try {
                    myResource.myConsumer();
                } catch (Exception e) {
                    e.printStackTrace();
                }
            }, "consumer").start();
    
            // 5秒后，停止生产和消费
            try {
                TimeUnit.SECONDS.sleep(5);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
    
    
            System.out.println("\n\n5秒中后，生产和消费线程停止，线程结束");
            myResource.stop();
        }
    }

[8efcadefa79647f6b00d68cf5d59bc8a.png]: https://img-blog.csdnimg.cn/8efcadefa79647f6b00d68cf5d59bc8a.png
[Semaphore]: https://so.csdn.net/so/search?q=Semaphore&spm=1001.2101.3001.7020
[72ba77a7c365498f97e78b301a9ba980.png]: https://img-blog.csdnimg.cn/72ba77a7c365498f97e78b301a9ba980.png
[d25650ee07294cb0a6e56a09d32d052e.png]: https://img-blog.csdnimg.cn/d25650ee07294cb0a6e56a09d32d052e.png