Java实现FutureTask的示例详解

目录
  • 前言
  • FutureTask
  • 自己实现FutureTask
    • 工具准备
    • FutureTask设计与实现
  • 总结

前言

在并发编程当中我们最常见的需求就是启动一个线程执行一个函数去完成我们的需求,而在这种需求当中,我们常常需要函数有返回值。比如我们需要同一个非常大的数组当中数据的和,让每一个线程求某一个区间内部的和,最终将这些和加起来,那么每个线程都需要返回对应区间的和。而在Java当中给我们提供了这种机制,去实现这一个效果——FutureTask

FutureTask

在自己写FutureTask之前我们首先写一个例子来回顾一下FutureTask的编程步骤:

写一个类实现Callable接口。

@FunctionalInterface
public interface Callable<V> {
    /**
     * Computes a result, or throws an exception if unable to do so.
     *
     * @return computed result
     * @throws Exception if unable to compute a result
     */
    V call() throws Exception;
}

实现接口就实现call即可,可以看到这个函数是有返回值的,而FutureTask返回给我们的值就是这个函数的返回值。

new一个FutureTask对象,并且new一个第一步写的类,new FutureTask<>(callable实现类)

最后将刚刚得到的FutureTask对象传入Thread类当中,然后启动线程即可new Thread(futureTask).start();

然后我们可以调用FutureTaskget方法得到返回的结果futureTask.get();

假如有一个数组data,长度为100000,现在有10个线程,第i个线程求数组[i * 10000, (i + 1) * 10000)所有数据的和,然后将这十个线程的结果加起来。

import java.lang.reflect.Array;
import java.util.Arrays;
import java.util.Random;
import java.util.concurrent.Callable;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.FutureTask;

public class FutureTaskDemo {

  public static void main(String[] args) throws ExecutionException, InterruptedException {
    int[] data = new int[100000];
    Random random = new Random();
    for (int i = 0; i < 100000; i++) {
      data[i] = random.nextInt(10000);
    }
    @SuppressWarnings("unchecked")
    FutureTask<Integer>[] tasks = (FutureTask<Integer>[]) Array.newInstance(FutureTask.class, 10);
    // 设置10个 futuretask 任务计算数组当中数据的和
    for (int i = 0; i < 10; i++) {
      int idx = i;
      tasks[i] = new FutureTask<>(() -> {
        int sum = 0;
        for (int k = idx * 10000; k < (idx + 1) * 10000; k++) {
          sum += data[k];
        }
        return sum;
      });
    }
    // 开启线程执行 futureTask 任务
    for (FutureTask<Integer> futureTask : tasks) {
      new Thread(futureTask).start();
    }
    int threadSum = 0;
    for (FutureTask<Integer> futureTask : tasks) {
      threadSum += futureTask.get();
    }
    int sum = Arrays.stream(data).sum();
    System.out.println(sum == threadSum); // 结果始终为 true
  }
}

可能你会对FutureTask的使用方式感觉困惑,或者不是很清楚,现在我们来仔细捋一下思路。

首先启动一个线程要么是继承自Thread类,然后重写Thread类的run方法,要么是给Thread类传递一个实现了Runnable的类对象,当然可以用匿名内部类实现。

既然我们的FutureTask对象可以传递给Thread类,说明FutureTask肯定是实现了Runnable接口,我们现在来看一下FutureTask的继承体系。

​ 可以发现的是FutureTask确实实现了Runnable接口,同时还实现了Future接口,这个Future接口主要提供了后面我们使用FutureTask的一系列函数比如get

看到这里你应该能够大致想到在FutureTask中的run方法会调用Callable当中实现的call方法,然后将结果保存下来,当调用get方法的时候再将这个结果返回。

自己实现FutureTask

工具准备

经过上文的分析你可能已经大致了解了FutureTask的大致执行过程了,但是需要注意的是,如果你执行FutureTaskget方法是可能阻塞的,因为可能Callablecall方法还没有执行完成。因此在get方法当中就需要有阻塞线程的代码,但是当call方法执行完成之后需要将这些线程都唤醒。

在本篇文章当中使用锁ReentrantLock和条件变量Condition进行线程的阻塞和唤醒,在我们自己动手实现FutureTask之前,我们先熟悉一下上面两种工具的使用方法。

ReentrantLock主要有两个方法:

  • lock对临界区代码块进行加锁。
  • unlock对临界区代码进行解锁。

Condition主要有三个方法:

  • await阻塞调用这个方法的线程,等待其他线程唤醒。
  • signal唤醒一个被await方法阻塞的线程。
  • signalAll唤醒所有被await方法阻塞的线程。
import java.util.concurrent.TimeUnit;
import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.ReentrantLock;

public class LockDemo {

  private ReentrantLock lock;
  private Condition condition;

  LockDemo() {
    lock = new ReentrantLock();
    condition = lock.newCondition();
  }

  public void blocking() {
    lock.lock();
    try {
      System.out.println(Thread.currentThread() + " 准备等待被其他线程唤醒");
      condition.await();
    } catch (InterruptedException e) {
      e.printStackTrace();
    }finally {
      lock.unlock();
    }
  }

  public void inform() throws InterruptedException {
    // 先休眠两秒 等他其他线程先阻塞
    TimeUnit.SECONDS.sleep(2);
    lock.lock();
    try {
      System.out.println(Thread.currentThread() + " 准备唤醒其他线程");
      condition.signal(); // 唤醒一个被 await 方法阻塞的线程
      // condition.signalAll(); // 唤醒所有被 await 方法阻塞的线程
    }finally {
      lock.unlock();
    }
  }

  public static void main(String[] args) {
    LockDemo lockDemo = new LockDemo();
    Thread thread = new Thread(() -> {
      lockDemo.blocking(); // 执行阻塞线程的代码
    }, "Blocking-Thread");
    Thread thread1 = new Thread(() -> {
      try {
        lockDemo.inform(); // 执行唤醒线程的代码
      } catch (InterruptedException e) {
        e.printStackTrace();
      }
    }, "Inform-Thread");
    thread.start();
    thread1.start();
  }
}

上面的代码的输出:

Thread[Blocking-Thread,5,main] 准备等待被其他线程唤醒
Thread[Inform-Thread,5,main] 准备唤醒其他线程

FutureTask设计与实现

在前文当中我们已经谈到了FutureTask的实现原理,主要有以下几点:

  • 构造函数需要传入一个实现了Callable接口的类对象,这个将会在FutureTaskrun方法执行,然后得到函数的返回值,并且将返回值存储起来。
  • 当线程调用get方法的时候,如果这个时候Callable当中的call已经执行完成,直接返回call函数返回的结果就行,如果call函数还没有执行完成,那么就需要将调用get方法的线程挂起,这里我们可以使用condition.await()将线程挂起。
  • call函数执行完成之后,需要将之前被get方法挂起的线程唤醒继续执行,这里使用condition.signalAll()将所有挂起的线程唤醒。
  • 因为是我们自己实现FutureTask,功能不会那么齐全,只需要能够满足我们的主要需求即可,主要是帮助大家了解FutureTask原理。

实现代码如下(分析都在注释当中):

import java.util.concurrent.Callable;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.ReentrantLock;

// 这里需要实现 Runnable 接口,因为需要将这个对象放入 Thread 类当中
// 而 Thread 要求传入的对象实现了 Runnable 接口
public class MyFutureTask<V> implements Runnable {

  private final Callable<V> callable;
  private Object returnVal; // 这个表示我们最终的返回值
  private final ReentrantLock lock;
  private final Condition condition;

  public MyFutureTask(Callable<V> callable) {
    // 将传入的 callable 对象存储起来 方便在后面的 run 方法当中调用
    this.callable = callable;
    lock = new ReentrantLock();
    condition = lock.newCondition();
  }

  @SuppressWarnings("unchecked")
  public V get(long timeout, TimeUnit unit) {
    if (returnVal != null) // 如果符合条件 说明 call 函数已经执行完成 返回值已经不为 null 了
      return (V) returnVal; // 直接将结果返回即可 这样不用竞争锁资源 提高程序执行效率
    lock.lock();
    try {
      // 这里需要进行二次判断 (双重检查)
      // 因为如果一个线程在第一次判断 returnVal 为空
      // 然后这个时候它可能因为获取锁而被挂起
      // 而在被挂起的这段时间,call 可能已经执行完成
      // 如果这个时候不进行判断直接执行 await方法
      // 那后面这个线程将无法被唤醒
      if (returnVal == null)
        condition.await(timeout, unit);
    } catch (InterruptedException e) {
      e.printStackTrace();
    } finally {
      lock.unlock();
    }
    return (V) returnVal;
  }

  @SuppressWarnings("unchecked")
  public V get() {
    if (returnVal != null)
      return (V) returnVal;
    lock.lock();
    try {
      // 同样的需要进行双重检查
      if (returnVal == null)
      	condition.await();
    } catch (InterruptedException e) {
      e.printStackTrace();
    } finally {
      lock.unlock();
    }
    return (V) returnVal;
  }

  @Override
  public void run() {
    if (returnVal != null)
      return;
    try {
      // 在 Runnable 的 run 方法当中
      // 执行 Callable 方法的 call 得到返回结果
      returnVal = callable.call();
    } catch (Exception e) {
      e.printStackTrace();
    }
    lock.lock();
    try {
      // 因为已经得到了结果
      // 因此需要将所有被 await 方法阻塞的线程唤醒
      // 让他们从 get 方法返回
      condition.signalAll();
    }finally {
      lock.unlock();
    }
  }
	// 下面是测试代码
  public static void main(String[] args) {
    MyFutureTask<Integer> ft = new MyFutureTask<>(() -> {
      TimeUnit.SECONDS.sleep(2);
      return 101;
    });
    Thread thread = new Thread(ft);
    thread.start();
    System.out.println(ft.get(100, TimeUnit.MILLISECONDS)); // 输出为 null
    System.out.println(ft.get()); // 输出为 101
  }
}

我们现在用我们自己写的MyFutureTask去实现在前文当中数组求和的例子:

public static void main(String[] args) throws ExecutionException, InterruptedException {
  int[] data = new int[100000];
  Random random = new Random();
  for (int i = 0; i < 100000; i++) {
    data[i] = random.nextInt(10000);
  }
  @SuppressWarnings("unchecked")
  MyFutureTask<Integer>[] tasks = (MyFutureTask<Integer>[]) Array.newInstance(MyFutureTask.class, 10);
  for (int i = 0; i < 10; i++) {
    int idx = i;
    tasks[i] = new MyFutureTask<>(() -> {
      int sum = 0;
      for (int k = idx * 10000; k < (idx + 1) * 10000; k++) {
        sum += data[k];
      }
      return sum;
    });
  }
  for (MyFutureTask<Integer> MyFutureTask : tasks) {
    new Thread(MyFutureTask).start();
  }
  int threadSum = 0;
  for (MyFutureTask<Integer> MyFutureTask : tasks) {
    threadSum += MyFutureTask.get();
  }
  int sum = Arrays.stream(data).sum();
  System.out.println(sum == threadSum); // 输出结果为 true
}

总结

在本篇文章当中主要给大家介绍了FutureTask的内部原理,并且我们自己通过使用ReentrantLockCondition实现了我们自己的FutureTask,本篇文章的主要内容如下:

FutureTask的内部原理:

  • FutureTask首先会继承Runnable接口,这样就可以将FutureTask的对象直接放入Thread类当中,作为构造函数的参数。
  • 我们在使用FutureTask的时候需要传入一个Callable实现类的对象,在函数call当中实现我们需要执行的函数,执行完成之后,将call函数的返回值保存下来,当有线程调用get方法时候将保存的返回值返回。

我们使用条件变量进行对线程的阻塞和唤醒。

  • 当有线程调用get方法时,如果call已经执行完成,那么可以直接将结果返回,否则需要使用条件变量将线程挂起。
  • call函数执行完成的时候,需要使用条件变量将所有阻塞在get方法的线程唤醒。

双重检查:

  • 我们在get方法当中首先判断returnVal是否为空,如果不为空直接将结果返回,这就可以不用去竞争锁资源了,可以提高程序执行的效率。
  • 但是我们在使用锁保护的临界区还需要进行判断,判断returnVal是否为空,因为如果一个线程在第一次判断 returnVal 为空,然后这个时候它可能因为获取锁而被挂起, 而在被挂起的这段时间,call 可能已经执行完成,如果这个时候不进行判断直接执行 await方法,那后面这个线程将无法被唤醒,因为在call函数执行完成之后调用了condition.signalAll(),如果线程在这之后执行await方法,那么将来再没有线程去将这些线程唤醒。

到此这篇关于Java实现FutureTask的示例详解的文章就介绍到这了,更多相关Java FutureTask内容请搜索我们以前的文章或继续浏览下面的相关文章希望大家以后多多支持我们!

(0)

相关推荐

  • 简谈java并发FutureTask的实现

    概述 在使用java多线程解决问题的时候,为了提高效率,我们常常会异步处理一些计算任务并在最后异步的获取计算结果,这个过程的实现离不开Future接口及其实现类FutureTask.FutureTask类实现了Runnable, Future接口,接下来我会通过源码对该类的实现进行详解. 使用 我们先看下FutureTask中的主要方法如下,可以看出FutureTask实现了任务及异步结果的集合功能.看到这块的方法,大家肯定会有疑问,Runnable任务的run方法返回空,FutureTask如

  • Java线程池FutureTask实现原理详解

    前言 线程池可以并发执行多个任务,有些时候,我们可能想要跟踪任务的执行结果,甚至在一定时间内,如果任务没有执行完成,我们可能还想要取消任务的执行,为了支持这一特性,ThreadPoolExecutor提供了 FutureTask 用于追踪任务的执行和取消.本篇介绍FutureTask的实现原理. 类视图 为了更好的理解FutureTask的实现原理,这里先提供几个重要接口和类的结构,如下图所示: RunnableAdapter ThreadPoolExecutor提供了submit接口用于提交任

  • Java多线程之FutureTask的介绍及使用

    一.FutureTask的理解 FutureTask属于java.util.concurrent 包:FutureTask表示可取消的异步计算.FutureTask类提供了一个Future的基本实现 ,具有启动和取消计算的方法,查询计算是否完整,并检索计算结果.结果只能在计算完成后才能检索; 如果计算尚未完成,则get方法将阻止. 一旦计算完成,则无法重新启动或取消计算(除非使用runAndReset()调用计算 ). 二.FutureTask类图 从上面的FutureTask类图中可以看出,F

  • Java FutureTask类使用案例解析

    FutureTask一个可取消的异步计算,FutureTask 实现了Future的基本方法,提空 start cancel 操作,可以查询计算是否已经完成,并且可以获取计算的结果.结果只可以在计算完成之后获取,get方法会阻塞当计算没有完成的时候,一旦计算已经完成,那么计算就不能再次启动或是取消. 一个FutureTask 可以用来包装一个 Callable 或是一个runnable对象.因为FurtureTask实现了Runnable方法,所以一个 FutureTask可以提交(submit

  • Java多线程并发FutureTask使用详解

    目录 基本使用 代码分析 继承关系 Future RunnableFuture FutureTask 状态 属性 内部类 构造方法 检索 FutureTask 状态 取消操作 计算结果 立刻获取结果或异常 run 方法组 本文基于最新的 OpenJDK 代码,预计发行版本为 19 . Java 的多线程机制本质上能够完成两件事情,异步计算和并发.并发问题通过解决线程安全的一系列 API 来解决:而异步计算,常见的使用是 Runnable 和 Callable 配合线程使用. FutureTask

  • Java中Future和FutureTask的示例详解及使用

    目录 一.Future 接口 二.FutureTask 三.使用 Callable 和 Future 四.小结(FutureTask核心原理) 总结 一.Future 接口 当 call()方法完成时,结果必须存储在主线程已知的对象中,以便主线程可以知道该线程返回的结果.为此,可以使用 Future 对象. 将 Future 视为保存结果的对象–它可能暂时不保存结果,但将来会保存(一旦Callable 返回).Future 基本上是主线程可以跟踪进度以及其他线程的结果的一种方式.要实现此接口,必

  • Java反射框架Reflections示例详解

    MAVEN 坐标 <dependency> <groupId>org.reflections</groupId> <artifactId>reflections</artifactId> <version>0.9.10</version> </dependency> Reflections 的作用 Reflections通过扫描classpath,索引元数据,并且允许在运行时查询这些元数据. 获取某个类型的所有

  • kotlin和Java的相互调用示例详解

    前言 互操作就是在Kotlin中可以调用其他编程语言的接口,只要它们开放了接口,Kotlin就可以调用其成员属性和成员方法,这是其他编程语言所无法比拟的.同时,在进行Java编程时也可以调用Kotlin中的API接口. 1.在kotlin中调用Java方法 Kotlin和Java是两种不同的语言,所以在互相调用的时候,会有一些特殊的语法.kotlin中对象属性默认就带有setter和getter方法,所以在kotlin中调用Java时直接变量名点属性就可获取到属性的setter和getter的一

  • Java 8 lambda初试示例详解

    表达式本质上是一个匿名方法.让我们来看下面这个例子: public int add(int x, int y) { return x + y; } 转成λ表达式后是这个样子: (int x, int y) -> x + y; 参数类型也可以省略,Java编译器会根据上下文推断出来: (x, y) -> x + y; //返回两数之和 或者 (x, y) -> { return x + y; } //显式指明返回值 可见λ表达式有三部分组成:参数列表,箭头(->),以及一个表达式或语

  • Java的枚举enum示例详解

    一. 什么是枚举 枚举是一种数据类型,具有集合的一些特点,可以存放多个元素,但存储对象有限且固定,枚举也有比较常见的使用场景,如我们需要表达性别(男.女),颜色(红.黄.蓝),星期(星期一.星期二...星期日),四季(春.夏.秋.冬),地理位置(东.西.南.北),方向(前.后.左.右)等,这些场景都非常适合枚举. 二. 定义枚举 java中使用enum来定义枚举,和class,interface具有同样的等级,(注意是enum,而不是Enum),定义枚举可以有两种情况 第一种:默认构造器(空构造

  • java 与testng利用XML做数据源的数据驱动示例详解

    java 与testng利用XML做数据源的数据驱动示例详解 testng的功能很强大,利用@DataProvider可以做数据驱动,数据源文件可以是EXCEL,XML,YAML,甚至可以是TXT文本.在这以XML为例: 备注:@DataProvider的返回值类型只能是Object[][]与Iterator<Object>[] TestData.xml: <?xml version="1.0" encoding="UTF-8"?> <

  • Java编程swing组件JLabel详解以及使用示例

    JLabel 对象可以显示文本.图像或同时显示二者.可以通过设置垂直和水平对齐方式,指定标签显示区中标签内容在何处对齐.默认情况下,标签在其显示区内垂直居中对齐.默认情况下,只显示文本的标签是开始边对齐:而只显示图像的标签则水平居中对齐. 还可以指定文本相对于图像的位置.默认情况下,文本位于图像的结尾边上,文本和图像都垂直对齐. 构造方法介绍: JLabel() 创建无图像并且其标题为空字符串的 JLabel. JLabel(Icon image) 创建具有指定图像的 JLabel 实例. JL

  • JSON在Java中的相互转换示例详解

    什么是JSON? JSON (JavaScript Object Notation) 是一种轻量级的数据交换格式. 易于人阅读和编写.同时也易于机器解析和生成. JSON采用完全独立于语言的文本格式,而且很多语言都提供了对json的支持(包括C, C++, C#, Java, JavaScript, Perl, Python等). 这样就使得JSON成为理想的数据交换格式. 数据交换格式,是指客户端和服务器之间通信,传递数据时,数据使用的格式是json JSON在java中的使用 json在ja

  • java中Servlet监听器的工作原理及示例详解

    监听器就是一个实现特定接口的普通java程序,这个程序专门用于监听另一个java对象的方法调用或属性改变,当被监听对象发生上述事件后,监听器某个方法将立即被执行. 监听器原理 监听原理 1.存在事件源 2.提供监听器 3.为事件源注册监听器 4.操作事件源,产生事件对象,将事件对象传递给监听器,并且执行监听器相应监听方法 监听器典型案例:监听window窗口的事件监听器 例如:swing开发首先制造Frame**窗体**,窗体本身也是一个显示空间,对窗体提供监听器,监听窗体方法调用或者属性改变:

  • Java使用Collections.sort()排序的示例详解

    Java中Collections.sort()排序详解,通过实例代码给大家讲解,具体代码如下所示: public static void main(String[] args) { List<String> list = new ArrayList<String>(); list.add("beijing"); list.add("shanghai"); list.add("hangzhou"); Collections.

随机推荐

其他