2

一文带你了解J.U.C的FutureTask、Fork/Join框架和BlockingQueue

 1 year ago
source link: https://blog.51cto.com/u_15214399/5373095
Go to the source link to view the article. You can view the picture content, updated content and better typesetting reading experience. If the link is broken, please click the button below to view the snapshot at that time.

摘要： J.U.C是Java并发编程中非常重要的工具包，今天，我们就来着重讲讲J.U.C里面的FutureTask、Fork/Join框架和BlockingQueue。

本文分享自华为云社区《​ ​【高并发】J.U.C组件扩展​​》，作者： 冰 河 。

FutureTask

FutureTask是J.U.C(java.util.concurrent)下的，但不是AQS(AbstractQueuedSynchronizer)的子类。其对线程结果的处理值得借鉴和在项目中使用。

Thread和Runnable执行完任务无法获取执行结果。Java1.5开始提供了Callable和Future，通过它们可以在任务执行完毕之后，得到任务执行的结果。

Callable与Runnable接口对比

Callable：泛型接口，提供一个call()方法，支持抛出异常，并且执行后有返回值
Runnable：接口，提供一个run()方法，不支持抛出异常，执行后无返回值

Future接口

对于具体的Callable和Runnable任务，可以进行取消，查询任务是否被取消，查询是否完成以及获取结果等。

Future可以监视目标线程调用call()的情况，当调用Future的get()方法时，就可以获得结果。此时，执行任务的线程可能不会直接完成，当前线程就开始阻塞，直到call()方法结束返回结果，当前线程才会继续执行。总之，Future可以得到别的线程任务方法的返回值。

FutureTask类

实现的接口为RunnableFuture，而RunnableFuture接口继承了Runnable和Future两个接口，所以FutureTask类最终也是执行Callable类型的任务。如果FutureTask类的构造方法参数是Runnable的话，会转换成Callable类型。

类实现了两个接口：Runnable和Future。所以，它即可以作为Runnable被线程执行，又可以作为Future得到Callable的返回值，这样设计的好处如下：

假设有一个很费时的逻辑，需要计算并且返回这个值，同时，这个值又不是马上需要，则可以使用Runnable和Future的组合，用另外一个线程去计算返回值，而当前线程在使用这个返回值之前，可以做其他的操作，等到需要这个返回值时，再通过Future得到。

Future示例代码如下：

package io.binghe.concurrency.example.aqs;

import lombok.extern.slf4j.Slf4j;
import java.util.concurrent.Callable;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;
@Slf4j
public class FutureExample {
    static class MyCallable implements Callable<String>{
        @Override
        public String call() throws Exception {
            log.info("do something in callable");
            Thread.sleep(5000);
            return "Done";
        }
    }
    public static void main(String[] args) throws Exception {
        ExecutorService executorService = Executors.newCachedThreadPool();
        Future<String> future = executorService.submit(new MyCallable());
        log.info("do something in main");
        Thread.sleep(1000);
        String result = future.get();
        log.info("result: {}", result);
        executorService.shutdown();
    }
}



FutureTask示例代码如下：



package io.binghe.concurrency.example.aqs;

import lombok.extern.slf4j.Slf4j;
import java.util.concurrent.Callable;
import java.util.concurrent.FutureTask;
@Slf4j
public class FutureTaskExample {
    public static void main(String[] args) throws Exception{
        FutureTask<String> futureTask = new FutureTask<String>(new Callable<String>() {
            @Override
            public String call() throws Exception {
                log.info("do something in callable");
                Thread.sleep(5000);
                return "Done";
            }
        });
        new Thread(futureTask).start();
        log.info("do something in main");
        Thread.sleep(1000);
        String result = futureTask.get();
        log.info("result: {}", result);
    }
}



Fork/Join框架



位于J.U.C(java.util.concurrent)中，是Java7中提供的用于执行并行任务的框架，其可以将大任务分割成若干个小任务，最终汇总每个小任务的结果后得到最终结果。基本思想和Hadoop的MapReduce思想类似。



主要采用的是工作窃取算法（某个线程从其他队列里窃取任务来执行），并行分治计算中的一种Work-stealing策略



为什么需要使用工作窃取算法呢？



假如我们需要做一个比较大的任务，我们可以把这个任务分割为若干互不依赖的子任务，为了减少线程间的竞争，于是把这些子任务分别放到不同的队列里，并为每个队列创建一个单独的线程来执行队列里的任务，线程和队列一一对应，比如A线程负责处理A队列里的任务。但是有的线程会先把自己队列里的任务干完，而其他线程对应的队列里还有任务等待处理。干完活的线程与其等着，不如去帮其他线程干活，于是它就去其他线程的队列里窃取一个任务来执行。而在这时它们会访问同一个队列，所以为了减少窃取任务线程和被窃取任务线程之间的竞争，通常会使用双端队列，被窃取任务线程永远从双端队列的头部拿任务执行，而窃取任务的线程永远从双端队列的尾部拿任务执行。



工作窃取算法的优点：



充分利用线程进行并行计算，并减少了线程间的竞争



工作窃取算法的缺点：



在某些情况下还是存在竞争，比如双端队列里只有一个任务时。并且该算法会消耗更多的系统资源，比如创建多个线程和多个双端队列。



对于Fork/Join框架而言，当一个任务正在等待它使用Join操作创建的子任务结束时，执行这个任务的工作线程查找其他未被执行的任务，并开始执行这些未被执行的任务，通过这种方式，线程充分利用它们的运行时间来提高应用程序的性能。为了实现这个目标，Fork/Join框架执行的任务有一些局限性。



Fork/Join框架局限性：



（1）任务只能使用Fork和Join操作来进行同步机制，如果使用了其他同步机制，则在同步操作时，工作线程就不能执行其他任务了。比如，在Fork/Join框架中，使任务进行了睡眠，那么，在睡眠期间内，正在执行这个任务的工作线程将不会执行其他任务了。
（2）在Fork/Join框架中，所拆分的任务不应该去执行IO操作，比如：读写数据文件
（3）任务不能抛出检查异常，必须通过必要的代码来出来这些异常



Fork/Join框架的核心类



Fork/Join框架的核心是两个类：ForkJoinPool和ForkJoinTask。ForkJoinPool负责实现工作窃取算法、管理工作线程、提供关于任务的状态以及执行信息。ForkJoinTask主要提供在任务中执行Fork和Join操作的机制。



示例代码如下：



package io.binghe.concurrency.example.aqs;

import lombok.extern.slf4j.Slf4j;
import java.util.concurrent.ForkJoinPool;
import java.util.concurrent.Future;
import java.util.concurrent.RecursiveTask;
@Slf4j
public class ForkJoinTaskExample extends RecursiveTask<Integer> {
    public static final int threshold = 2;
    private int start;
    private int end;
    public ForkJoinTaskExample(int start, int end) {
        this.start = start;
        this.end = end;
    }
    @Override
    protected Integer compute() {
        int sum = 0;
        //如果任务足够小就计算任务
        boolean canCompute = (end - start) <= threshold;
        if (canCompute) {
            for (int i = start; i <= end; i++) {
                sum += i;
            }
        } else {
            // 如果任务大于阈值，就分裂成两个子任务计算
            int middle = (start + end) / 2;
            ForkJoinTaskExample leftTask = new ForkJoinTaskExample(start, middle);
            ForkJoinTaskExample rightTask = new ForkJoinTaskExample(middle + 1, end);

// 执行子任务
            leftTask.fork();
            rightTask.fork();

// 等待任务执行结束合并其结果
            int leftResult = leftTask.join();
            int rightResult = rightTask.join();

// 合并子任务
            sum = leftResult + rightResult;
        }
        return sum;
    }
    public static void main(String[] args) {
        ForkJoinPool forkjoinPool = new ForkJoinPool();

//生成一个计算任务，计算1+2+3+4
        ForkJoinTaskExample task = new ForkJoinTaskExample(1, 100);

//执行一个任务
        Future<Integer> result = forkjoinPool.submit(task);

try {
            log.info("result:{}", result.get());
        } catch (Exception e) {
            log.error("exception", e);
        }
    }
}



BlockingQueue



阻塞队列，是线程安全的。



被阻塞的情况如下：



（1）当队列满时，进行入队列操作
（2）当队列空时，进行出队列操作



使用场景如下：



主要在生产者和消费者场景



BlockingQueue的方法



BlockingQueue 具有 4 组不同的方法用于插入、移除以及对队列中的元素进行检查。如果请求的操作不能得到立即执行的话，每个方法的表现也不同。这些方法如下：



一文带你了解J.U.C的FutureTask、Fork/Join框架和BlockingQueue_J.U.C



四组不同的行为方式解释:



如果试图的操作无法立即执行，抛一个异常。



如果试图的操作无法立即执行，返回一个特定的值(常常是 true / false)。



如果试图的操作无法立即执行，该方法调用将会发生阻塞，直到能够执行。



如果试图的操作无法立即执行，该方法调用将会发生阻塞，直到能够执行，但等待时间不会超过给定值。返回一个特定值以告知该操作是否成功(典型的是 true / false)。



BlockingQueue的实现类如下：



  • ArrayBlockingQueue：有界的阻塞队列（容量有限，必须在初始化的时候指定容量大小，容量大小指定后就不能再变化），内部实现是一个数组，以FIFO的方式存储数据，最新插入的对象是尾部，最新移除的对象是头部。
  • DelayQueue：阻塞的是内部元素，DelayQueue中的元素必须实现一个接口——Delayed（存在于J.U.C下）。Delayed接口继承了Comparable接口，这是因为Delayed接口中的元素需要进行排序，一般情况下，都是按照Delayed接口中的元素过期时间的优先级进行排序。应用场景主要有：定时关闭连接、缓存对象、超时处理等。内部实现使用PriorityQueue和ReentrantLock。
  • LinkedBlockingQueue：大小配置是可选的，如果初始化时指定了大小，则是有边界的；如果初始化时未指定大小，则是无边界的（其实默认大小是Integer类型的最大值）。内部实现时一个链表，以FIFO的方式存储数据，最新插入的对象是尾部，最新移除的对象是头部。
  • PriorityBlockingQueue：带优先级的阻塞队列，无边界，但是有排序规则，允许插入空对象(也就是null)。所有插入的对象必须实现Comparable接口，队列优先级的排序规则就是按照对Comparable接口的实现来定义的。可以从PriorityBlockingQueue中获得一个迭代器Iterator，但这个迭代器并不保证按照优先级的顺序进行迭代。
  • SynchronousQueue：队列内部仅允许容纳一个元素，当一个线程插入一个元素后，就会被阻塞，除非这个元素被另一个线程消费。因此，也称SynchronousQueue为同步队列。SynchronousQueue是一个无界非缓存的队列。准确的说，它不存储元素，放入元素只有等待取走元素之后，才能再次放入元素



​ ​点击关注，第一时间了解华为云新鲜技术~​​

        

        

            

                report
            

        

    







        
Recommend

        



                

    
About Joyk

    
 





Aggregate valuable and interesting links.

Joyk means Joy of geeK