手写理解Callable,Future,Executor

2023-02-06 本文已影响0人 pq217

前言

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) {
        // 从线程池中选一个线程去执行
    }
}

最后

以上代码的命名基本参照jdk的源码，可以自行对照，相信再看源码就会非常清晰，也可以结合Executor源码详解解读源码

手写理解Callable,Future,Executor

前言

需求

实现

封装

多线程

扩展

最后

猜你喜欢

热点阅读