MK
摩柯社区 - 一个极简的技术知识社区
AI 面试

Java编程中的线程池与异步任务管理

2022-04-206.8k 阅读

Java编程中的线程池与异步任务管理

线程池基础概念

在Java编程中，线程是程序执行的最小单位，多个线程可以并发执行不同的任务，从而提高程序的执行效率。然而，如果频繁地创建和销毁线程，会带来额外的系统开销，如线程创建和销毁的时间消耗、内存分配与回收等。线程池则是一种用于管理线程的机制，它通过预先创建一定数量的线程，并将这些线程保存在一个“池”中，当有任务需要执行时，从线程池中取出一个空闲线程来执行任务，任务执行完毕后，线程并不会被销毁，而是返回线程池等待下一个任务。

线程池的优点主要体现在以下几个方面：

  1. 降低资源消耗：避免了频繁创建和销毁线程带来的开销，提高了系统资源的利用率。
  2. 提高响应速度：当有任务到达时，无需等待线程创建，直接从线程池中获取线程执行任务，加快了任务的响应时间。
  3. 便于管理：可以对线程池中的线程进行统一的管理，如设置线程池的大小、线程的优先级、任务队列的容量等，从而更好地控制并发执行的任务数量和资源使用情况。

Java中的线程池实现

在Java中，线程池的实现主要依赖于java.util.concurrent包中的Executor框架。Executor框架提供了一系列接口和类来支持异步任务的执行和线程池的管理。下面介绍几个关键的接口和类。

  1. Executor接口：这是Executor框架的基础接口，它定义了一个简单的方法execute(Runnable task)，用于提交一个可运行的任务。该接口的实现类负责执行提交的任务，但并不关心任务的执行方式（如是否使用线程池、是否在当前线程执行等）。
public interface Executor {
    void execute(Runnable task);
}
  1. ExecutorService接口：ExecutorService接口继承自Executor接口，它提供了更丰富的方法来管理线程池的生命周期，如关闭线程池、提交可返回结果的任务等。
public interface ExecutorService extends Executor {
    void shutdown();
    List<Runnable> shutdownNow();
    boolean isShutdown();
    boolean isTerminated();
    <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;
}

  1. AbstractExecutorService类：这是一个抽象类，它实现了ExecutorService接口的部分方法，为具体的线程池实现类提供了基础的实现框架。

  2. ThreadPoolExecutor类：ThreadPoolExecutor是ExecutorService接口的主要实现类，它提供了灵活的线程池配置和管理功能。通过ThreadPoolExecutor，可以设置核心线程数、最大线程数、线程存活时间、任务队列等参数，以满足不同场景下的需求。

public class ThreadPoolExecutor extends AbstractExecutorService {
    public ThreadPoolExecutor(int corePoolSize,
                              int maximumPoolSize,
                              long keepAliveTime,
                              TimeUnit unit,
                              BlockingQueue<Runnable> workQueue) {
        this(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue,
             Executors.defaultThreadFactory(), defaultHandler);
    }
    // 其他构造函数和方法
}

在上述构造函数中：

  • corePoolSize：核心线程数，线程池中会始终保持的线程数量，即使这些线程处于空闲状态。
  • maximumPoolSize：最大线程数，线程池允许创建的最大线程数量。
  • keepAliveTime：线程存活时间，当线程池中的线程数量超过核心线程数时，多余的空闲线程在被销毁之前等待新任务的最长时间。
  • unit：keepAliveTime的时间单位。
  • workQueue：任务队列，用于存储提交但尚未被执行的任务。常用的任务队列实现类有ArrayBlockingQueue、LinkedBlockingQueue、SynchronousQueue等。

线程池的工作流程

当一个任务提交到线程池时，线程池的工作流程如下：

  1. 检查核心线程数：线程池首先检查当前运行的线程数是否小于核心线程数。如果小于核心线程数，则创建一个新的线程来执行任务。
  2. 检查任务队列：如果当前运行的线程数已经达到核心线程数，则将任务放入任务队列中等待执行。
  3. 检查最大线程数：如果任务队列已满，线程池会检查当前运行的线程数是否小于最大线程数。如果小于最大线程数，则创建一个新的线程来执行任务。
  4. 执行拒绝策略：如果当前运行的线程数已经达到最大线程数，并且任务队列也已满，此时线程池无法再接受新的任务，会执行拒绝策略。Java提供了四种默认的拒绝策略：
    • ThreadPoolExecutor.AbortPolicy：这是默认的拒绝策略，当任务无法处理时，抛出RejectedExecutionException异常。
    • ThreadPoolExecutor.CallerRunsPolicy：将任务交给调用者线程来执行，即谁提交的任务，就由谁所在的线程来执行该任务。
    • ThreadPoolExecutor.DiscardPolicy：直接丢弃任务，不做任何处理。
    • ThreadPoolExecutor.DiscardOldestPolicy：丢弃任务队列中最老的一个任务，然后尝试重新提交当前任务。

线程池的创建方式

  1. 使用ThreadPoolExecutor直接创建：通过ThreadPoolExecutor的构造函数可以根据具体需求精确地配置线程池的参数。
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;

public class CustomThreadPoolExample {
    public static void main(String[] args) {
        // 创建任务队列
        BlockingQueue<Runnable> workQueue = new LinkedBlockingQueue<>(10);
        // 创建线程池
        ThreadPoolExecutor executor = new ThreadPoolExecutor(
                5, // 核心线程数
                10, // 最大线程数
                10, // 线程存活时间
                TimeUnit.SECONDS,
                workQueue
        );

        // 提交任务
        for (int i = 0; i < 20; i++) {
            int taskNumber = i;
            executor.submit(() -> {
                System.out.println("Task " + taskNumber + " is running on thread " + Thread.currentThread().getName());
                try {
                    Thread.sleep(1000);
                } catch (InterruptedException e) {
                    Thread.currentThread().interrupt();
                }
            });
        }

        // 关闭线程池
        executor.shutdown();
    }
}

在上述代码中，我们创建了一个核心线程数为5，最大线程数为10，线程存活时间为10秒，任务队列容量为10的线程池。然后提交了20个任务，观察线程池的执行情况。

  1. 使用Executors工具类创建：Executors工具类提供了一些静态方法来快速创建不同类型的线程池，这些方法底层也是基于ThreadPoolExecutor实现的，但在配置上做了一些预设。
    • newFixedThreadPool(int nThreads)：创建一个固定大小的线程池，核心线程数和最大线程数都为nThreads，任务队列使用LinkedBlockingQueue。
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

public class FixedThreadPoolExample {
    public static void main(String[] args) {
        // 创建固定大小的线程池
        ExecutorService executor = Executors.newFixedThreadPool(5);

        // 提交任务
        for (int i = 0; i < 10; i++) {
            int taskNumber = i;
            executor.submit(() -> {
                System.out.println("Task " + taskNumber + " is running on thread " + Thread.currentThread().getName());
                try {
                    Thread.sleep(1000);
                } catch (InterruptedException e) {
                    Thread.currentThread().interrupt();
                }
            });
        }

        // 关闭线程池
        executor.shutdown();
    }
}

- **newCachedThreadPool()**：创建一个可缓存的线程池，核心线程数为0，最大线程数为`Integer.MAX_VALUE`，线程存活时间为60秒，任务队列使用`SynchronousQueue`。如果线程池中的线程在60秒内没有被使用，就会被销毁。当有新任务提交时，如果有空闲线程则复用，否则创建新线程。

import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

public class CachedThreadPoolExample {
    public static void main(String[] args) {
        // 创建可缓存的线程池
        ExecutorService executor = Executors.newCachedThreadPool();

        // 提交任务
        for (int i = 0; i < 10; i++) {
            int taskNumber = i;
            executor.submit(() -> {
                System.out.println("Task " + taskNumber + " is running on thread " + Thread.currentThread().getName());
                try {
                    Thread.sleep(1000);
                } catch (InterruptedException e) {
                    Thread.currentThread().interrupt();
                }
            });
        }

        // 关闭线程池
        executor.shutdown();
    }
}

- **newSingleThreadExecutor()**：创建一个单线程的线程池，核心线程数和最大线程数都为1，任务队列使用`LinkedBlockingQueue`。该线程池始终只有一个线程在执行任务，所有任务按照提交顺序依次执行。

import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

public class SingleThreadExecutorExample {
    public static void main(String[] args) {
        // 创建单线程的线程池
        ExecutorService executor = Executors.newSingleThreadExecutor();

        // 提交任务
        for (int i = 0; i < 10; i++) {
            int taskNumber = i;
            executor.submit(() -> {
                System.out.println("Task " + taskNumber + " is running on thread " + Thread.currentThread().getName());
                try {
                    Thread.sleep(1000);
                } catch (InterruptedException e) {
                    Thread.currentThread().interrupt();
                }
            });
        }

        // 关闭线程池
        executor.shutdown();
    }
}

虽然Executors工具类提供了方便的线程池创建方式，但在实际应用中，特别是对于高并发、对性能要求较高的场景，建议使用ThreadPoolExecutor直接创建线程池，以便根据具体需求精确配置参数，避免因默认配置不合理而导致性能问题。

异步任务管理

在Java中，除了使用线程池来执行任务外，还涉及到异步任务的管理，即如何提交任务、获取任务执行结果、处理任务执行过程中的异常等。

  1. 使用Runnable接口提交任务：Runnable接口是Java中定义任务的一种方式，它只有一个run()方法，任务的逻辑在run()方法中实现。通过Executor或ExecutorService的execute()方法可以提交Runnable任务，但Runnable任务没有返回值，也无法抛出受检异常。
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

public class RunnableTaskExample {
    public static void main(String[] args) {
        ExecutorService executor = Executors.newFixedThreadPool(5);

        // 创建Runnable任务
        Runnable task = () -> {
            System.out.println("Runnable task is running on thread " + Thread.currentThread().getName());
            try {
                Thread.sleep(1000);
            } catch (InterruptedException e) {
                Thread.currentThread().interrupt();
            }
        };

        // 提交任务
        executor.execute(task);

        // 关闭线程池
        executor.shutdown();
    }
}
  1. 使用Callable接口提交任务：Callable接口与Runnable接口类似，但它的call()方法可以返回一个结果，并且可以抛出受检异常。通过ExecutorService的submit()方法提交Callable任务后，会返回一个Future对象，通过该对象可以获取任务的执行结果、判断任务是否完成、取消任务等。
import java.util.concurrent.*;

public class CallableTaskExample {
    public static void main(String[] args) {
        ExecutorService executor = Executors.newFixedThreadPool(5);

        // 创建Callable任务
        Callable<Integer> task = () -> {
            System.out.println("Callable task is running on thread " + Thread.currentThread().getName());
            Thread.sleep(1000);
            return 42;
        };

        // 提交任务并获取Future对象
        Future<Integer> future = executor.submit(task);

        try {
            // 获取任务执行结果，该方法会阻塞直到任务完成
            Integer result = future.get();
            System.out.println("Task result: " + result);
        } catch (InterruptedException | ExecutionException e) {
            e.printStackTrace();
        }

        // 关闭线程池
        executor.shutdown();
    }
}

在上述代码中，Callable任务返回一个整数结果，通过Future的get()方法获取该结果。如果任务尚未完成，get()方法会阻塞当前线程，直到任务完成并返回结果。

  1. 使用FutureTask类：FutureTask类实现了Future接口和Runnable接口，因此它既可以作为Runnable任务提交到线程池执行，也可以通过Future接口获取任务执行结果。
import java.util.concurrent.*;

public class FutureTaskExample {
    public static void main(String[] args) {
        ExecutorService executor = Executors.newFixedThreadPool(5);

        // 创建Callable任务
        Callable<Integer> callableTask = () -> {
            System.out.println("Callable task is running on thread " + Thread.currentThread().getName());
            Thread.sleep(1000);
            return 42;
        };

        // 创建FutureTask对象
        FutureTask<Integer> futureTask = new FutureTask<>(callableTask);

        // 提交FutureTask任务
        executor.submit(futureTask);

        try {
            // 获取任务执行结果，该方法会阻塞直到任务完成
            Integer result = futureTask.get();
            System.out.println("Task result: " + result);
        } catch (InterruptedException | ExecutionException e) {
            e.printStackTrace();
        }

        // 关闭线程池
        executor.shutdown();
    }
}
  1. 处理异步任务的异常：当Callable任务抛出异常时，可以在Future的get()方法中捕获ExecutionException，该异常的getCause()方法可以获取任务抛出的原始异常。
import java.util.concurrent.*;

public class ExceptionHandlingExample {
    public static void main(String[] args) {
        ExecutorService executor = Executors.newFixedThreadPool(5);

        // 创建Callable任务，故意抛出异常
        Callable<Integer> task = () -> {
            System.out.println("Callable task is running on thread " + Thread.currentThread().getName());
            throw new RuntimeException("Task failed");
        };

        // 提交任务并获取Future对象
        Future<Integer> future = executor.submit(task);

        try {
            Integer result = future.get();
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
        } catch (ExecutionException e) {
            System.out.println("Task execution failed: " + e.getCause());
        }

        // 关闭线程池
        executor.shutdown();
    }
}

在上述代码中，Callable任务故意抛出一个RuntimeException，在Future的get()方法中捕获ExecutionException，并打印出原始异常信息。

  1. CompletableFuture类：CompletableFuture是Java 8引入的一个强大的异步编程工具，它扩展了Future接口，提供了更丰富的异步任务处理方法，如链式调用、异步任务组合、异常处理等。
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ExecutionException;

public class CompletableFutureExample {
    public static void main(String[] args) {
        // 创建一个异步任务
        CompletableFuture<Integer> future = CompletableFuture.supplyAsync(() -> {
            System.out.println("CompletableFuture task is running on thread " + Thread.currentThread().getName());
            try {
                Thread.sleep(1000);
            } catch (InterruptedException e) {
                Thread.currentThread().interrupt();
            }
            return 42;
        });

        // 处理任务结果
        future.thenApply(result -> {
            System.out.println("Result processed on thread " + Thread.currentThread().getName());
            return result * 2;
        }).thenAccept(finalResult -> {
            System.out.println("Final result: " + finalResult);
        }).exceptionally(ex -> {
            System.out.println("Task failed: " + ex);
            return null;
        });

        // 主线程等待一段时间，确保异步任务有足够时间执行
        try {
            Thread.sleep(2000);
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
        }
    }
}

在上述代码中，CompletableFuture.supplyAsync()方法创建了一个异步任务，thenApply()方法对任务结果进行处理，thenAccept()方法消费最终结果，exceptionally()方法处理任务执行过程中的异常。通过这些方法的链式调用，可以方便地构建复杂的异步任务处理流程。

线程池与异步任务管理的最佳实践

  1. 合理配置线程池参数：根据应用程序的特点和硬件资源，合理设置核心线程数、最大线程数、任务队列容量等参数。如果核心线程数设置过小，可能导致任务排队等待，影响响应速度；如果设置过大，可能会占用过多的系统资源。最大线程数也需要根据系统负载和任务类型进行调整，避免线程过多导致系统性能下降。
  2. 选择合适的任务队列：不同的任务队列适用于不同的场景。ArrayBlockingQueue是一个有界队列，适用于需要限制任务数量的场景；LinkedBlockingQueue可以是有界或无界的，无界队列可能会导致内存耗尽，需要谨慎使用；SynchronousQueue不存储任务，直接将任务交给线程处理，适用于任务处理速度较快，不希望任务在队列中等待的场景。
  3. 优雅地关闭线程池：在应用程序关闭时，应该优雅地关闭线程池，避免任务丢失或线程强制终止。可以通过调用ExecutorService的shutdown()方法来启动关闭过程，该方法会拒绝新的任务提交，并等待已提交的任务执行完毕。如果需要立即停止所有任务，可以调用shutdownNow()方法，但可能会导致一些任务无法正常完成。
  4. 处理异步任务的异常：在异步任务执行过程中，要妥善处理可能抛出的异常，避免异常丢失导致程序出现难以排查的问题。可以通过Future的get()方法捕获异常，或者使用CompletableFuture的exceptionally()方法进行异常处理。
  5. 避免线程池饥饿：线程池饥饿是指由于某些任务长时间占用线程，导致其他任务无法及时执行的情况。为了避免线程池饥饿，可以合理设置任务的优先级，或者采用公平调度策略，确保所有任务都有机会执行。
  6. 监控和调优：在生产环境中，对线程池的运行状态进行监控是非常重要的。可以通过ThreadPoolExecutor提供的一些方法，如getActiveCount()获取当前活动线程数、getQueue().size()获取任务队列中的任务数量等，来了解线程池的负载情况，并根据监控数据进行调优。

总结

线程池与异步任务管理是Java编程中实现高效并发编程的重要手段。通过合理使用线程池和异步任务管理机制，可以充分利用系统资源，提高程序的执行效率和响应速度。在实际应用中，需要根据具体场景选择合适的线程池类型和配置参数，并妥善处理异步任务的提交、执行和结果获取，以确保程序的稳定性和性能。同时，不断学习和实践，积累经验，才能更好地掌握这一重要的编程技术。