深入理解高并发编程 - 深度解析线程池中重要的顶层接口和抽象类

心已赠人 2023-10-14 16:17 51阅读 0赞

## 1、接口和抽象类总览 ##

线程池提供了一组接口和抽象类，用于实现线程池的各种策略和功能。

Executor 接口： 是线程池的核心接口，定义了一个执行任务的方法 void execute(Runnable command)。
    
    ExecutorService 接口： 扩展了 Executor 接口，提供了更丰富的任务提交和管理功能，如任务提交、关闭线程池、获取 Future 对象等。
    
    AbstractExecutorService 抽象类： 实现了 ExecutorService 接口的骨架，为一些基本方法提供了默认实现，简化了自定义线程池的开发。
    
    ScheduledExecutorService 接口： 继承了 ExecutorService 接口，提供了支持任务定时执行和周期性执行的功能。

### 1.1、Executor 接口 ###

Executor 接口是 Java 标准库中定义的一个简单的线程执行器接口，它只包含一个方法 execute(Runnable command)，用于执行传入的 Runnable 任务。Executor 接口在 Java 5 中引入，是线程池框架的基础之一。下面是 Executor 接口的源码及其简要解析：

package java.util.concurrent;
    
    public interface Executor {
        void execute(Runnable command);
    }

Executor 接口定义了一个单一的方法 execute，该方法接受一个 Runnable 对象作为参数，用于执行一个任务。  
使用 Executor 接口时，需要提供一个实现了 execute 方法的类，以便在这个方法中执行传入的任务。以下是一个简单的示例，展示了如何使用 Executor 接口创建一个线程执行器并提交任务：

import java.util.concurrent.Executor;
    
    public class SimpleExecutorExample {
        public static void main(String[] args) {
            Executor executor = new SimpleExecutor();
            
            executor.execute(() -> {
                System.out.println("Task executed by executor.");
            });
        }
    
        static class SimpleExecutor implements Executor {
            @Override
            public void execute(Runnable command) {
                new Thread(command).start();
            }
        }
    }

在上述示例中，实现了一个简单的 Executor 接口，它使用新线程来执行传入的任务。当调用 execute 方法时，会创建一个新线程并运行传入的任务。

需要注意的是，尽管 Executor 接口只有一个方法，但它在 Java 并发编程中扮演着重要的角色，它是更高级的线程池框架的基础，并在整个 Java 并发库中有着广泛的应用。Executor 接口的存在使得线程的创建和管理更加灵活和可控，有助于避免不必要的线程创建和销毁开销，提高了多线程编程的效率和可维护性。

### 1.2、ExecutorService 接口 ###

ExecutorService 接口是 Java 并发库中的一个更高级别的线程池框架，它扩展了 Executor 接口，并提供了更丰富的线程池管理和任务执行的功能。ExecutorService 接口在 Java 5 中引入，为线程池的使用提供了更多的灵活性和控制能力。以下是 ExecutorService 接口的源码及其简要解析：

package java.util.concurrent;
    
    public interface ExecutorService extends Executor {
        void shutdown();
        List<Runnable> shutdownNow();
        boolean isShutdown();
        boolean isTerminated();
        boolean awaitTermination(long timeout, TimeUnit unit) throws InterruptedException;
        <T> Future<T> submit(Callable<T> task);
        <T> Future<T> submit(Runnable task, T result);
        Future<?> submit(Runnable task);
        <T> List<Future<T>> invokeAll(Collection<? extends Callable<T>> tasks) throws InterruptedException;
        <T> List<Future<T>> invokeAll(Collection<? extends Callable<T>> tasks, long timeout, TimeUnit unit) throws InterruptedException;
        <T> T invokeAny(Collection<? extends Callable<T>> tasks) throws InterruptedException, ExecutionException;
        <T> T invokeAny(Collection<? extends Callable<T>> tasks, long timeout, TimeUnit unit) throws InterruptedException, ExecutionException, TimeoutException;
    }

ExecutorService 接口扩展了 Executor 接口，继承了 execute 方法，同时添加了一系列新的方法，用于更细粒度地管理线程池、提交任务和获取任务执行结果。  
以下是一些 ExecutorService 接口中常用方法的简要说明：

shutdown()：优雅地关闭线程池，等待已提交的任务完成。
    shutdownNow()：尝试立即停止所有执行中的任务，并返回未启动的任务列表。
    isShutdown()：判断线程池是否已经关闭。
    isTerminated()：判断线程池中的所有任务是否已经完成。
    awaitTermination(long timeout, TimeUnit unit)：等待线程池中的所有任务在指定时间内完成。
    submit(Callable<T> task)：提交一个可调用的任务并返回一个 Future 对象以获取任务执行结果。
    submit(Runnable task, T result)：提交一个 Runnable 任务，并返回一个 Future 对象以获取结果。
    invokeAll(Collection<? extends Callable<T>> tasks)：提交多个可调用的任务，并等待它们全部完成，返回包含 Future 对象的列表。
    invokeAny(Collection<? extends Callable<T>> tasks)：提交多个可调用的任务，等待任意一个任务完成并返回其结果。

使用 ExecutorService 接口可以更方便地管理线程池和任务执行，以及获取任务的执行结果。它提供了更多的方法来控制线程池的状态和任务的执行，适用于各种多线程编程场景。

### 1.3、AbstractExecutorService 抽象类 ###

AbstractExecutorService 是一个抽象类，实现了 ExecutorService 接口的部分方法，为自定义线程池提供了一些通用的实现。它充当了一个骨架（框架）类，开发者可以继承这个抽象类并重写其中的方法来创建自己的线程池实现。下面是 AbstractExecutorService 抽象类的源码及其简要解析：

package java.util.concurrent;
    
    public abstract class AbstractExecutorService implements ExecutorService {
        // 实现了 ExecutorService 接口的 shutdown 方法
        public void shutdown() {
            throw new UnsupportedOperationException();
        }
    
        // 实现了 ExecutorService 接口的 isShutdown 方法
        public List<Runnable> shutdownNow() {
            shutdown();
            return Collections.emptyList();
        }
    
        // 实现了 ExecutorService 接口的 isShutdown 方法
        public boolean isShutdown() {
            return false;
        }
    
        // 实现了 ExecutorService 接口的 isTerminated 方法
        public boolean isTerminated() {
            return false;
        }
    
        // 实现了 ExecutorService 接口的 awaitTermination 方法
        public boolean awaitTermination(long timeout, TimeUnit unit)
            throws InterruptedException {
            return true;
        }
    
        // 实现了 ExecutorService 接口的 submit 方法
        public <T> Future<T> submit(Callable<T> task) {
            if (task == null) throw new NullPointerException();
            RunnableFuture<T> ftask = newTaskFor(task);
            execute(ftask);
            return ftask;
        }
    
        // 实现了 ExecutorService 接口的 submit 方法
        public <T> Future<T> submit(Runnable task, T result) {
            if (task == null) throw new NullPointerException();
            RunnableFuture<T> ftask = newTaskFor(task, result);
            execute(ftask);
            return ftask;
        }
    
        // 实现了 ExecutorService 接口的 submit 方法
        public Future<?> submit(Runnable task) {
            if (task == null) throw new NullPointerException();
            RunnableFuture<Void> ftask = newTaskFor(task, null);
            execute(ftask);
            return ftask;
        }
    
        // 实现了 ExecutorService 接口的 invokeAll 方法
        public <T> List<Future<T>> invokeAll(Collection<? extends Callable<T>> tasks)
            throws InterruptedException {
            if (tasks == null) throw new NullPointerException();
            List<Future<T>> futures = new ArrayList<Future<T>>(tasks.size());
            boolean done = false;
            try {
                for (Callable<T> t : tasks) {
                    RunnableFuture<T> f = newTaskFor(t);
                    futures.add(f);
                    execute(f);
                }
                for (int i = 0, size = futures.size(); i < size; i++) {
                    Future<T> f = futures.get(i);
                    if (!f.isDone()) {
                        try {
                            f.get();
                        } catch (CancellationException ignore) {
                        } catch (ExecutionException ignore) {
                        }
                    }
                }
                done = true;
                return futures;
            } finally {
                if (!done)
                    for (int i = 0, size = futures.size(); i < size; i++)
                        futures.get(i).cancel(true);
            }
        }
    
        // 实现了 ExecutorService 接口的 invokeAll 方法
        public <T> List<Future<T>> invokeAll(Collection<? extends Callable<T>> tasks,
                                             long timeout, TimeUnit unit)
            throws InterruptedException {
            if (tasks == null || unit == null) throw new NullPointerException();
            long nanos = unit.toNanos(timeout);
            List<Future<T>> futures = new ArrayList<Future<T>>(tasks.size());
            boolean done = false;
            try {
                for (Callable<T> t : tasks)
                    futures.add(newTaskFor(t));
                long lastTime = System.nanoTime();
                Iterator<Future<T>> it = futures.iterator();
                while (it.hasNext()) {
                    long now = System.nanoTime();
                    if (nanos <= 0) {
                        done = false;
                        break;
                    }
                    Future<T> f = it.next();
                    if (!f.isDone()) {
                        if (nanos >= spinForTimeoutThreshold)
                            ForkJoinPool.managedBlock(new ForkJoinPool.ManagedBlocker() {
                                public boolean block() {
                                    boolean recheck = true;
                                    while (nanos > 0) {
                                        long ms = TimeUnit.NANOSECONDS.toMillis(nanos);
                                        LockSupport.parkNanos(ms);
                                        if ((recheck = !f.isDone()) || (nanos = now - System.nanoTime()) > 0)
                                            break;
                                    }
                                    return recheck;
                                }
                                public boolean isReleasable() {
                                    return done;
                                }
                            });
                        if (!f.isDone()) {
                            if (nanos <= 0) {
                                done = false;
                                break;
                            }
                            nanos -= System.nanoTime() - lastTime;
                            lastTime = System.nanoTime();
                        }
                    }
                    it.remove();
                }
                for (int i = 0, size = futures.size(); i < size; i++) {
                    Future<T> f = futures.get(i);
                    if (!f.isDone())
                        f.cancel(true);
                }
                done = true;
                return futures;
            } finally {
                if (!done)
                    for (int i = 0, size = futures.size(); i < size; i++)
                        futures.get(i).cancel(true);
            }
        }
    
        // 实现了 ExecutorService 接口的 invokeAny 方法
        public <T> T invokeAny(Collection<? extends Callable<T>> tasks)
            throws InterruptedException, ExecutionException {
            try {
                return doInvokeAny(tasks, false, 0);
            } catch (TimeoutException cannotHappen) {
                assert false;
                return null;
            }
        }
    
        // 实现了 ExecutorService 接口的 invokeAny 方法
        public <T> T invokeAny(Collection<? extends Callable<T>> tasks,
                               long timeout, TimeUnit unit)
            throws InterruptedException, ExecutionException, TimeoutException {
            return doInvokeAny(tasks, true, unit.toNanos(timeout));
        }
    
        // 实现了 ExecutorService 接口的 submit 方法
        protected <T> RunnableFuture<T> newTaskFor(Runnable runnable, T value) {
            return new FutureTask<T>(runnable, value);
        }
    
        // 实现了 ExecutorService 接口的 submit 方法
        protected <T> RunnableFuture<T> newTaskFor(Callable<T> callable) {
            return new FutureTask<T>(callable);
        }
    
        // 实现了 ExecutorService 接口的 invokeAny 方法
        private <T> T doInvokeAny(Collection<? extends Callable<T>> tasks,
                                  boolean timed, long nanos)
            throws InterruptedException, ExecutionException, TimeoutException {
            if (tasks == null) throw new NullPointerException();
            int ntasks = tasks.size();
            if (ntasks == 0)
                throw new IllegalArgumentException();
            List<Future<T>> futures = new ArrayList<Future<T>>(ntasks);
            ExecutorCompletionService<T> ecs =
                new ExecutorCompletionService<T>(this);
    
            try {
                ExecutionException ee = null;
                long lastTime = (timed) ? System.nanoTime() : 0;
                Iterator<? extends Callable<T>> it = tasks.iterator();
    
                // 开始提交任务
                futures.add(ecs.submit(it.next()));
                --ntasks;
                int active = 1;
    
                for (;;) {
                    Future<T> f = ecs.poll();
                    if (f == null) {
                        if (ntasks > 0) {
                            --ntasks;
                            futures.add(ecs.submit(it.next()));
                            ++active;
                        } else if (active == 0)
                            break;
                        else if (timed) {
                            f = ecs.poll(nanos, TimeUnit.NANOSECONDS);
                            if (f == null)
                                throw new TimeoutException();
                            long now = System.nanoTime();
                            nanos -= now - lastTime;
                            lastTime = now;
                        } else
                            f = ecs.take();
                    }
                    if (f != null) {
                        --active;
                        try {
                            return f.get();
                        } catch (ExecutionException eex) {
                            ee = eex;
                        } catch (RuntimeException rex) {
                            ee = new ExecutionException(rex);
                        }
                    }
                }
    
                if (ee == null)
                    ee = new ExecutionException(null);
                throw ee;
    
            } finally {
                for (int i = 0, size = futures.size(); i < size; i++)
                    futures.get(i).cancel(true);
            }
        }
    
        // 阻塞时长阈值，用于实现 awaitTermination 的时间计算
        static final long DEFAULT_TIMED_SPIN_THRESHOLD = 1000L;
        private static final long spinForTimeoutThreshold = 
            (Runtime.getRuntime().availableProcessors() > 1) ?
            1000L :
            DEFAULT_TIMED_SPIN_THRESHOLD;
    }

AbstractExecutorService 抽象类实现了 ExecutorService 接口的一些方法，提供了通用的实现。其中，execute 方法没有在抽象类中实现，需要在子类中具体实现。其他方法中的具体实现逻辑有助于理解如何使用和扩展这些方法来创建自定义的线程池实现。

需要注意，AbstractExecutorService 抽象类提供的是一些默认行为，可以继承它并重写其中的方法来实现自定义的线程池行为。例如，可以在子类中实现 execute 方法，将任务提交到实际的线程池中，或者在 shutdown 方法中执行一些资源释放的操作等。

总之，AbstractExecutorService 抽象类为自定义线程池提供了一个基础，帮助开发者更轻松地实现自己的线程池，并在其中定制线程池的行为和特性。

### 1.4、ScheduledExecutorService 接口 ###

ScheduledExecutorService 接口是 Java 并发库中的一个接口，扩展了 ExecutorService 接口，用于支持任务的定时执行和周期性执行。它提供了一些方法来安排任务在指定的延迟或时间间隔后执行，适用于需要定时执行的场景。以下是 ScheduledExecutorService 接口的源码及其简要解析：

package java.util.concurrent;
    
    public interface ScheduledExecutorService extends ExecutorService {
        // 延迟指定时间后执行任务，返回一个 ScheduledFuture 对象
        ScheduledFuture<?> schedule(Runnable command, long delay, TimeUnit unit);
    
        // 延迟指定时间后执行任务，并返回一个 ScheduledFuture 对象
        <V> ScheduledFuture<V> schedule(Callable<V> callable, long delay, TimeUnit unit);
    
        // 延迟指定时间后周期性地执行任务，返回一个 ScheduledFuture 对象
        ScheduledFuture<?> scheduleAtFixedRate(Runnable command, long initialDelay, long period, TimeUnit unit);
    
        // 延迟指定时间后在上一次任务完成后再执行任务，返回一个 ScheduledFuture 对象
        ScheduledFuture<?> scheduleWithFixedDelay(Runnable command, long initialDelay, long delay, TimeUnit unit);
    }

ScheduledExecutorService 接口定义了一些方法来支持任务的定时执行和周期性执行：

schedule(Runnable command, long delay, TimeUnit unit)：延迟指定时间后执行任务，返回一个 ScheduledFuture 对象，可以用于获取任务的执行结果或取消任务。
    
    schedule(Callable<V> callable, long delay, TimeUnit unit)：延迟指定时间后执行任务，并返回一个 ScheduledFuture 对象，可以用于获取任务的执行结果或取消任务。
    
    scheduleAtFixedRate(Runnable command, long initialDelay, long period, TimeUnit unit)：延迟指定时间后周期性地执行任务，任务之间的执行时间间隔是固定的。
    
    scheduleWithFixedDelay(Runnable command, long initialDelay, long delay, TimeUnit unit)：延迟指定时间后在上一次任务完成后再执行任务，任务之间的执行时间间隔是固定的。

使用 ScheduledExecutorService 接口可以轻松地实现任务的定时执行和周期性执行。它提供了一些灵活的方法来满足不同场景下的定时任务需求。通常情况下，你可以使用 Executors.newScheduledThreadPool(int corePoolSize) 方法来创建一个 ScheduledExecutorService 实例，然后通过调用其方法来提交定时任务。