手写理解Callable,Future,Executor

您是不是想找：线程是什么意思线程撕裂者cpu线程和进程的关系和区别线程数是不是越多越好线程和进程的主要区别线程撕裂者线程撕裂者cpu天梯图线程的5种状态线程池七大核心参数线程安全

前言

Callable,Future,Executor都是java.util.concurrent包下的工具类，作者李二狗，为了彻底吃透它们的概念，今天就假设这些类都不存在，自己通过实际场景封装出这些工具的山寨版

需求

假设你需要写一个简单的方法，两个值求和，非常简单

public int sum(int x, int y) {
    return x + y;
}

但需求增加了，需要计算的过程在一个新线程中执行，这代码该怎么写？就会出现以下两个问题：

怎么获取到线程执行的结果？
怎么知道新线程什么时候执行完？

实现

首先第一个问题，如何获取新线程结果，这个也好解决，虽然新线程里的变量我取不到，但内存是线程共享的啊，只要提前定义一个结果变量即可

private volatile int outcome;

public int sum(int x, int y) {
    new Thread(()->{
        outcome = x + y;
    }).start();
    return outcome;
}

很明显，最终结果肯定不对，因为返回的时候新线程可能都没开始运行，这就是第二个问题：怎么知道新线程什么时候执行完？

解决的方案当然也很多，只要线程之间进行通讯一下即可，我们用LockSupport方法来实现通讯

public class OperationTest {

    private volatile int outcome;

    private Thread waitThread;

    public int sum(int x, int y) throws InterruptedException {
        waitThread = Thread.currentThread();
        new Thread(() -> {
            outcome = x + y;
            // 计算完成通知等待线程
            LockSupport.unpark(waitThread);
        }).start();
        // 等待计算完成
        LockSupport.park(this);
        return outcome;
    }

    public static void main(String[] args) throws InterruptedException {
        System.out.println(new OperationTest().sum(2,3));
    }
}

此时我们就完成了这个需求，但看一下代码真的好麻烦啊，明明就是一个1+1等于几的事，写了这么多代码，如果明天再来个写减法的需求，我还要重写这么一堆

封装

其实上面解决的两个问题，完全就是通用性的问题，加法这样处理，减法一样也是这个解决思路，那么我们就可以把加法、减法等有返回值的方法用函数式接口给抽象化

// 使用泛型兼容各种类型返回
@FunctionalInterface
public interface Callable<V> {
    V call();
}

接下来我们就要封装一个开启新线程执行它的工具，提供如下功能：

只要传入一个方法作为参数，就可以开启一个新线程去执行这个方法，并返回执行结果
也可以开启新线程执行普通的Runable方法

这个工具命名为ExecutorService

public interface ExecutorService {
    /**
     * 执行callable并返回执行结果
     * @param task
     * @param <T>
     */
    <T> T submit(Callable<T> task);

    /**
     * 也可以执行Runnable
     * @param runnable
     */
    void execute(Runnable runnable);
}

然后开始实现这个工具，暂时叫做NewThreadExecutor(新线程执行器)

public class NewThreadExecutor implements ExecutorService {

    private volatile Object outcome;

    private Thread waitThread;

    @Override
    public <T> T submit(Callable<T> task) {
        waitThread = Thread.currentThread();
        execute(()->{
            outcome = task.call();
            // 计算完成通知等待线程
            LockSupport.unpark(waitThread);
        });
        // 等待计算完成
        LockSupport.park(this);
        return (T) outcome;
    }

    @Override
    public void execute(Runnable runnable) {
        // 执行方式就是开启一个线程去执行
        new Thread(runnable).start();
    }

}

这时我们再实现上面的需求就轻而易举了

int x = 2;
int y = 3;
Integer sub = new NewThreadExecutor().submit(() -> {
    return x + y;
});
System.out.println(sub);

这样通过我们的逻辑和执行解耦，可以方便使用工具执行减法、乘法或其它复杂运算逻辑

多线程

再回头看一下我们这个工具，实际上非常不合理，开了一个新线程去执行函数，整个过程主线程却全程傻等

相当于一个主管带一个员工干活，而员工干活时，主管干不了别的事只能等着，那干脆主管自己干得了呗，何必聘请这么一个员工

而我们希望开启新线程后主线程可以去干别的(比如分配新任务给其它线程执行)，等全分配完任务再统一获取结果，这样才算是多线程并行作业

那么如何改造代码呐？

首先，调用submit方法不能阻塞，应该直接返回一个对象，主线程再想要获取的时候，才通过这个对象阻塞获取结果

这个对象不是运行结果，但通过它可以获得结果，他就像一个未来的约定，我们先使用代码给它抽象出来，命名为Future

public interface Future<V> {
    // 是否运行完成
    boolean isDone();
    // 获取运行结果
    V get();
}

此时我们的ExecutorService返回结果变为Future对象

public interface ExecutorService {
    /**
     * 执行callable并返回future
     * @param task
     * @param <T>
     */
    <T> Future<T> submit(Callable<T> task);

    /**
     * 也可以执行Runnable
     * @param runnable
     */
    void execute(Runnable runnable);
}

那么此时如何改造NewThreadExecutor这个实现呐？

首先要实现Future抽象，这个对象可以获取到执行结果，那么它肯定可以访问到存储执行结果的对象(outcome)和等待线程对象(waitThread)，那不妨就把这两个对象放入Future实现中，同时最终执行的Runable方法也要可以访问到这两个对象，那不妨就让Future的实现同时就是最终执行的Runable，即可执行的Future，取名为FutureTask

public class FutureTask<V> implements Future<V>, Runnable {

    private Callable<V> callable; // 要执行的方法

    private volatile Object outcome; // 执行结果

    private Thread waitThread; // 等待的线程

    public FutureTask(Callable<V> callable) {
        this.callable = callable;
    }

    @Override
    public boolean isDone() {
        return outcome!=null;
    }

    @Override
    public V get() {
        waitThread = Thread.currentThread();
        if (isDone()) { // 如果已经执行完直接返回
            return (V) outcome;
        }
        // 否则等待
        LockSupport.park(this);
        return (V) outcome;
    }

    @Override
    public void run() {
        // 开始执行
        outcome = callable.call();
        // 计算完成通知等待线程
        LockSupport.unpark(waitThread);
    }
}

此时NewThreadExecutor改造如下

public class NewThreadExecutor implements ExecutorService {

    @Override
    public <T> Future<T> submit(Callable<T> task) {
        FutureTask<T> futureTask = new FutureTask<>(task);
        execute(futureTask);
        return futureTask; // 直接返回
    }

    @Override
    public void execute(Runnable runnable) {
        // 执行方式就是开启一个线程去执行
        new Thread(runnable).start();
    }

}

这时我们就可以让两个子线程分别同时计算两个结果，最终主线程求和(真正的做到多线程计算)

Future<Integer> future1 = new NewThreadExecutor().submit(() -> {
    return 3 + 4;
});
Future<Integer> future2 = new NewThreadExecutor().submit(() -> {
    return 1 + 2;
});
System.out.println(future1.get()+future2.get());

扩展

以上封装的工具，达到了传入一个方法开启一个新线程计算的功能，并且使用future概念避免了阻塞

但工具还能再扩展一下，比如有一天领导让实现传入一个方法指定某一线程执行，或传入方法从几个固定线程中选一个空闲的去执行(线程池)

由于我们做到了逻辑和执行的分离解耦，所以只要重写一下execute就可以了，而无论如何执行submit的逻辑是不变的，我们可以继续给它抽象出来，命名为AbstractExecutorService

public abstract class AbstractExecutorService implements ExecutorService {

    @Override
    public <T> Future<T> submit(Callable<T> task) {
        FutureTask<T> futureTask = new FutureTask<>(task);
        execute(futureTask);
        return futureTask; // 直接返回
    }

}

此时它的继承者就可以传入方法并返回future，只需关注如何执行即可，比如我们的开启新线程执行工具

public class NewThreadExecutor extends AbstractExecutorService {

    @Override
    public void execute(Runnable runnable) {
        // 执行方式就是开启一个线程去执行
        new Thread(runnable).start();
    }

}

再比如使用线程池去执行

public class ThreadPoolExecutor extends AbstractExecutorService {
    @Override
    public void execute(Runnable runnable) {
        // 从线程池中选一个线程去执行
    }
}