JUC 工具类+阻塞队列介绍

淩亂°似流年 2024-03-26 23:19 76阅读 0赞

### 一、CountDownLatch(计数器) ###

CountDownLatch 俗称为计数器，主要是让一些线程阻塞直到另外一些工作线程完成后才被唤醒。

CountDownLatch 主要有两个方法，当一个或多个线程调用 await 方法时，调用线程会被阻塞，其他线程调用 countDown 方法计算器减 1(调用 countDown 方法时线程不会阻塞)，当计算器的值变为 0，因调用 await 方法被阻塞的线程被唤醒，继续执行。

#### 1.**CountDownLatch** 源码 ####

**CountDownLatch 相关方法源码介绍：**

public CountDownLatch(int count) {
            if (count < 0) throw new IllegalArgumentException("count < 0");
            this.sync = new Sync(count);
    }

上述方法设置指定的线程数(count)通过了，计数器才会被唤醒

public void countDown() {
            sync.releaseShared(1);
    }

上述方法每次执行都会对当前总数进行 -1 的操作，如若等待的线程数已经达到了阀值，所有的等待线程都会被唤醒，该方法不会起任何作用。

public boolean await(long timeout, TimeUnit unit)
            throws InterruptedException {
            return sync.tryAcquireSharedNanos(1, unit.toNanos(timeout));
    }

上述方法主要实现的是如果在指定的时长未完成计数器的归零操作，则放弃当前操作，程序停止运行，参数 1：timeout 时长，参数 2：unit 时间单位

public void await() throws InterruptedException {
            sync.acquireSharedInterruptibly(1);
     }

上述方法主要实现的是如果未完成计数器的归零操作，程序会一直在等待，等到计数器归零

#### 2.请求示例 ####

/**
     * @author vnjohn
     * @NAME: CountDownLatchDemo
     * @DATE: 2020/12/5
     */
    public class CountDownLatchDemo {
        public static void main(String[] args) throws Exception {
            closeDoor();
        }
    
        /**
         * 关门案例
         * @throws InterruptedException
         */
        private static void closeDoor() throws InterruptedException {
            CountDownLatch countDownLatch = new CountDownLatch(26);
            for (int i = 1; i <= 6; i++) {
                new Thread(() -> {
                    System.out.println(Thread.currentThread().getName() + "\t" + "上完自习");
                    countDownLatch.countDown();
                }, String.valueOf(i)).start();
            }
            countDownLatch.await(1, TimeUnit.SECONDS);
            System.out.println(Thread.currentThread().getName() + "\t班长锁门离开教室");
        }
    }

运行结果如下：

[![alt][]

![alt][]][alt_ _alt]

在指定时长未完成计数器的归零操作，程序超时停止运行。

### 二、CyclicBarrier(循环屏障) ###

#### 1.前言 ####

CyclicBarrier 的字面意思是可循环(Cyclic) 使用的屏障(barrier)，它要做的事情是，让一组线程到达一个屏障(也可以叫做同步点)时被阻塞，知道最后一个线程到达屏障时，屏障才会开门，所有被屏障拦截的线程才会继续干活，线程进行屏障通过 CyclicBarrier 的 await() 方法。

#### 2.CyclicBarrier 数据结构 ####

CyclicBarrier 底层是基于 ReetrantLock 和 AbstractQueuedSynchronizer 来实现的，所以，CyclicBarrier 的数据结构也依附于 AQS 的数据结构，在后面在详细介绍 AQS 的数据结构，在这里不做任何阐述。

#### 3.请求示例(集齐七龙珠) ####

/**
     * @author vnjohn
     * @NAME: CyclicBarrierDemo
     * @DATE: 2020/12/5
     */
    public class CyclicBarrierDemo {
        public static void main(String[] args) {
            CyclicBarrier cyclicBarrier=new CyclicBarrier(7,()->{
                System.out.println("召唤神龙");
            });
    
            for (int i = 1; i <=7; i++) {
                final int temp = i;
                new Thread(()->{
                    System.out.println(Thread.currentThread().getName()+"\t 收集到第"+ temp +"颗龙珠");
                    try {
                        cyclicBarrier.await();
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    } catch (BrokenBarrierException e) {
                        e.printStackTrace();
                    }
                },String.valueOf(i)).start();
            }
        }
    }

执行结果如下：

[![alt][alt 1]

![alt 1][]][alt_alt 1_ _alt 1]

### 三、Semaphore(信号量) ###

#### 1.基本概念 ####

信号量的主要用于两个目的，一个是用于多和共享资源的相互排斥使用，另一个用于并发资源数的控制。

#### 2.常用方法介绍 ####

public Semaphore(int permits)：创建指定条数的信号量数
    public Semaphore(int permits, boolean fair)：创建指定条数的信号量数并标识是否公平，true 为公平(等待时长会久一点)，false 为非公平
    semaphore.acquire()：获取资源
    semaphore.release()：释放资源

#### 3.请求示例(抢车位) ####

/**
     * @author vnjohn
     * @NAME: SemaphoreDemo
     * @DATE: 2020/12/6
     */
    public class SemaphoreDemo {
        public static void main(String[] args) {
            //模拟3个停车位
            Semaphore semaphore = new Semaphore(3);
            //模拟6部汽车
            for (int i = 1; i <= 6; i++) {
                new Thread(() -> {
                    try {
                        //抢到资源
                        semaphore.acquire();
                        System.out.println(Thread.currentThread().getName() + "\t抢到车位");
                        try {
                            TimeUnit.SECONDS.sleep(3);
                        } catch (InterruptedException e) {
                            e.printStackTrace();
                        }
                        System.out.println(Thread.currentThread().getName() + "\t 停3秒离开车位");
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    } finally {
                        //释放资源
                        semaphore.release();
                    }
                }, String.valueOf(i)).start();
            }
        }
    }

执行结果如下：

[![alt][alt 2]

![alt 2][]][alt_alt 2_ _alt 2]

### 四、阻塞队列(BlockingQueue) ###

#### 1.基本概念 ####

阻塞队列：顾名思义，首先它是一个队列，而一个阻塞队列在数据结构中所起的作用大致如图所示：

[![alt][alt 3]

![alt 3][]][alt_alt 3_ _alt 3]**线程 1 往阻塞队列中添加元素，线程 2 从队列中移除元素**

当阻塞队列是空时，从队列中**获取**元素的操作将会被阻塞

当阻塞队列已满时，往队列中**添加**元素的操作将会被阻塞

\*试图从空的阻塞队列中获取元素的线程将会被阻塞，直到其他的线程往空的队列**插入新的元素**

同样

\*试图往已满的阻塞队列中添加新元素的线程同样也会被阻塞，直到其他线程从队列中**移除一个或者多个元素或者全清空队列后**使队列重新变得空闲起来并后续新增

#### 2.为什么用？有什么好处？ ####

在多线程领域：所谓阻塞，在某些情况下会**挂起**线程(即线程阻塞)，一旦条件满足，被挂起的线程又会自动被唤醒

**为什么需要使用 BlockingQueue？**

好处是我们不需要关心什么时候需要阻塞线程，什么时候需要唤醒线程，因为这一切 BlockingQueue 都一手给你包办好了

在 concurrent 包发布之前，在多线程环境下，**我们每个程序猿都必须自己去控制这些细节，尤其还要兼顾效率和线程安全**，而这会给我们的程序带来不小的复杂度。

#### 3.BlockingQueue 的核心方法 ####

[![alt][alt 4]

![alt 4][]][alt_alt 4_ _alt 4]

*  抛出异常：add(e)=>当阻塞队列满时，再往队列里面 add 插入元素会抛 **java.lang.IllegalStateException: Queue full**，remove()=>当阻塞队列空时，再往队列 remove 元素时会抛出 **java.util.NoSuchElementException**
 *  特殊值：offer(e)=> 插入方法，成功返回 true 失败返回 false，poll=> 移除方法，成功返回元素，队列里面没有就返回 null .
 *  阻塞：put(e)=>当阻塞队列满时，生产者继续往队列里面 put 元素，队列会一直阻塞直到 put 数据 or 响应中断退出。take()=>当阻塞队列空时，消费者试图从队列 take 元素，队列会一直阻塞消费者线程直到队列可用.
 *  超时退出：offer(e,time,unit)=>当阻塞队列满时，队列会阻塞生产者线程一定时间，超过限时后生产者线程就会退出，poll(time,unit)=>当阻塞为空时，队列会阻塞消费者线程一定时间，超过限时后消费者线程就会退出。

代码示例：

**1) 抛出异常组**

/** 抛出异常组 */
        private static void isException() {
            BlockingQueue<String> blockingQueue=new ArrayBlockingQueue<>(3);
    //        添加进入三个队列
            System.out.println(blockingQueue.add("A"));
            System.out.println(blockingQueue.add("B"));
            System.out.println(blockingQueue.add("C"));
    //        以上 三个队列不会出错  下面这个队列添加进去的时候 由于队列已满 会报  Queue full 错误
            System.out.println(blockingQueue.add("d"));
    //        以下的这个方法是查询当前队列是否有元素 默认是查找第一个
            System.out.println(blockingQueue.element());
    //        移除当前的三个队列
            System.out.println(blockingQueue.remove());
            System.out.println(blockingQueue.remove());
            System.out.println(blockingQueue.remove());
    //        以上三个队列移除成功 由于队列中已经没有多余的元素,会报找不到元素异常
    //        System.out.println(blockingQueue.remove());
        }

**2) 特殊值组**

/** 特殊值组 */
        private static void isBoolean() {
            BlockingQueue<String> blockingQueue=new ArrayBlockingQueue<>(3);
    //        添加进入三个队列
            System.out.println(blockingQueue.offer("A"));
            System.out.println(blockingQueue.offer("B"));
            System.out.println(blockingQueue.offer("C"));
            System.out.println(blockingQueue.offer("D"));
            // 先检查第一个是否有值
            System.out.println(blockingQueue.peek());
    //        移除当前三个队列
            System.out.println(blockingQueue.poll());
            System.out.println(blockingQueue.poll());
            System.out.println(blockingQueue.poll());
            System.out.println(blockingQueue.poll());
        }

**3) 阻塞组**

/** 阻塞组 */
        private static void isBlocking() {
            BlockingQueue<String> blockingQueue = new ArrayBlockingQueue<>(3);
    //        添加三个队列
            try {
                blockingQueue.put("A");
                blockingQueue.put("B");
                blockingQueue.put("C");
    //      当队列已满 一直等待有空缺的地方   处于阻塞状态
    //            blockingQueue.put("D");
    //        移除三个队列
                blockingQueue.take();
                blockingQueue.take();
                blockingQueue.take();
    //        当队列已空 等待新的队列进入  处于阻塞状态
    //            blockingQueue.take();
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }

**4) 超时组**

/** 超时监控队列 */
        private static void isTimeBlocking() {
            BlockingQueue<String> blockingQueue = new ArrayBlockingQueue<>(3);
    //        添加三个队列
            try {
                System.out.println(blockingQueue.offer("A", 2L, TimeUnit.SECONDS));
                System.out.println(blockingQueue.offer("B", 2L, TimeUnit.SECONDS));
                System.out.println(blockingQueue.offer("C", 2L, TimeUnit.SECONDS));
                // 如果队列已满 等待两秒钟 看是否有空的队列 没有则返回false
                // System.out.println(blockingQueue.offer("D", 2L, TimeUnit.SECONDS));
                // 移除三个队列
                System.out.println(blockingQueue.poll(2L, TimeUnit.SECONDS));
                System.out.println(blockingQueue.poll(2L, TimeUnit.SECONDS));
                System.out.println(blockingQueue.poll(2L, TimeUnit.SECONDS));
    //        如果队列已空 等待两秒钟 看是否有新的队列进入 没有则返回null
                System.out.println(blockingQueue.poll(2L, TimeUnit.SECONDS));
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }

#### 4.架构梳理+种类分析 ####

##### 1) 架构介绍 #####

[![alt][alt 5]

![alt 5][]][alt_alt 5_ _alt 5]

##### 2) 种类分析 #####

*  **ArrayBlockingQueue**

public ArrayBlockingQueue(int capacity) {
            this(capacity, false);
    }

由数组结构组成的有界阻塞队列，capacity 是传入的 int，表示当前的队列数，第二个参数 fair 是传入的 boolean 类型，表示多线程环境下竞争是否公平

*  **LinkedBlockingQueue**

// 默认的构造函数
    public LinkedBlockingQueue() {
            this(Integer.MAX_VALUE);
    }
    // 带指定大小的链表队列
    public LinkedBlockingQueue(int capacity) {
            if (capacity <= 0) throw new IllegalArgumentException();
            this.capacity = capacity;
            last = head = new Node<E>(null);
    }

由链表结构组成的有界(但大小默认值 Integer.MAX\_VALUE )阻塞队列

*  PriorityBlockingQueue

支持优先级排序的无界阻塞队列

*  DelayQueue

是一个支持延时获取元素的无界阻塞队列。内部用 `PriorityQueue` 实现

*  **SynchronousQueue**

不存储元素的阻塞队列，也即是单个元素的队列

**理论**

SynchronousQueue 没有容量，与其他 BlockingQueue 不同，SynchronousQueue 是一个不存储元素的 BlockingQueue 每个 put 操作必须要等待一个 take 操作，否则不能继续添加元素，反之亦然。

**SynchronousQueue 代码案例演示：**

/**
     * @author vnjohn
     * 阻塞队列SynchronousQueue演示
     * @NAME: SynchronousQueueDemo
     * @DATE: 2020/12/7
     */
    public class SynchronousQueueDemo {
        public static void main(String[] args) {
            BlockingQueue<String> blockingQueue = new SynchronousQueue<>();
            new Thread(() -> {
                try {
                    System.out.println(Thread.currentThread().getName() + "\t put "+System.currentTimeMillis()/1000);
                    blockingQueue.put("1");
                    System.out.println(Thread.currentThread().getName() + "\t put "+System.currentTimeMillis()/1000);
                    blockingQueue.put("2");
                    System.out.println(Thread.currentThread().getName() + "\t put "+System.currentTimeMillis()/1000);
                    blockingQueue.put("3");
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }, "AAA").start();
    
            new Thread(() -> {
                try {
                    try {
                        TimeUnit.SECONDS.sleep(5);
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                    System.out.println(Thread.currentThread().getName() + "\t take " + blockingQueue.take()+System.currentTimeMillis()/1000);
                    try {
                        TimeUnit.SECONDS.sleep(5);
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                    System.out.println(Thread.currentThread().getName() + "\t take " + blockingQueue.take()+System.currentTimeMillis()/1000);
                    try {
                        TimeUnit.SECONDS.sleep(5);
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                    System.out.println(Thread.currentThread().getName() + "\t take " + blockingQueue.take()+System.currentTimeMillis()/1000);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }, "BBB").start();
        }
    }

执行结果如下：

[![alt][alt 6]

![alt 6][]][alt_alt 6_ _alt 6]

*  LinkedTransferQueue

由链表结构组成的无界阻塞队列，LinkedTransferQueue 是 SynchronousQueue 和 LinkedBlockingQueue 的合体，性能比 LinkedBlockingQueue 更高(没有锁操作)，比 SynchronousQueue 能存储更多的元素。

*  LinkedBlockingDeque

LinkedBlockingDeque 是一个基于链表的双端阻塞队列。和LinkedBlockingQueue 类似，区别在于该类实现了 Deque 接口，而LinkedBlockingQueue 实现了 Queue 接口

#### 5.用在哪里？ ####

##### 1) 生产者消费者模式 #####

*  **传统版**

**使用场景描述：**

一个初始值为 0 的变量，两个线程交替操作，一个加 1 一个减 1 来 5 轮

**代码思路：**

线程 A(生产者) 当初始值不为 0 时进行**唤醒(signalAll)**所有消费者线程，为 0 时进行线程阻塞**等待(wait)**当前线程生产再操作，线程 B(消费者) 当初始值为 0 时**唤醒(signalAll)**所有生产者线程，不为 0 时进行线程阻塞，**等待(wait)**当前线程消费再操作

**代码示例：**

/**
     * @author vnjohn
     * @NAME: ProdConsumerTraditionDemo
     * @DATE: 2020/12/7
     */
    public class ProdConsumerTraditionDemo {
        /**
         * 共享资源类
         */
        static class ShareData {
            private int num = 0;
            private Lock lock = new ReentrantLock();
            private Condition condition = lock.newCondition();
    
            public void increment() throws Exception {
                lock.lock();   //ReentrantLock
                try {
                    // 判断
                    while (num != 0) {
                        // 等待 不生产
                        condition.await();
                    }
                    // 干活
                    num++;
                    System.out.println(Thread.currentThread().getName() + "\t" + num);
                    // 通知唤醒
                    condition.signalAll();
                } finally {
                    lock.unlock();
                }
            }
    
            public void deIncrement() throws Exception {
                lock.lock();
                try {
                    // 判断
                    while (num == 0) {
                        // 等待 不生产
                        condition.await();
                    }
                    // 干活
                    num--;
                    System.out.println(Thread.currentThread().getName() + "\t" + num);
                    // 通知唤醒
                    condition.signalAll();
                } finally {
                    lock.unlock();
                }
            }
        }
    
        public static void main(String[] args) {
            ShareData shareData = new ShareData();
            new Thread(() -> {
                for (int i = 1; i <= 5; i++) {
                    try {
                        shareData.increment();
                    } catch (Exception e) {
                        e.printStackTrace();
                    }
                }
            }, "AA").start();
    
            new Thread(() -> {
                for (int i = 1; i <= 5; i++) {
                    try {
                        shareData.deIncrement();
                    } catch (Exception e) {
                        e.printStackTrace();
                    }
                }
            }, "BB").start();
        }
    }

*  **阻塞队列版**

**使用场景描述：**

在五秒钟内，完成阻塞队列数为 10(包括生产者、消费者)的交互，实现原子性操作保证线程之间交互的安全性

**代码思路：**

生产者和消费者之间要有一个标识来标注它们之间的交互是否完成，需要一个有界的阻塞队列来存储生产数据和消费数据的容器，避免一定的资源浪费，要保证过程一定是原子的，**注：消费者和生产者是同时启动的**

**代码示例：**

class MyResource {
        /** 默认开启 进行生产消费的交互 */
        private volatile boolean flag = true;
        /** 默认值是0 */
        private AtomicInteger atomicInteger = new AtomicInteger();
    
        private BlockingQueue<String> blockingQueue = null;
    
        public MyResource(BlockingQueue<String> blockingQueue) {
            this.blockingQueue = blockingQueue;
            System.out.println(blockingQueue.getClass().getName());
        }
    
        public void myProd() throws Exception {
            String data = null;
            boolean returnValue;
            while (flag) {
                // 原子操作 自增 1
                data = atomicInteger.incrementAndGet() + "";
                // 在 2s 之内插入队列,返回插入是否成功的结果
                returnValue = blockingQueue.offer(data, 2L, TimeUnit.SECONDS);
                if (returnValue) {
                    System.out.println(Thread.currentThread().getName() + "\t 插入队列数据" + data + "成功");
                } else {
                    System.out.println(Thread.currentThread().getName() + "\t 插入队列数据" + data + "失败");
                }
                TimeUnit.SECONDS.sleep(1);
            }
            System.out.println(Thread.currentThread().getName() + "\t 停止 表示 flag:" + flag);
        }
    
        public void myConsumer() throws Exception {
            String result = null;
            while (flag) {
                // 在 2s 内移除队列,返回移除是否成功的结果
                result = blockingQueue.poll(2L, TimeUnit.SECONDS);
                if(null==result||"".equalsIgnoreCase(result)){
                    flag=false;
                    System.out.println(Thread.currentThread().getName()+"\t"+"超过2m没有取到 消费退出");
                    System.out.println();
                    System.out.println();
                    return;
                }
                System.out.println(Thread.currentThread().getName() + "\t消费队列" + result + "成功");
            }
        }
        public void stop() throws Exception{
            flag=false;
        }
    }
    
    /**
     * volatile/CAS/atomicInteger/BlockQueue/线程交互/原子引用
     * @author vnjohn
     * @NAME: ProdConsumerBlockQueueDemo
     * @DATE: 2020/12/7
     */
    public class ProdConsumerBlockQueueDemo {
        public static void main(String[] args) throws Exception {
            MyResource myResource = new MyResource(new ArrayBlockingQueue<>(10));
            new Thread(()->{
                System.out.println(Thread.currentThread().getName()+"\t生产线程启动");
                try {
                    myResource.myProd();
                } catch (Exception e) {
                    e.printStackTrace();
                }
            },"Prod").start();
    
            new Thread(()->{
                System.out.println(Thread.currentThread().getName()+"\t消费线程启动");
                try {
                    myResource.myConsumer();
                } catch (Exception e) {
                    e.printStackTrace();
                }
            },"consumer").start();
            try { TimeUnit.SECONDS.sleep(5); } catch (InterruptedException e) { e.printStackTrace(); }
            System.out.println();
            System.out.println();
            System.out.println("时间到,停止活动");
            myResource.stop();
        }
    }

代码执行结果如下：

[![alt][alt 7]

![alt 7][]][alt_alt 7_ _alt 7]

##### 2) 线程池 #####

线程池相关的知识在后续文章给出~~

##### 3) 消息队列 #####

消息队列 RabbitMQ 相关知识后续文章给出~~

##### 参考文献 #####

[https://www.cnblogs.com/leesf456/p/5392816.html][https_www.cnblogs.com_leesf456_p_5392816.html]

[https://www.jianshu.com/p/ae6977886cec][https_www.jianshu.com_p_ae6977886cec]

#### *相关* ####

[alt]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/11/321174c6b97d4207bd60bc532a883288.png
[alt_ _alt]: https://i0.wp.com/www.vnjohn.com/wp-content/uploads/2020/12/image-4.png?ssl=1
[alt 1]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/11/af3c7903431741bcae08f5a75f42a597.png
[alt_alt 1_ _alt 1]: https://i0.wp.com/www.vnjohn.com/wp-content/uploads/2020/12/image-5.png?ssl=1
[alt 2]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/11/9846de7d203045dda03d9684dd1b98fd.png
[alt_alt 2_ _alt 2]: https://i0.wp.com/www.vnjohn.com/wp-content/uploads/2020/12/image-6.png?ssl=1
[alt 3]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/11/e3bc4a871ea84054a86061effa7a6ee4.png
[alt_alt 3_ _alt 3]: https://i0.wp.com/www.vnjohn.com/wp-content/uploads/2020/12/image-7.png?ssl=1
[alt 4]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/11/cc7eb0c8726f47ec98f133fdc7921799.png
[alt_alt 4_ _alt 4]: https://i0.wp.com/www.vnjohn.com/wp-content/uploads/2020/12/image-8.png?ssl=1
[alt 5]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/11/ba72f48ad8dc4812946efd62735d09b9.png
[alt_alt 5_ _alt 5]: https://i0.wp.com/www.vnjohn.com/wp-content/uploads/2020/12/image-9.png?ssl=1
[alt 6]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/11/e0a0aedb010241689825985788a42dd2.png
[alt_alt 6_ _alt 6]: https://i0.wp.com/www.vnjohn.com/wp-content/uploads/2020/12/image-10.png?ssl=1
[alt 7]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/11/26b0e9c724c04b82b253a5ba83b1c6bc.png
[alt_alt 7_ _alt 7]: https://i0.wp.com/www.vnjohn.com/wp-content/uploads/2020/12/image-11.png?ssl=1
[https_www.cnblogs.com_leesf456_p_5392816.html]: https://www.cnblogs.com/leesf456/p/5392816.html
[https_www.jianshu.com_p_ae6977886cec]: https://www.jianshu.com/p/ae6977886cec