Java 中的FutureTask 2

1. 前言

前两篇简单分析了 Future接口和 FutureTask 本文将介绍 FutureTask 的使用方式。

2. FutureTask 的使用

FutureTask 有两个构造函数，分别传入 Runnale 和 Callable 实例。所以 FutureTask 的使用和这两个接口有关系。

2.1 结合 Callable

Callable接口只定义了一个 call()方法，与Runnable的run()方法相比，该方法有返回值，泛型V就是要返回的结果类型，可以返回子任务的执行结果。如果你期望获取线程执行的结果可以使用 Callable ,而且 Callable 可以抛出异常。我们定义一个简单的任务：

package cn.felord;
import java.util.concurrent.Callable;

/**
 * @author felord.cn
 * @since 2020/3/21 20:35
 */
public class MyTask implements Callable<Integer> {
    @Override
    public Integer call() {
        int total = 0;
        try {
            for (int i = 0; i < 10; i++) {
                System.out.println(" thread: " + Thread.currentThread().getName() + " i = " + i);
                Thread.sleep(1000);
                total += i;
            }
        } catch (InterruptedException e) {
            System.out.println("task is interrupted");
            // 遇到异常需要中断后返回以保证结束线程
            return 0;
        }
        return total;
    }
}

然后下面是在一个 main 方法中的模版步骤：

    public static void main(String[] args) {
        // 第一步：声明具体的计算任务
        MyTask myTask = new MyTask();
        // 第二步：将任务传入初始化 FutureTask
        FutureTask<Integer> futureTask = new FutureTask<>(myTask);
        // 第三步： 将 FutureTask 交给一个线程去执行
        Thread thread = new Thread(futureTask);

        thread.setName("future task thread");
        thread.start();

        try {
            Integer integer = futureTask.get();
            System.out.println("integer = " + integer);
        } catch (InterruptedException | ExecutionException e) {
            e.printStackTrace();
        }
        System.out.println(" task is over ");

    }

运行后似乎返回了预期值 45 ，看似没有问题。但是当我们使用超时机制也就是将上述代码中获取结果的代码改为后Integer integer = futureTask.get(5000, TimeUnit.MILLISECONDS); 发现结果是这样的：

 thread: future task thread i = 0
 thread: future task thread i = 1
 thread: future task thread i = 2
 thread: future task thread i = 3
 thread: future task thread i = 4
java.util.concurrent.TimeoutException
    at java.util.concurrent.FutureTask.get(FutureTask.java:205)
    at cn.felord.Test.main(Test.java:26)
 thread: future task thread i = 5
 task is over 
 thread: future task thread i = 6
 thread: future task thread i = 7
 thread: future task thread i = 8
 thread: future task thread i = 9

我们强迫任务超时，结果任务的计算线程依然在进行计算，所以需要我们对超时的异常进行一些处理要么中断计算要么继续 get。

获取 FutureTask 的结果超时不意味着任务的结束。而且通常不建议用以上的方式进行任务计算。

2.2 结合 Runnable 并定义出结果

构造是这样：

  public FutureTask(Runnable runnable, V result) {
        this.callable = Executors.callable(runnable, result);
        this.state = NEW;       // ensure visibility of callable
  }

开始我以为这个 result 是通过 runnable 计算出来的，然而我错了：

   static final class RunnableAdapter<T> implements Callable<T> {
        final Runnable task;
        final T result;
        RunnableAdapter(Runnable task, T result) {
            this.task = task;
            this.result = result;
        }
        public T call() {
            task.run();
            return result;
        }
    }

这个是 Executors.callable 方法的底层，我们并不能在线程计算中去操作 result 。不太明白 JDK 为什么提供这种”然并卵”的的方式。运行给定的任务并返回给定的结果（就是你传进去的参数result，也就是任务的执行和给定的结果没有关系。

2.3 配合线程池使用

上面两种一般我们只是在研究中使用，一般操作线程都建议使用特定的线程池来进行。所以以下方式才是正统的使用 FutureTask 的方法：

package cn.felord;

import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicInteger;

/**
 * @author dax
 * @since 2020/3/21 20:49
 */
public class Test {
    public static void main(String[] args) {
        // 第一步：声明具体的计算任务
        MyTask myTask = new MyTask();

        // 第二步： 将 FutureTask 交给线程池去执行
        // ExecutorService pool = Executors.newCachedThreadPool();
        // 通常推荐使用自定义线程池 或者 Spring 提供的线程池以防止OOM 同时应该考虑线程池的关闭策略
        ExecutorService pool = newThreadPool();

        try {
        // 获取结果    
            Integer integer = pool.submit(myTask).get();
            System.out.println("integer = " + integer);
        } catch (InterruptedException | ExecutionException e) {
            e.printStackTrace();
        }
    }


    private static ExecutorService newThreadPool() {
        ThreadFactory namedThreadFactory = new ThreadFactory() {
            final AtomicInteger sequence = new AtomicInteger(1);

            @Override
            public Thread newThread(Runnable r) {
                Thread thread = new Thread(r);
                int seq = this.sequence.getAndIncrement();
                thread.setName("future-task-thread" + (seq > 1 ? "-" + seq : ""));
                if (!thread.isDaemon()) {
                    thread.setDaemon(true);
                }

                return thread;
            }
        };
        return new ThreadPoolExecutor(5, 200,
                0L, TimeUnit.MILLISECONDS,
                new LinkedBlockingQueue<>(1024),
                namedThreadFactory,
                new ThreadPoolExecutor.AbortPolicy());
    }
}

这种方式是不是更加安全和简单？

3. 容易出现的问题

1. 如果具体的 Callable 异常进行了 try-catch 一定要返回一个结果，否则线程会继续执行。

2. 调用get()方法会一直阻塞到任务执行完毕才返回。

3. get(long timeout, TimeUnit unit)用来在一定时间内获取执行结果。如果超时则后面的代码会继续执行，同时计算并不会因此而中断。

4. cancel(boolean mayInterruptIfRunning) 并不一定能立即中断任务，需要调用 isDone 或者 isCancelled 判断，更加保险的是 调用任务线程的 isInterrupted() 进行中断状态判断。

4. 总结

今天对 FutureTask 的使用方法进行了罗列，同时对使用中容易出现的一些误区进行了说明，希望对你有所帮助。如果你有什么问题可以留言讨论。