上一篇:Lock接口 线程池* 下一篇:本系列完

5505 2018-10-06 2020-06-25

前言：这次来看下线程池。

一、概述

Java中的线程池是运用场景最多的并发框架，几乎所有需要异步或并发执行任务的程序都可以使用线程池。在开发过程中，合理地使用线程池能够带来3个好处。

  1. 降低资源消耗，通过重复利用已创建的线程降低线程创建和销毁造成的损耗。
  2. 提高响应速度，当任务到达时，任务可以不需要等到线程创建就能立即执行。
  3. 提高线程的可管理性，线程是稀缺资源，如果无限制地创建，不仅会消耗系统资源，还会降低系统的稳定性，使用线程池可以进行统一分配、调优和监控。

但是，要做到合理利用线程池，必须对其原理了如指掌。

二、实现原理

当向线程池提交一个任务之后，线程池的主要处理流程如下图

Java线程池处理流程

从图中可以看到，当提交一个新任务到线程池时，线程池的处理流程如下

  1. 线程池判断核心线程池里的线程是否都在执行任务。如果不是，则创建一个新的工作线程来执行任务。如果核心线程池里的线程都在执行任务，则进入下一个流程。
  2. 线程池判断工作队列是否已经满了。如果工作队列没有满，则将新提交的任务存储在这个工作队列里。如果工作队列满了，则进入下个流程。
  3. 线程池判断线程池的线程是否都处于工作状态。如果没有，则创建一个新的工作线程来执行任务。如果已经满了，则交给饱和策略来处理这个任务。

ThreadPoolExecutor执行execute方法分下面4中情况

  1. 如果当前运行的线程少于corePoolSize，则创建新线程来执行任务（注意，执行这一步骤需要获取全局锁）。
  2. 如果运行的线程等于或多于corePoolSize，则将任务加入BlockingQueue。
  3. 如果无法将任务加入BlockingQueue（队列已满），则创建新的线程来处理任务（注意，执行这一步骤需要获取全局锁）。
  4. 如果创建新线程将使当前运行的线程超出maximumPoolSize，任务将拒绝执行，并调用RejectExecutionHandler.rejectedExecution()方法。

ThreadPoolExecutor采取上述步骤的总体设计思路，是为了在执行execute()方法时，尽可能地避免获取全局锁（那将会是一个严重的可伸缩瓶颈）。在ThreadPoolExecutor完成预热之后（当前运行的线程数大于等于corePoolSize），几乎所有的execute()方法调用都是执行步骤2，而步骤2不需要获取全局锁。

源代码实现如下

// 来自ThreadPoolExecutor类
/**
 * Executes the given task sometime in the future.  The task
 * may execute in a new thread or in an existing pooled thread.
 *
 * If the task cannot be submitted for execution, either because this
 * executor has been shutdown or because its capacity has been reached,
 * the task is handled by the current {@code RejectedExecutionHandler}.
 *
 * @param command the task to execute
 * @throws RejectedExecutionException at discretion of
 *         {@code RejectedExecutionHandler}, if the task
 *         cannot be accepted for execution
 * @throws NullPointerException if {@code command} is null
 */
public void execute(Runnable command) {
    if (command == null)
        throw new NullPointerException();
    /*
     * Proceed in 3 steps:
     *
     * 1. If fewer than corePoolSize threads are running, try to
     * start a new thread with the given command as its first
     * task.  The call to addWorker atomically checks runState and
     * workerCount, and so prevents false alarms that would add
     * threads when it shouldn't, by returning false.
     *
     * 2. If a task can be successfully queued, then we still need
     * to double-check whether we should have added a thread
     * (because existing ones died since last checking) or that
     * the pool shut down since entry into this method. So we
     * recheck state and if necessary roll back the enqueuing if
     * stopped, or start a new thread if there are none.
     *
     * 3. If we cannot queue task, then we try to add a new
     * thread.  If it fails, we know we are shut down or saturated
     * and so reject the task.
     */
    // 获取当前运行的线程数量
    int c = ctl.get();
    // 如果工作线程数量小于corePoolSize
    if (workerCountOf(c) < corePoolSize) {
        // 如果添加工作线程成功，则直接返回，否则继续往下走
        if (addWorker(command, true))
            return;
        // 创建工作线程失败
        c = ctl.get();
    }
    // 如果线程池没关闭，且入工作队列成功
    if (isRunning(c) && workQueue.offer(command)) {
        // 注释中提到的double-check
        int recheck = ctl.get();
        if (! isRunning(recheck) && remove(command))
            reject(command);
        else if (workerCountOf(recheck) == 0)
            addWorker(null, false);
    }
    // 加入非核心线程组
    else if (!addWorker(command, false))
        reject(command);
}

线程池创建线程时，会将线程封装成工作线程Worker，Worker在执行完任务后，还会循环获取工作队列里的任务来执行。代码实现如下

// 来自ThreadPoolExecutor.Worker
private final class Worker extends AbstractQueuedSynchronizer implements Runnable {
    /** Delegates main run loop to outer runWorker  */
    public void run() {
        runWorker(this);
    }
}

/**
 * Main worker run loop.  Repeatedly gets tasks from queue and
 * executes them, while coping with a number of issues:
 *
 * 1. We may start out with an initial task, in which case we
 * don't need to get the first one. Otherwise, as long as pool is
 * running, we get tasks from getTask. If it returns null then the
 * worker exits due to changed pool state or configuration
 * parameters.  Other exits result from exception throws in
 * external code, in which case completedAbruptly holds, which
 * usually leads processWorkerExit to replace this thread.
 *
 * 2. Before running any task, the lock is acquired to prevent
 * other pool interrupts while the task is executing, and then we
 * ensure that unless pool is stopping, this thread does not have
 * its interrupt set.
 *
 * 3. Each task run is preceded by a call to beforeExecute, which
 * might throw an exception, in which case we cause thread to die
 * (breaking loop with completedAbruptly true) without processing
 * the task.
 *
 * 4. Assuming beforeExecute completes normally, we run the task,
 * gathering any of its thrown exceptions to send to afterExecute.
 * We separately handle RuntimeException, Error (both of which the
 * specs guarantee that we trap) and arbitrary Throwables.
 * Because we cannot rethrow Throwables within Runnable.run, we
 * wrap them within Errors on the way out (to the thread's
 * UncaughtExceptionHandler).  Any thrown exception also
 * conservatively causes thread to die.
 *
 * 5. After task.run completes, we call afterExecute, which may
 * also throw an exception, which will also cause thread to
 * die. According to JLS Sec 14.20, this exception is the one that
 * will be in effect even if task.run throws.
 *
 * The net effect of the exception mechanics is that afterExecute
 * and the thread's UncaughtExceptionHandler have as accurate
 * information as we can provide about any problems encountered by
 * user code.
 *
 * @param w the worker
 */
final void runWorker(Worker w) {
    Thread wt = Thread.currentThread();
    Runnable task = w.firstTask;
    w.firstTask = null;
    w.unlock(); // allow interrupts
    boolean completedAbruptly = true;
    try {
        while (task != null || (task = getTask()) != null) {
            w.lock();
            // If pool is stopping, ensure thread is interrupted;
            // if not, ensure thread is not interrupted.  This
            // requires a recheck in second case to deal with
            // shutdownNow race while clearing interrupt
            if ((runStateAtLeast(ctl.get(), STOP) ||
                 (Thread.interrupted() &&
                  runStateAtLeast(ctl.get(), STOP))) &&
                !wt.isInterrupted())
                wt.interrupt();
            try {
                beforeExecute(wt, task);
                Throwable thrown = null;
                try {
                    task.run();
                } catch (RuntimeException x) {
                    thrown = x; throw x;
                } catch (Error x) {
                    thrown = x; throw x;
                } catch (Throwable x) {
                    thrown = x; throw new Error(x);
                } finally {
                    afterExecute(task, thrown);
                }
            } finally {
                task = null;
                w.completedTasks++;
                w.unlock();
            }
        }
        completedAbruptly = false;
    } finally {
        processWorkerExit(w, completedAbruptly);
    }
}

三、线程池的使用

1、线程池的创建

我们可以通过ThreadPoolExecutor来创建一个线程池。创建一个线程池时需要输入以下几个参数，如下

  1. corePoolSize（线程池的基本大小）：当提交一个任务到线程池时，线程池会创建一个线程来执行任务，即使其他空闲的基本线程能够执行新任务也会创建线程，等到需要执行的任务数大于线程池基本大小时就不再创建。如果调用了线程池的prestartAllCoreThreads()方法，线程池会提前创建并启动所有基本线程。

  2. runnableTaskQueue（任务队列）：用于保存等待执行任务的阻塞队列。可以选择一下几个阻塞队列。

    • ArrayBlockingQueue：是一个基于数据结构的有界阻塞队列，此队列按FIFO（先进先出）原则对元素进行排序。
    • LinkedBlockingQueue：一个基于链表结构的阻塞队列，此队列按FIFO排序元素，吞吐量通常要高于ArrayBlockingQueue。静态工厂方法Executors.newFixedThreadPool()使用了这个队列。
    • SynchronousQueue：一个不存储元素的阻塞队列。每个插入操作必须等到另一个线程调用移除操作，否则插入操作一直处于阻塞状态，吞吐量通常要高于LinkedBlockingQueue，静态工厂方法Executors.newCachedThreadPool使用了这个队列。
    • PriorityBlockingQueue：一个具有优先级的无限阻塞队列。

  3. maximumPoolSize（线程池最大数量）：线程池允许创建的最大线程数。如果队列满了，并且已创建的线程数小于最大线程数，则线程池会再创建新的线程执行任务。值得注意的是，如果使用了无界的任务队列这个参数就没什么效果了。

  4. ThreadFactory：用于设置创建线程的工厂，可以通过线程框架给每个创建出来的线程设置更有意义的名字。

  5. RejectedExecutionHandler（饱和策略）：当队列和线程池都满了，说明线程池处于饱和状态，那么必须采取一种策略处理提交的新任务。这个策略默认情况下是AbortPolicy，表示无法处理新任务时抛出异常。在JDK 1.5中Java线程池框架提供了以下四种策略

    • AbortPolicy：直接抛出异常。
    • CallerRunsPolicy：使用调用者所在线程来运行任务。
    • DiscardOldestPolicy：丢弃队列里最近的一个任务，并执行当前任务。
    • DiscardPolicy：不处理，丢弃掉。

    当然，也可以根据应用场景需要来实现RejectedExecutionHandler接口自定义策略。如记录日志或持久化存储不能处理的任务。

  6. keepAliveTime（线程活动保持的时间）：线程池的工作线程（非核心线程）空闲后，保持存活的时间。所以，如果任务很多，并且每个任务执行的时间比较多，可以调大这个值，提高线程的利用率。可以通过allowCoreThreadTimeOut(true)设置核心线程的空闲时间。

  7. TimeUnit（线程活动保持时间的单位）：可选的单位有天（DAYS）、小时（HOURS）、分钟（MINUTES）、毫秒（MILLISECONDS）、微妙（MICROSECONDS）、千分之一毫秒和纳秒（NANOSECONDS，千分之一微秒）。突然明白1毫秒 = 10 ^6纳秒。

2、向线程池提交任务

可以使用两个方法向线程池提交任务，分别为execute()和submit()方法。execute()方法用于提交不需要返回值的任务，所以无法判断任务被线程池执行成功。

submit方法用于提交需要返回值的任务。线程池会返回一个Future类型的对象，通过这个future对象可以判断任务是否执行成功，并且可以通过future的get()方法来获取返回值，get()方法会阻塞当前线程直到任务完成，而使用get(long timeout, TimeUnit unit)方法则会阻塞当前线程一段时间后立即返回，这时候有可能任务没有执行完成。

public class ThreadPoolTest {

    static int i;

    public static void main(String[] args) {
        ThreadPoolExecutor threadPool = new ThreadPoolExecutor(10, 20, 1, TimeUnit.HOURS,
                new ArrayBlockingQueue<>(10), new ThreadFactory() {
            @Override
            public Thread newThread(Runnable r) {
                return new Thread(r, "线程" + ++i);
            }
        });
        threadPool.execute(new Runnable() {
            @Override
            public void run() {
                System.out.println(Thread.currentThread() + "正在执行我");
            }
        });
        // 默认参数为 ThreadPoolExecutor(nThreads, nThreads, 0L, TimeUnit.MILLISECONDS, new LinkedBlockingQueue<Runnable>(), Executors.defaultThreadFactory(), new AbortPolicy());
        ExecutorService executor = Executors.newFixedThreadPool(10);
        Future future = executor.submit(new Runnable() {
            @Override
            public void run() {
                System.out.println(Thread.currentThread() + "正在执行我");
            }
        });
        try {
            // 正常情况下返回null
            Object o = future.get();
            System.out.println("返回结果：" + o);
        } catch (InterruptedException | ExecutionException e) {
            // 处理中断异常、处理无法执行任务异常
            e.printStackTrace();
        }
        finally {
            // 不推荐threadPool.shutdownNow();返回List<Runnable>
            threadPool.shutdown();
        }
    }
}

3、关闭线程池

可以通过调用线程池的shutdown或shutdownNow方法来关闭线程池。它们的原理是遍历线程池中的工作线程，然后逐个调用线程的interrupt方法来中断线程，所以无法响应中断的任务可能永远无法终止。但是它们存在一定的区别，shutdownNow首先将线程池状态设置成STOP，然后尝试停止所有的正在执行或暂停任务线程，并返回等待执行任务的列表，而shutdown只是将线程池的状态设置成SHUTDOWN状态，然后中断所有没有正在执行任务的线程。

只要调用了这两个关闭方法中的任意一个，isShutdown方法就会返回true。当所有的任务都已关闭后，才表示线程池关闭成功，这时调用isTerminaed方法会返回true。至于应该调用哪种方法来关闭线程池，应该由线程池的任务特性决定，通常调用shutdown方法来关闭线程池，如果任务不一定要执行完，则可以调用shutdownNow方法（通俗的理解，shutdownNow是强制关闭，返回为未执行任务列表；而shutdown会等待当前任务执行完毕才真正关闭）。

/**  
 *   RUNNING:  Accept new tasks and process queued tasks
 *   SHUTDOWN: Don't accept new tasks, but process queued tasks
 *   STOP:     Don't accept new tasks, don't process queued tasks,
 *             and interrupt in-progress tasks
 *   TIDYING:  All tasks have terminated, workerCount is zero,
 *             the thread transitioning to state TIDYING
 *             will run the terminated() hook method
 *   TERMINATED: terminated() has completed
 */
// runState is stored in the high-order bits
private static final int RUNNING    = -1 << COUNT_BITS;
private static final int SHUTDOWN   =  0 << COUNT_BITS;
private static final int STOP       =  1 << COUNT_BITS;
private static final int TIDYING    =  2 << COUNT_BITS;
private static final int TERMINATED =  3 << COUNT_BITS;

4、合理地配置线程池

要想合理地配置线程池，就必须首先分析任务特性，可以从以下几个角度来分析。

  • 任务的性质：CPU密集型任务、IO密集型任务和混合型任务。
  • 任务的优先级：高、中和低。
  • 任务的执行时间：长、中和短。
  • 任务的依赖性：是否依赖其他系统资源，如数据库连结。

性质不同的任务可以用不同规模的线程池分开处理。CPU密集性任务应配置尽可能小的线程，如配置N(cpu) + 1个线程的线程池。由于IO密集型任务线程并不是一直在执行任务，则应配置尽可能多的线程，如2 × N(cpu)。混合型任务，如果可以拆分，将其拆分成一个CPU密集型任务和一个IO密集型任务，只要这两个任务执行的时间相差不是太大，那么分解后执行的吞吐量将高于串行执行的吞吐量。如果这两个任务执行时间相差不是很大，则没必要进行分解。可以通过Runtime.getRuntime().availableProcessors()方法获取当前设备的CPU个数。

优先级不同的任务可以使用优先级队列PriorityBlockingQueue来处理。他可以让优先级高的任务先执行。

执行时间不同的任务可以交给不同规模的线程池来处理，或者可以使用优先级队列，让执行时间短的任务先执行。

依赖数据库连接池的任务，因为线程提交SQL后需要等待数据库返回结果，等待的时间越长，则CPU空闲的时间就越长，那么线程数应该设置得越大，这样才能更好地利用CPU。

建议使用有界队列。有界队列能增加系统的稳定性和预警能力，可以根据需要设大一点，比如几千。

5、线程池的监控

如果在系统中大量使用线程池，则有必要对线程池进行监控，方便在出现问题时，可以根据线程池的使用状况快速定位问题。可以通过线程池提供的参数进行监控，在监控线程池的时候可以使用以下属性。

  • taskCount：线程池需要执行的任务数量。
  • completedTaskCount：线程池在运行过程中已完成的任务数量，小于或等于taskCount。
  • largestPoolSize：线程池曾经创建过的最大线程数量。通过这个数据可以知道线程是否曾经满过。如该数值等于线程池的最大大小，则表示线程池曾经满过。
  • getPoolSize：线程池的线程数量。如果线程池不销毁的话，线程池里的线程不会自动销毁，所以这个值只增不减。
  • getActiveCount：获取活动的线程数。

通过扩展线程池进行监控。可以通过继承线程池来自定义线程池，重写线程池的beforeExecute、afterExecute和terminated方法，也可以在任务执行前、执行后和线程池关闭前执行一些代码来进行监控。例如，监控任务的平均执行时间、最大执行时间和最小执行时间等。这几个方法在线程池里都是空方法。

四、实例演示

1、基本使用

public class ThreadPoolTest {

    static int i = 0;

    public static void main(String[] args) throws ExecutionException, InterruptedException {
        Runnable runnable = new Runnable() {
            @Override
            public void run() {
                System.out.println(Thread.currentThread().getName() + "打印了这条信息");
            }
        };
        ThreadPoolExecutor executor = new ThreadPoolExecutor(10, 20, 1, TimeUnit.HOURS,
                new ArrayBlockingQueue<>(10), new ThreadFactory() {
            @Override
            public Thread newThread(@NotNull Runnable r) {
                return new Thread(r, "线程" + i++);
            }
        });
        executor.execute(runnable);
        System.out.println(executor.getActiveCount());
        executor.shutdown();

        ThreadPoolExecutor executor1 = new ThreadPoolExecutor(5, 10, 10, TimeUnit.MINUTES,
                new LinkedBlockingDeque<>(10), new ThreadFactory() {
            @Override
            public Thread newThread(@NotNull Runnable r) {
                return new Thread(r,"线程" + i++);
            }
        }, new ThreadPoolExecutor.DiscardPolicy());
        executor1.prestartAllCoreThreads();
        executor1.execute(runnable);

        ThreadPoolExecutor executor2 = new ThreadPoolExecutor(5, 10, 10, TimeUnit.MINUTES,
                new PriorityBlockingQueue<>(10, new Comparator<Runnable>() {
                    @Override
                    public int compare(Runnable o1, Runnable o2) {
                        return 0;
                    }
                }), new ThreadFactory() {
            @Override
            public Thread newThread(@NotNull Runnable r) {
                return new Thread(r,"线程" + i++);
            }
        }, new ThreadPoolExecutor.DiscardPolicy());
        executor2.execute(runnable);

        ThreadPoolExecutor executor3 = new ThreadPoolExecutor(5, 10, 10, TimeUnit.MINUTES,
                new SynchronousQueue<>(true), new ThreadFactory() {
            @Override
            public Thread newThread(@NotNull Runnable r) {
                Thread thread = new Thread(r,"线程" + i++);
                thread.setDaemon(true);
                return thread;
            }
        }, new ThreadPoolExecutor.DiscardOldestPolicy());
        executor3.execute(runnable);
        
        ExecutorService executorService = Executors.newFixedThreadPool(10);
        Future future = executorService.submit(runnable);
        System.out.println(future.get());
    }
}

2、赛场跑步

考虑如下场景：10个运动员在赛场跑步，直到10个运动员全部到场后，宣布比赛终止。

public class Pool1 {

    public static void main(String[] args) throws InterruptedException {
        // 计数器为0时，返回await。调用latch.countDown()则减1
        CountDownLatch latch = new CountDownLatch(10);
        // 当barrier.await()线程等于10时，打开栏栅，放行所有线程
        CyclicBarrier barrier = new CyclicBarrier(10);
        ExecutorService executor = Executors.newFixedThreadPool(10);
        for (int i = 0; i < 10; i++) {
            executor.execute(new Thread("少年" + i) {
                @Override
                public void run() {
                    try {
                        barrier.await();
                        System.out.println(Thread.currentThread().getName() + "开始出发（同时）");
                        Thread.sleep(1000);
                        System.out.println(Thread.currentThread().getName() + "到达终点");
                        latch.countDown();
                    } catch (InterruptedException | BrokenBarrierException e) {
                        e.printStackTrace();
                    }
                }
            });
        }
        latch.await();
        System.out.println("比赛完成");
    }
}
// 输出如下
pool-1-thread-1开始出发（同时）
pool-1-thread-2开始出发（同时）
pool-1-thread-4开始出发（同时）
pool-1-thread-5开始出发（同时）
pool-1-thread-6开始出发（同时）
pool-1-thread-7开始出发（同时）
pool-1-thread-8开始出发（同时）
pool-1-thread-9开始出发（同时）
pool-1-thread-10开始出发（同时）
pool-1-thread-3开始出发（同时）
pool-1-thread-1到达终点
pool-1-thread-2到达终点
pool-1-thread-5到达终点
pool-1-thread-6到达终点
pool-1-thread-4到达终点
pool-1-thread-8到达终点
pool-1-thread-7到达终点
pool-1-thread-9到达终点
pool-1-thread-10到达终点
pool-1-thread-3到达终点
比赛完成

3、购买奢侈品

考虑如下场景：某奢侈品为保证购物体验，最多只能进入6人，其他人必须在外等候，另一个出来了才能进去一个。

public class Pool2 {

    static int j = 0;
    // 当barrier.await()线程等于10时，打开栏栅，放行所有线程
    static CyclicBarrier barrier = new CyclicBarrier(10);
    // 只允许6个线程
    static Semaphore semaphore = new Semaphore(6, false);

    public static void visit() throws InterruptedException, BrokenBarrierException {
        barrier.await();
        semaphore.acquire();
        try {
            System.out.println(Thread.currentThread().getName() + "进去购物奢侈品");
            Thread.sleep(500);
            System.out.println(Thread.currentThread().getName() + "购物奢侈品成功，出来了 ");
        } catch (InterruptedException e) {
            e.printStackTrace();
        } finally {
            semaphore.release();
        }
    }

    public static void main(String[] args) {
        ExecutorService executor = Executors.newFixedThreadPool(10, new ThreadFactory() {
            @Override
            public Thread newThread(@NotNull Runnable r) {
                return new Thread(r, "顾客" + ++j);
            }
        });
        for (int i = 0; i < 10; i++) {
            executor.execute(new Runnable() {
                @SneakyThrows
                @Override
                public void run() {
                    Pool2.visit();
                }
            });
        }
    }
}
// 输出如下
顾客1进去购物奢侈品
顾客3进去购物奢侈品
顾客6进去购物奢侈品
顾客2进去购物奢侈品
顾客5进去购物奢侈品
顾客4进去购物奢侈品
顾客1购物奢侈品成功，出来了 
顾客2购物奢侈品成功，出来了 
顾客7进去购物奢侈品
顾客8进去购物奢侈品
顾客6购物奢侈品成功，出来了 
顾客9进去购物奢侈品
顾客5购物奢侈品成功，出来了 
顾客3购物奢侈品成功，出来了 
顾客4购物奢侈品成功，出来了 
顾客10进去购物奢侈品
顾客7购物奢侈品成功，出来了 
顾客8购物奢侈品成功，出来了 
顾客9购物奢侈品成功，出来了 
顾客10购物奢侈品成功，出来了

五、异步编程

这里简单提一下，不做深入，就不单独拿一个章节了。

1、Future

public class FutureTest {

    static int i = 0;

    public static void main(String[] args) throws ExecutionException, InterruptedException {
        ThreadFactory factory = new ThreadFactory() {
            @Override
            public Thread newThread(@NotNull Runnable r) {
                return new Thread(r, "线程" + ++i);
            }
        };
        Callable<Integer> callable = new Callable<Integer>() {
            @Override
            public Integer call() throws Exception {
                Thread.sleep(4000);
                return 1;            }
        };
        ThreadPoolExecutor executor = new ThreadPoolExecutor(5, 10, 10, TimeUnit.MINUTES,
                new ArrayBlockingQueue<>(10), factory, new ThreadPoolExecutor.CallerRunsPolicy());
        executor.allowCoreThreadTimeOut(true);


        Future<Integer> future = executor.submit(callable);
        long start = System.nanoTime();
        System.out.println("当前时间：" + start);
        // 会阻塞在这里，相当于Thread.join()
        System.out.println(future.get());
        System.out.println("耗时" + (TimeUnit.NANOSECONDS.toMillis(System.nanoTime() - start)) + "毫秒");

        future = executor.submit(callable);
        start = System.nanoTime();
        System.out.println("当前时间：" + start);
        while (!future.isDone()) {
        }
        System.out.println(future.get());
        System.out.println("耗时" + (TimeUnit.NANOSECONDS.toMillis(System.nanoTime() - start)) + "毫秒");
    }
}
// 输出如下
当前时间：6974644856252
1
耗时4000毫秒
当前时间：6978645893583
1
耗时4000毫秒

当调用future的get()方法时，当前主线程是堵塞的。另一种获取返回结果的方式是调用isDone轮询，等完成再获取。虽然能达到目的，但未免有点有失优雅。

2、CompletableFuture

CompletableFuture就是用来解决上面问题的，而且提供了强大的任务编排能力。

1、串行执行

public class SerialThread {

    public static void main(String[] args) throws ExecutionException, InterruptedException {
        // 默认使用CompletableFuture内部默认的线程池
        // useCommonPool ? ForkJoinPool.commonPool() : new ThreadPerTaskExecutor();
        CompletableFuture<Integer> future = CompletableFuture.supplyAsync(new Supplier<Integer>() {
            @SneakyThrows
            @Override
            public Integer get() {
                Thread.sleep(1000);
                System.out.println("任务一执行完毕");
                return 1;
            }
        }).thenApplyAsync(new Function<Integer, Integer>() {
            @SneakyThrows
            @Override
            public Integer apply(Integer integer) {
                Thread.sleep(1000);
                System.out.println("任务二执行完毕");
                return integer + 2;
            }
        });
        System.out.println(future.get());
    }
}
// 输出如下
任务一执行完毕
任务二执行完毕
3

2、返回执行最快的那个

public class ParallelThread {

    public static void main(String[] args) throws ExecutionException, InterruptedException {
        CompletableFuture<String> server1 = CompletableFuture.supplyAsync(new Supplier<String>() {
            @SneakyThrows
            @Override
            public String get() {
                Thread.sleep(1000);
                System.out.println("服务器一返回消息");
                return "服务器一";
            }
        });
        System.out.println(11);
        CompletableFuture<String> server2 = CompletableFuture.supplyAsync(new Supplier<String>() {
            @SneakyThrows
            @Override
            public String get() {
                Thread.sleep(1000);
                System.out.println("服务器二返回消息");
                return "服务器二";
            }
        });
        System.out.println(22);
        CompletableFuture<String> result = server2.applyToEither(server1, new Function<String, String>() {
            @Override
            public String apply(String s) {
                return s + "返回消息更快";
            }
        });
        System.out.println(33);
        System.out.println(result.get());
        // 仍会执行所有任务
        Thread.sleep(4000);

    }
}
// 输出如下
11
22
33
服务器一返回消息
服务器一返回消息更快
服务器二返回消息

3、并行执行

public class ParallelThread1 {

    public static void main(String[] args) throws ExecutionException, InterruptedException {
        CompletableFuture f1 = CompletableFuture.runAsync(()->{
            try {
                TimeUnit.MILLISECONDS.sleep(100);
                System.out.println("execute f1");
            } catch (InterruptedException e) {
                throw new RuntimeException(e);
            }
        });
        CompletableFuture f2 = CompletableFuture.runAsync(()->{
            try {
                TimeUnit.MILLISECONDS.sleep(1000);
                System.out.println("execute f2");
            } catch (InterruptedException e) {
                throw new RuntimeException(e);
            }
        });

        CompletableFuture all =  CompletableFuture.allOf(f1,f2);
        all.get();
        System.out.println("execute all");
    }
}

这里只是抛砖引玉，更多CompletableFuture的特性有待读者自己去发掘。

总访问次数: 410次, 一般般帅 创建于 2018-10-06, 最后更新于 2020-06-25

进大厂！ 欢迎关注微信公众号，第一时间掌握最新动态！