JUC必要掌握（Callable&Future、JUC 三大辅助类、阻塞队列）

水深无声 2023-09-28 14:09 40阅读 0赞

#### 1. Callable&Future 接口 ####

1.1 Callable

记得刚开始学习多线程的时候，只学习了创建线程的2种方式，一种是直接继承Thread，另外一种就是实现Runnable接口。这2种方式都有一个缺陷就是：在执行完任务之后无法获取执行结果。

如果需要获取执行结果，就必须通过共享变量或者使用线程通信的方式来达到效果，这样使用起来就比较麻烦。

Callable是一个接口，一个函数式接口，也是个泛型接口。call（）有返回值，且返回值类型与泛型参数类型相同，且可以抛出异常。Callable可以看作是Runnable接口的补充。

1.2 Future

当 call（）方法完成时，结果必须存储在主线程已知的对象中，以便主线程可 以知道该线程返回的结果。为此，可以使用 Future 对象。 将 Future 视为保存结果的对象–它可能暂时不保存结果，但将来会保存（一旦 Callable 返回）。

Future 基本上是主线程可以跟踪进度以及其他线程的结果的 一种方式。要实现此接口，必须重写 5 种方法，这里列出了重要的方法,如下:

public boolean cancel（boolean mayInterrupt）：用于停止任务。
    ==如果尚未启动，它将停止任务。如果已启动，则仅在 mayInterrupt 为 true
    时才会中断任务。==
     
    • public Object get（）抛出 InterruptedException，ExecutionException：
    用于获取任务的结果。
    ==如果任务完成，它将立即返回结果，否则将等待任务完成，然后返回结果。==
     
    • public boolean isDone（）：如果任务完成，则返回 true，否则返回 false
    可以看到 Callable 和 Future 做两件事-Callable 与 Runnable 类似，因为它封
    装了要在另一个线程上运行的任务，而 Future 用于存储从另一个线程获得的结
    果。实际上，future 也可以与 Runnable 一起使用。
    要创建线程，需要 Runnable。为了获得结果，需要 future。

1.3 FutureTask

该类提供了Future的基本实现，其中包含启动和取消计算的方法、查看计算是否完成的查询以及检索计算结果的方法。只有当计算完成时才能检索到结果;如果计算还没有完成，get方法将阻塞。一旦计算完成，就不能重新启动或取消计算(除非使用runAndReset调用计算)。

FutureTask可用于包装Callable或Runnable对象。因为FutureTask实现了Runnable，所以FutureTask可以提交给Executor执行。

下面是其包含的方法：

Modifier and Type	Method and Description
        boolean	cancel(boolean mayInterruptIfRunning)
    
        尝试取消执行此任务。
        protected void	done()
    
        此任务转换到状态 isDone （无论是正常还是通过取消）调用的受保护方法。
        V	get()
    
        等待计算完成，然后检索其结果。
        V	get(long timeout, TimeUnit unit)
    
        如果需要等待最多在给定的时间计算完成，然后检索其结果（如果可用）。
        boolean	isCancelled()
    
        如果此任务在正常完成之前取消，则返回 true 。
        boolean	isDone()
    
        返回 true如果任务已完成。
        void	run()
    
        将此未来设置为其计算结果，除非已被取消。
        protected boolean	runAndReset()
    
        执行计算而不设置其结果，然后将此将来重置为初始状态，如果计算遇到异常或被取消，则不执行此操作。
        protected void	set(V v)
    
        将此未来的结果设置为给定值，除非此未来已被设置或已被取消。
        protected void	setException(Throwable t)
    
        导致这个未来报告一个ExecutionException与给定的可抛弃的原因，除非这个未来已经被设置或被取消。

1.4 话不多说直接上demo（CV运行，感受一下）

1.4.1 普通小demo，拿去即可运行

package com.example.onlyqi.juc;
     
    import java.util.concurrent.Callable;
    import java.util.concurrent.ExecutionException;
    import java.util.concurrent.FutureTask;
     
    /**
     * @author onlyqi
     * @date 2022/5/31 10:16
     * @description
     */
    public class Test01 {
        public static void main(String[] args) throws ExecutionException, InterruptedException {
            FutureTask futureTask=new FutureTask(new Callable() {
                @Override
                public String call() throws Exception {
                    Thread.sleep(300);
                    System.out.println("calld01方法执行了");
                    return "call方法返回值01";
                }
            });
            futureTask.run();
            FutureTask futureTask1=new FutureTask(new Callable() {
                @Override
                public String call() throws Exception {
                    Thread.sleep(300);
                    System.out.println("calld02方法执行了");
                    return "call方法返回值02";
                }
            });
            futureTask1.run();
            System.out.println("获取返回值: " + futureTask.get());
            System.out.println("获取返回值1: " + futureTask1.get());
        }
    }

运行结果：

1.4.2 线程池方式

CallableTask：

package com.example.onlyqi.juc.CallableTask;
    import java.util.ArrayList;
    import java.util.List;
    import java.util.concurrent.Callable;
     
    /**
     * @author onlyqi
     * @date 2022/5/31 11:23
     * @description
     */
    public class CallableTask implements Callable<List<Integer>> {
    //   @Resource
    //    private MemberMapper memberMapper;
        private Integer num;
     
        public CallableTask(Integer num) {
            this.num = num;
        }
     
        public CallableTask() {
        }
     
        @Override
        public List<Integer> call() throws Exception {
            List<Integer>list=new ArrayList<>();
            for (Integer i = 0; i < num; i++) {
                //*************memberMappe的操作。。。。写自己的业务逻辑**********//
                list.add(i);
            }
            return list;
        }
    }
    
    package com.example.onlyqi.juc.CallableTask;
     
    import java.util.ArrayList;
    import java.util.List;
    import java.util.concurrent.*;
     
    /**
     * @author onlyqi
     * @date 2022/5/31 11:26
     * @description
     */
    public class Task {
        public static void main(String[] args) {
            CallableTask callableTask1 =new CallableTask(5);
            CallableTask callableTask2 =new CallableTask(10);
            CallableTask callableTask3 =new CallableTask(20);
     
            FutureTask<List<Integer>> futureTask = new FutureTask<>(callableTask1);
            FutureTask<List<Integer>> futureTask2= new FutureTask<>(callableTask2);
            FutureTask<List<Integer>> futureTask3= new FutureTask<>(callableTask3);
     
            List<FutureTask<List<Integer>>>listTask=new ArrayList<>();
            listTask.add(futureTask);
            listTask.add(futureTask2);
            listTask.add(futureTask3);
            ThreadPoolExecutor memberGradeThreadPoolExecutor   = new ThreadPoolExecutor(3, 10, 10, TimeUnit.SECONDS, new LinkedBlockingQueue<>(10000));
            for (FutureTask<List<Integer>> listFutureTask : listTask) {
                memberGradeThreadPoolExecutor.execute(listFutureTask);
            }
            listTask.forEach(s-> {
                try {
                    System.out.println("========"+s.get());
                } catch (InterruptedException e) {
                    e.printStackTrace();
                } catch (ExecutionException e) {
                    e.printStackTrace();
                }
            });
        }
    }

运行结果：

#### 2. JUC 三大辅助类 ####

2.1 减少计数 CountDownLatch

允许一个或多个线程等待直到在其他线程中执行的一组操作完成的同步辅助。

A CountDownLatch用给定的计数初始化。 await方法阻塞，直到由于countDown()方法的调用而导致当前计数达到零，之后所有等待线程被释放，并且任何后续的await 调用立即返回。 这是一个一次性的现象 - 计数无法重置。 如果您需要重置计数的版本，请考虑使用CyclicBarrier 。

A CountDownLatch是一种通用的同步工具，可用于多种用途。 一个CountDownLatch为一个计数的CountDownLatch用作一个简单的开/关锁存器，或者门：所有线程调用await在门口等待，直到被调用countDown()的线程打开。 一个CountDownLatch初始化N可以用来做一个线程等待，直到N个线程完成某项操作，或某些动作已经完成N次。

CountDownLatch一个有用的属性是，它不要求调用countDown线程等待计数到达零之前继续，它只是阻止任何线程通过await ，直到所有线程可以通过。

示例用法：这是一组类，其中一组工作线程使用两个倒计时锁存器：

第一个是启动信号，防止任何工作人员进入，直到驾驶员准备好继续前进;
    第二个是完成信号，允许司机等到所有的工作人员完成。

上代码，拿去即可运行，感受一下 ：

package com.example.onlyqi.juc.fuzhu;
     
    import java.util.concurrent.CountDownLatch;
     
    /**
     * @author onlyqi
     * @date 2022/5/31 14:32
     * @description
     */
    public class Test01 {
        /**
         * 6 个同学陆续离开教室后值班同学才可以关门
         * @param args
         */
        public static void main(String[] args) throws Exception{
            //定义一个数值为 6 的计数器
            CountDownLatch countDownLatch = new CountDownLatch(6);
            //创建 6 个同学
            for (int i = 1; i <= 6; i++) {
                new Thread(() ->{
                    try{
                        if(Thread.currentThread().getName().equals("同学 6")){
                            Thread.sleep(2000);
                        }
                        System.out.println(Thread.currentThread().getName() + "离开了");
                        //计数器减一,不会阻塞
                        countDownLatch.countDown();
                    }catch (Exception e){
                        e.printStackTrace();
                    }
                }, "同学" + i).start();
            }
            //主线程 await 休息
            System.out.println("主线程睡觉");
            countDownLatch.await();
            //全部离开后自动唤醒主线程
            System.out.println("全部离开了,现在的计数器为" + countDownLatch.getCount());
        }
     
    }

运行结果：

2.2 循环栅栏 CyclicBarrier

允许一组线程全部等待彼此达到共同屏障点的同步辅助。 循环阻塞在涉及固定大小的线程方的程序中很有用，这些线程必须偶尔等待彼此。 屏障被称为循环 ，因为它可以在等待的线程被释放之后重新使用。

CyclicBarrier 的构造方法第一个参数是目标障碍数，每次执行 CyclicBarrier 一 次障碍数会加一，如果达到了目标障碍数，才会执行 cyclicBarrier.await()之后 的语句。可以将 CyclicBarrier 理解为加 1 操作

上代码，拿去即可运行，感受一下 ：

package com.example.onlyqi.juc.fuzhu;
     
    import java.util.concurrent.CyclicBarrier;
     
    /**
     * @author onlyqi
     * @date 2022/5/31 14:45
     * @description
     */
    public class Test02 {
        //定义神龙召唤需要的龙珠总数
        private final static int NUMBER = 7;
        /**
         * 集齐 7 颗龙珠就可以召唤神龙
         * @param args
         */
        public static void main(String[] args) {
            //定义循环栅栏
            CyclicBarrier cyclicBarrier = new CyclicBarrier(NUMBER, () ->{
                System.out.println("集齐" + NUMBER + "颗龙珠,现在召唤神龙!!!!!!!!!");
            });
            //定义 7 个线程分别去收集龙珠
            for (int i = 1; i <= 7; i++) {
                new Thread(()->{
                    try {
                        if(Thread.currentThread().getName().equals("龙珠3号")){
                            System.out.println("龙珠 3 号抢夺战开始,孙悟空开启超级赛亚人模式!");
                            Thread.sleep(5000);
                            System.out.println("龙珠 3 号抢夺战结束,孙悟空打赢了,拿到了龙珠 3 号!");
                        }else{
                            System.out.println(Thread.currentThread().getName() + "收集到 了!!!!");
                        }
                        cyclicBarrier.await();
                    }catch (Exception e){
                        e.printStackTrace();
                    }
                }, "龙珠" + i + "号").start();
            }
        }
     
    }

运行结果：

2.3 信号灯 Semaphore

一个计数信号量。 在概念上，信号量维持一组许可证。 如果有必要，每个acquire()都会阻塞，直到许可证可用，然后才能使用它。 每个release()添加许可证，潜在地释放阻塞获取方。 但是，没有使用实际的许可证对象; Semaphore只保留可用数量的计数，并相应地执行。

Semaphore 的构造方法中传入的第一个参数是最大信号量（可以看成最大线 程池），每个信号量初始化为一个最多只能分发一个许可证。使用 acquire 方 法获得许可证，release 方法释放许可。

上代码，拿去即可运行，感受一下 ：

package com.example.onlyqi.juc.fuzhu;
     
    import java.util.concurrent.Semaphore;
     
    /**
     * @author onlyqi
     * @date 2022/5/31 14:54
     * @description
     */
    public class Test03 {
        /**
         * 抢车位, 10 部汽车 1 个停车位
         * @param args
         */
        public static void main(String[] args) throws Exception{
            //定义 3 个停车位
            Semaphore semaphore = new Semaphore(1);
            //模拟 6 辆汽车停车
            for (int i = 1; i <= 10; i++) {
                Thread.sleep(100);
                //停车
                new Thread(() ->{
                    try {
                        System.out.println(Thread.currentThread().getName() + "找车位 ing");
                        semaphore.acquire();
                        System.out.println(Thread.currentThread().getName() + "汽车停车成功!");
                        Thread.sleep(10000);
                    }catch (Exception e){
                        e.printStackTrace();
                    }finally {
                        System.out.println(Thread.currentThread().getName() + "溜了溜了");
                        semaphore.release();
                    }
                }, "汽车" + i).start();
            }
        }
    }

运行结果：

#### 3. 阻塞队列 ####

3.1 BlockingQueue 概述

当阻塞队列是空时，从队列中获取元素的操作将会被阻塞
    当阻塞队列是满时，往队列里添加元素的操作将会被阻塞

换言之：

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

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

那为什么需要BlockingQueue呢？

在concurrent包发布以前，在多线程环境下，我们都必须自己去控制这些阻塞/唤醒的细节，特别是还得兼顾效率和线程安全问题，这将会对于我们的编码开发带来不少的复杂度。

而阻塞队列带来的好处就是我们不需要去关心什么时候需要阻塞线程，又什么时候需要去唤醒线程，因为这一切BlockingQueue都自行一并处理。

BlockingQueue和JDK集合包中的Queue接口兼容，同时在其基础上增加了阻塞功能。 3.2 BlockingQueue常用方法示例

offer、put、poll和take

入队：

（1）offer(E e)：如果队列没满，返回true，如果队列已满，返回false（不阻塞）

（2）offer(E e, long timeout, TimeUnit unit)：可以设置阻塞时间，如果队列已满，则进行阻

塞。超过阻塞时间，则返回false

（3）put(E e)：队列没满的时候是正常的插入，如果队列已满，则阻塞，直至队列空出位置

出队：

（1）poll()：如果有数据，出队，如果没有数据，返回null （不阻塞）

（2）poll(long timeout, TimeUnit unit)：可以设置阻塞时间，如果没有数据，则阻塞，超过

阻塞时间，则返回null

（3）take()：队列里有数据会正常取出数据并删除；但是如果队列里无数据，则阻塞，直到队

列里有数据 add、remove、element

add 方法

add 方法是往队列里添加一个元素，如果队列满了，就会抛出异常来提示队列已满。示例代码如下：

private static void addTest() \{ BlockingQueue blockingQueue = new ArrayBlockingQueue(2); blockingQueue.add(1); blockingQueue.add(1); blockingQueue.add(1); \}

在这段代码中，我们创建了一个容量为 2 的 BlockingQueue，并且尝试往里面放 3 个值，超过了容量上限，那么在添加第三个值的时候就会得到异常：

Exception in thread "main" java.lang.IllegalStateException:Queue full

remove 方法

remove 方法的作用是删除元素，如果我们删除的队列是空的，由于里面什么都没有，所以也无法删除任何元素，那么 remove 方法就会抛出异常。示例代码如下：

private static void removeTest() \{ ArrayBlockingQueue blockingQueue = new ArrayBlockingQueue(2); blockingQueue.add(1); blockingQueue.add(1); blockingQueue.remove(); blockingQueue.remove(); blockingQueue.remove(); \}

在这段代码中，我们往一个容量为 2 的 BlockingQueue 里放入 2 个元素，并且删除 3 个元素。在删除前面两个元素的时候会正常执行，因为里面依然有元素存在，但是在删除第三个元素时，由于队列里面已经空了，所以便会抛出异常：

Exception in thread "main" java.util.NoSuchElementException

element 方法

element 方法是返回队列的头部节点，但是并不删除。和 remove 方法一样，如果我们用这个方法去操作一个空队列，想获取队列的头结点，可是由于队列是空的，我们什么都获取不到，会抛出和前面 remove 方法一样的异常：NoSuchElementException。示例代码如下：

private static void elementTest() \{ ArrayBlockingQueue blockingQueue = new ArrayBlockingQueue(2); blockingQueue.element(); \}

offer、poll、peek

方法

抛出异常

返回特定值

阻塞

阻塞特定时间

入队

add(e)

offer(e)

put(e)

offer(e, time, unit)

出队

remove()

poll()

take()

poll(time, unit)

获取队首元素

element()

peek()

不支持

3.3 上代码，拿去即可运行，感受一下 ：

package com.example.onlyqi.block;
     
    import java.util.concurrent.ArrayBlockingQueue;
    import java.util.concurrent.BlockingQueue;
    import java.util.concurrent.TimeUnit;
     
    /**
     * @author onlyqi
     * @date 2022/5/31 17:05
     * @description
     */
    public class Block {
        public static void main(String[] args) throws InterruptedException {
    // List list = new ArrayList();
            BlockingQueue<String> blockingQueue = new ArrayBlockingQueue<>(3);
    //第一组
     System.out.println(blockingQueue.add("a"));
     System.out.println(blockingQueue.add("b"));
     System.out.println(blockingQueue.add("c"));
     System.out.println(blockingQueue.element());
     System.out.println(blockingQueue.add("x"));
     System.out.println(blockingQueue.remove());
     System.out.println(blockingQueue.remove());
     System.out.println(blockingQueue.remove());
    // System.out.println(blockingQueue.remove());
    // 第二组
    // System.out.println(blockingQueue.offer("a"));
    // System.out.println(blockingQueue.offer("b"));
    // System.out.println(blockingQueue.offer("c"));
    // System.out.println(blockingQueue.offer("x"));
    // System.out.println(blockingQueue.poll());
    // System.out.println(blockingQueue.poll());
    // System.out.println(blockingQueue.poll());
    // System.out.println(blockingQueue.poll());
    // 第三组
    // blockingQueue.put("a");
    // blockingQueue.put("b");
    // blockingQueue.put("c");
    // //blockingQueue.put("x");
    // System.out.println(blockingQueue.take());
    // System.out.println(blockingQueue.take());
    // System.out.println(blockingQueue.take());
    // System.out.println(blockingQueue.take());
    // 第四组
    //        System.out.println(blockingQueue.offer("a"));
    //        System.out.println(blockingQueue.offer("b"));
    //        System.out.println(blockingQueue.offer("c"));
    //        System.out.println("-----------"+blockingQueue.peek());
    //        System.out.println(blockingQueue.offer("a",3L, TimeUnit.SECONDS));
        }
    }

运行结果：

3.4 常见的阻塞队列

BlockingQueue 接口的实现类都被放在了 J.U.C 包中，常见的和常用的实现类包括 ArrayBlockingQueue、LinkedBlockingQueue、SynchronousQueue、PriorityBlockingQueue，以及 DelayQueue。

3.4.1 ArrayBlockingQueue

ArrayBlockingQueue 是最典型的有界队列，其内部是用数组存储元素的，利用 ReentrantLock 实现线程安全。

在创建它的时候就需要指定它的容量，之后也不可以再扩容了，在构造函数中我们同样可以指定是否是公平的，代码如下：

ArrayBlockingQueue(int capacity, boolean fair)

第一个参数是容量，第二个参数是是否公平。正如 ReentrantLock 一样，如果 ArrayBlockingQueue 被设置为非公平的，那么就存在插队的可能；如果设置为公平的，那么等待了最长时间的线程会被优先处理，其他线程不允许插队，不过这样的公平策略同时会带来一定的性能损耗，因为非公平的吞吐量通常会高于公平的情况。

3.4.2 LinkedBlockingQueue

这是一个内部用链表实现的 BlockingQueue。如果我们不指定它的初始容量，那么它容量默认就为整型的最大值 Integer.MAX\_VALUE，由于这个数非常大，通常不可能放入这么多的数据，所以 LinkedBlockingQueue 也被称作无界队列，代表它几乎没有界限。

SynchronousQueue

SynchronousQueue 最大的不同之处在于，它的容量为 0，所以没有一个地方来暂存元素，导致每次取数据都要先阻塞，直到有数据被放入；同理，每次放数据的时候也会阻塞，直到有消费者来取。

SynchronousQueue 的容量不是 1 而是 0，因为 SynchronousQueue 不需要去持有元素，它所做的就是直接传递（direct handoff）。由于每当需要传递的时候，SynchronousQueue 会把元素直接从生产者传给消费者，在此期间并不需要做存储，所以如果运用得当，它的效率是很高的。

由于它的容量为 0，所以相比于一般的阻塞队列，SynchronousQueue 的很多方法的实现是很有意思的，我们来举几个例子：

SynchronousQueue 的 peek 方法永远返回 null，代码如下：

public E peek() \{ return null; \}

因为 peek 方法的含义是取出头结点，但是 SynchronousQueue 的容量是 0，所以连头结点都没有，peek 方法也就没有意义，所以始终返回 null。同理，element 始终会抛出 NoSuchElementException 异常。

而 SynchronousQueue 的 size 方法始终返回 0，因为它内部并没有容量，代码如下：

public int size() \{ return 0; \}

直接 return 0，同理，isEmpty 方法始终返回 true：

public boolean isEmpty() \{ return true; \}

因为它始终都是空的

3.4.3 PriorityBlockingQueue

ArrayBlockingQueue 和 LinkedBlockingQueue 都是采用先进先出的顺序进行排序，可是如果有的时候我们需要自定义排序怎么办呢？这时就需要使用 PriorityBlockingQueue。

PriorityBlockingQueue 是一个支持优先级的无界阻塞队列，可以通过自定义类实现 compareTo() 方法来指定元素排序规则，或者初始化时通过构造器参数 Comparator 来指定排序规则。同时，插入队列的对象必须是可比较大小的，也就是 Comparable 的，否则会抛出 ClassCastException 异常。

它的 take 方法在队列为空的时候会阻塞，但是正因为它是无界队列，而且会自动扩容，所以它的队列永远不会满，所以它的 put 方法永远不会阻塞，添加操作始终都会成功，也正因为如此，它的成员变量里只有一个 Condition：

private final Condition notEmpty;

这和之前的 ArrayBlockingQueue 拥有两个 Condition（分别是 notEmpty 和 notFull）形成了鲜明的对比，我们的 PriorityBlockingQueue 不需要 notFull，因为它永远都不会满，真是“有空间就可以任性”。

3.4.4 DelayQueue

DelayQueue 这个队列比较特殊，具有“延迟”的功能。我们可以设定让队列中的任务延迟多久之后执行，比如 10 秒钟之后执行，这在例如“30 分钟后未付款自动取消订单”等需要延迟执行的场景中被大量使用。

它是无界队列，放入的元素必须实现 Delayed 接口，而 Delayed 接口又继承了 Comparable 接口，所以自然就拥有了比较和排序的能力，代码如下：

public interface Delayed extends Comparable<Delayed> {
        long getDelay(TimeUnit unit);
    }

可以看出这个 Delayed 接口继承自 Comparable，里面有一个需要实现的方法，就是 getDelay。这里的 getDelay 方法返回的是“还剩下多长的延迟时间才会被执行”，如果返回 0 或者负数则代表任务已过期。

元素会根据延迟时间的长短被放到队列的不同位置，越靠近队列头代表越早过期。

DelayQueue 内部使用了 PriorityQueue 的能力来进行排序，而不是自己从头编写，我们在工作中可以学习这种思想，对已有的功能进行复用，不但可以

![641ccf4b7689442783fbd2e1746b3e8e.jpeg][]

[641ccf4b7689442783fbd2e1746b3e8e.jpeg]: https://img-blog.csdnimg.cn/641ccf4b7689442783fbd2e1746b3e8e.jpeg