Java实现FutureTask的示例详解

@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;
}

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
  }
}

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();
  }
}

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
  }
}

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
}