发布于 由admin

JAVA线程池shutdown后仍有任务执行的根因以及完全关闭策略

JAVA线程池shutdown后仍有任务执行的根因以及完全关闭策略

大家好，今天我们来深入探讨一个在Java并发编程中经常遇到的问题：线程池在调用 shutdown() 方法后，仍然会有任务继续执行。这个问题如果不理解清楚，可能会导致程序资源泄漏、数据不一致等严重后果。本次讲座将深入剖析其根因，并提供几种完全关闭线程池的策略。

线程池的工作机制回顾

首先，让我们简单回顾一下Java线程池的工作机制。Java提供了 ExecutorService 接口及其实现类 ThreadPoolExecutor 来管理线程池。线程池的核心组件包括：

  • 核心线程数（corePoolSize）： 线程池中始终保持的线程数量，即使它们处于空闲状态。
  • 最大线程数（maximumPoolSize）： 线程池中允许存在的最大线程数量。
  • 阻塞队列（BlockingQueue）： 用于存放等待执行的任务。
  • 拒绝策略（RejectedExecutionHandler）： 当线程池饱和时，用于处理新提交的任务。

当我们向线程池提交任务时，线程池会按照以下流程处理：

  1. 如果当前线程数小于 corePoolSize，则创建新的线程来执行任务。
  2. 如果当前线程数大于等于 corePoolSize，则将任务放入阻塞队列。
  3. 如果阻塞队列已满，且当前线程数小于 maximumPoolSize，则创建新的线程来执行任务。
  4. 如果阻塞队列已满，且当前线程数大于等于 maximumPoolSize，则执行拒绝策略。

shutdown() 方法的含义与局限性

shutdown() 方法的主要作用是：

  • 不再接受新的任务提交。 调用 shutdown() 之后，再调用 execute() 或 submit() 提交任务将会抛出 RejectedExecutionException。
  • 允许已提交的任务继续执行。 线程池会等待所有已提交的任务（包括正在执行的和在队列中等待的）执行完毕后才关闭。
  • 不会中断正在执行的任务。 shutdown() 并不会强制中断正在运行的线程。

关键在于：shutdown() 方法仅仅是停止接受新任务，并等待现有任务执行完成，它并不保证任务会立即停止，也不会中断正在执行的任务。 这就是问题的根源。

任务继续执行的根因分析

即使调用了 shutdown() 方法，仍然有任务执行的原因主要有以下几种：

  1. 阻塞队列中存在未执行的任务： shutdown() 只是不再接受新任务，但队列中已经存在的任务仍然会被线程池中的线程取出并执行。

    import java.util.concurrent.*;

public class ShutdownExample {

    public static void main(String[] args) throws InterruptedException {
        ExecutorService executor = Executors.newFixedThreadPool(2);

        // 提交5个任务
        for (int i = 0; i < 5; i++) {
            final int taskNumber = i;
            executor.submit(() -> {
                try {
                    System.out.println("Task " + taskNumber + " started");
                    Thread.sleep(2000); // 模拟任务执行时间
                    System.out.println("Task " + taskNumber + " finished");
                } catch (InterruptedException e) {
                    System.out.println("Task " + taskNumber + " interrupted");
                }
            });
        }

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

        // 等待所有任务完成 (最多等待10秒)
        executor.awaitTermination(10, TimeUnit.SECONDS);

        System.out.println("Executor finished");
    }
}

    在这个例子中，线程池只有两个线程，但有5个任务。 shutdown() 被调用后，线程池仍然会执行队列中剩余的3个任务。

  2. 任务内部的长时间阻塞或死循环： 如果任务内部存在长时间的阻塞操作（如网络请求、IO操作等）或者死循环，即使线程池已经调用了 shutdown()，这些任务也会一直运行下去，直到阻塞解除或循环结束。

    import java.util.concurrent.*;

public class BlockingTaskExample {

    public static void main(String[] args) throws InterruptedException {
        ExecutorService executor = Executors.newFixedThreadPool(1);

        executor.submit(() -> {
            try {
                System.out.println("Task started, entering blocking operation");
                Thread.sleep(Long.MAX_VALUE); // 模拟长时间阻塞
                System.out.println("Task finished");
            } catch (InterruptedException e) {
                System.out.println("Task interrupted");
            }
        });

        executor.shutdown();
        executor.awaitTermination(5, TimeUnit.SECONDS); // 等待5秒

        System.out.println("Executor finished"); // 可能永远不会执行到这里
    }
}

    在这个例子中，任务 Thread.sleep(Long.MAX_VALUE) 导致线程长时间阻塞，awaitTermination() 超时后，线程池仍然无法完全关闭。

  3. 任务内部提交了新的任务： 如果一个任务在执行过程中又向线程池提交了新的任务，那么这些新的任务也会被执行。即使外部已经调用了 shutdown()，这些内部提交的任务仍然会占用线程资源。

    import java.util.concurrent.*;

public class RecursiveTaskExample {

    public static void main(String[] args) throws InterruptedException {
        ExecutorService executor = Executors.newFixedThreadPool(1);

        executor.submit(() -> {
            System.out.println("Initial task started");
            // 提交新的任务
            executor.submit(() -> {
                System.out.println("Inner task started");
                try {
                    Thread.sleep(2000);
                } catch (InterruptedException e) {
                    System.out.println("Inner task interrupted");
                }
                System.out.println("Inner task finished");
            });

            try {
                Thread.sleep(1000);
            } catch (InterruptedException e) {
                System.out.println("Initial task interrupted");
            }
            System.out.println("Initial task finished");
        });

        executor.shutdown();
        executor.awaitTermination(5, TimeUnit.SECONDS);

        System.out.println("Executor finished");
    }
}

    在这个例子中，初始任务提交了一个新的任务。即使 shutdown() 被调用，内部任务仍然会被执行。 注意：在实际生产环境中，应该避免这种在任务内部提交新任务的设计，这会使线程池的管理变得复杂。

  4. 未正确处理中断信号： 如果任务没有正确处理中断信号（InterruptedException），即使线程池调用了 shutdownNow()，任务也可能继续运行。

线程池的完全关闭策略

为了确保线程池能够完全关闭，我们需要采取更严格的策略。以下是一些常用的方法：

  1. 使用 shutdownNow() 方法： shutdownNow() 方法会尝试停止所有正在执行的任务，暂停处理正在等待的任务，并返回等待执行的任务列表。 与 shutdown() 不同，shutdownNow() 会尝试中断正在执行的线程，但这仅仅是“尝试”，如果任务没有正确处理中断信号，它仍然可能继续运行。

    import java.util.concurrent.*;
import java.util.List;

public class ShutdownNowExample {

    public static void main(String[] args) throws InterruptedException {
        ExecutorService executor = Executors.newFixedThreadPool(1);

        executor.submit(() -> {
            try {
                System.out.println("Task started");
                while (!Thread.currentThread().isInterrupted()) {
                    // 模拟长时间运行的任务
                    System.out.println("Task running...");
                    Thread.sleep(1000);
                }
                System.out.println("Task finished (interrupted)");
            } catch (InterruptedException e) {
                System.out.println("Task interrupted");
            }
        });

        Thread.sleep(3000); // 等待任务运行一段时间
        List<Runnable> droppedTasks = executor.shutdownNow();

        System.out.println("Shutdown initiated, dropped tasks: " + droppedTasks.size());
        executor.awaitTermination(5, TimeUnit.SECONDS);

        System.out.println("Executor finished");
    }
}

    在这个例子中，如果任务内部没有检查 Thread.currentThread().isInterrupted()，即使调用了 shutdownNow()，任务也可能继续运行，直到手动停止。

  2. 结合 shutdownNow() 和 awaitTermination()： 先调用 shutdownNow() 尝试中断任务，然后使用 awaitTermination() 方法等待一段时间。如果在指定时间内所有任务都已完成，则线程池关闭成功；否则，可以采取进一步的措施，例如记录日志、发送告警等。

    import java.util.concurrent.*;
import java.util.List;

public class ShutdownNowAwaitExample {

    public static void main(String[] args) throws InterruptedException {
        ExecutorService executor = Executors.newFixedThreadPool(1);

        executor.submit(() -> {
            try {
                System.out.println("Task started");
                while (!Thread.currentThread().isInterrupted()) {
                    // 模拟长时间运行的任务
                    System.out.println("Task running...");
                    Thread.sleep(1000);
                }
                System.out.println("Task finished (interrupted)");
            } catch (InterruptedException e) {
                System.out.println("Task interrupted");
            }
        });

        Thread.sleep(3000); // 等待任务运行一段时间
        List<Runnable> droppedTasks = executor.shutdownNow();

        System.out.println("Shutdown initiated, dropped tasks: " + droppedTasks.size());

        if (!executor.awaitTermination(5, TimeUnit.SECONDS)) {
            System.err.println("Executor termination timed out!  Some tasks may still be running.");
            // 可以采取进一步的措施，例如记录日志、发送告警等
        }

        System.out.println("Executor finished");
    }
}

  3. 在任务中检查中断状态并及时退出： 任务应该定期检查 Thread.currentThread().isInterrupted() 的状态，如果发现线程已经被中断，则应该立即停止执行并退出。

    import java.util.concurrent.*;

public class InterruptibleTaskExample {

    public static void main(String[] args) throws InterruptedException {
        ExecutorService executor = Executors.newFixedThreadPool(1);

        executor.submit(() -> {
            try {
                System.out.println("Task started");
                while (!Thread.currentThread().isInterrupted()) {
                    // 模拟长时间运行的任务
                    System.out.println("Task running...");
                    Thread.sleep(1000); // 模拟任务执行时间
                }
                System.out.println("Task finished (interrupted)");
            } catch (InterruptedException e) {
                System.out.println("Task interrupted");
            }
        });

        Thread.sleep(3000); // 等待任务运行一段时间
        executor.shutdownNow();
        executor.awaitTermination(5, TimeUnit.SECONDS);

        System.out.println("Executor finished");
    }
}

    在这个例子中，任务通过 while (!Thread.currentThread().isInterrupted()) 循环检查中断状态，一旦检测到中断信号，就会退出循环，从而结束任务。

  4. 使用 Future 对象取消任务： ExecutorService.submit() 方法会返回一个 Future 对象，可以使用 Future.cancel(true) 方法来尝试取消任务。 cancel(true) 会尝试中断正在执行的任务，但同样依赖于任务对中断信号的处理。

    import java.util.concurrent.*;

public class FutureCancelExample {

    public static void main(String[] args) throws InterruptedException, ExecutionException {
        ExecutorService executor = Executors.newFixedThreadPool(1);

        Future<?> future = executor.submit(() -> {
            try {
                System.out.println("Task started");
                while (!Thread.currentThread().isInterrupted()) {
                    // 模拟长时间运行的任务
                    System.out.println("Task running...");
                    Thread.sleep(1000);
                }
                System.out.println("Task finished (interrupted)");
            } catch (InterruptedException e) {
                System.out.println("Task interrupted");
            }
        });

        Thread.sleep(3000); // 等待任务运行一段时间
        boolean cancelled = future.cancel(true); // 尝试取消任务
        System.out.println("Task cancelled: " + cancelled);

        executor.shutdown();
        executor.awaitTermination(5, TimeUnit.SECONDS);

        System.out.println("Executor finished");
    }
}

    在这个例子中，future.cancel(true) 会尝试中断任务，任务内部通过检查中断状态来响应取消操作。

  5. 自定义线程池： 可以通过继承 ThreadPoolExecutor 类，并重写 beforeExecute() 和 afterExecute() 方法来监控任务的执行情况，并在必要时采取措施。 这种方式可以提供更精细的控制，但需要更多的代码和更高的复杂性。

    import java.util.concurrent.*;

public class CustomThreadPoolExecutor extends ThreadPoolExecutor {

    public CustomThreadPoolExecutor(int corePoolSize, int maximumPoolSize, long keepAliveTime, TimeUnit unit, BlockingQueue<Runnable> workQueue) {
        super(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue);
    }

    @Override
    protected void beforeExecute(Thread t, Runnable r) {
        super.beforeExecute(t, r);
        System.out.println("Before executing task: " + r.toString());
    }

    @Override
    protected void afterExecute(Runnable r, Throwable t) {
        super.afterExecute(r, t);
        if (t != null) {
            System.err.println("Task threw an exception: " + t.getMessage());
        }
        System.out.println("After executing task: " + r.toString());
    }

    public static void main(String[] args) throws InterruptedException {
        BlockingQueue<Runnable> queue = new LinkedBlockingQueue<>(10);
        CustomThreadPoolExecutor executor = new CustomThreadPoolExecutor(2, 4, 10, TimeUnit.SECONDS, queue);

        for (int i = 0; i < 5; i++) {
            final int taskNumber = i;
            executor.submit(() -> {
                try {
                    System.out.println("Task " + taskNumber + " started");
                    Thread.sleep(2000);
                    System.out.println("Task " + taskNumber + " finished");
                } catch (InterruptedException e) {
                    System.out.println("Task " + taskNumber + " interrupted");
                }
            });
        }

        executor.shutdown();
        executor.awaitTermination(10, TimeUnit.SECONDS);

        System.out.println("Executor finished");
    }
}

    在这个例子中，beforeExecute() 和 afterExecute() 方法分别在任务执行前后被调用，可以用于监控任务的执行状态。

各种方法的比较

以下表格总结了上述各种关闭策略的特点：

方法 说明 优点 缺点 适用场景
shutdown() 停止接受新任务，等待现有任务执行完成。 简单易用，不会强制中断任务。 无法保证线程池立即关闭，如果任务内部存在阻塞或死循环，线程池可能永远无法关闭。 任务可以正常完成，对关闭时间没有严格要求。
shutdownNow() 尝试停止所有正在执行的任务，暂停处理正在等待的任务，并返回等待执行的任务列表。 可以更快地关闭线程池。 依赖于任务对中断信号的处理，如果任务没有正确处理中断信号，它仍然可能继续运行。 需要尽快关闭线程池，但不介意任务被中断。
shutdownNow() + awaitTermination() 先调用 shutdownNow() 尝试中断任务，然后使用 awaitTermination() 方法等待一段时间。 可以尝试快速关闭线程池，并在超时后采取进一步措施。 仍然依赖于任务对中断信号的处理，如果任务没有正确处理中断信号，线程池可能无法完全关闭。 需要尽快关闭线程池，并对关闭失败的情况进行处理。
任务中检查中断状态 任务定期检查 Thread.currentThread().isInterrupted() 的状态，如果发现线程已经被中断，则立即停止执行并退出。 确保任务可以响应中断信号，从而可以更好地配合 shutdownNow() 方法。 需要修改任务代码，增加中断检查的逻辑。 所有的并发任务。
Future.cancel(true) 尝试取消任务。 可以针对单个任务进行取消操作。 仍然依赖于任务对中断信号的处理，如果任务没有正确处理中断信号，取消操作可能无效。 需要对单个任务进行控制，例如取消长时间运行的任务。
自定义线程池 通过继承 ThreadPoolExecutor 类，并重写 beforeExecute() 和 afterExecute() 方法来监控任务的执行情况，并在必要时采取措施。 可以提供更精细的控制，例如监控任务的执行状态、记录日志等。 需要更多的代码和更高的复杂性。 需要对线程池进行高度定制，例如监控任务的执行状态、记录日志等。

最佳实践建议

  • 始终使用 shutdownNow() 和 awaitTermination() 方法结合使用。 这可以确保线程池尽可能快地关闭，并且可以在超时后采取进一步的措施。
  • 在任务中检查中断状态并及时退出。 这可以确保任务能够响应中断信号，从而可以更好地配合 shutdownNow() 方法。
  • 避免在任务内部提交新的任务。 这会使线程池的管理变得复杂。
  • 合理设置线程池的参数。 例如，corePoolSize、maximumPoolSize 和 keepAliveTime 等。
  • 使用合适的拒绝策略。 例如，CallerRunsPolicy 或 DiscardPolicy。
  • 监控线程池的状态。 例如，使用 JConsole 或 VisualVM 等工具。

总结一下

理解 shutdown() 方法的局限性至关重要。要完全关闭 Java 线程池，我们需要结合 shutdownNow() 和 awaitTermination() 方法，并在任务内部正确处理中断信号。同时，应该尽量避免在任务内部提交新的任务，并合理设置线程池的参数。通过这些策略，我们可以确保线程池能够安全可靠地关闭，避免资源泄漏和数据不一致等问题。只有真正理解线程池的工作原理，才能编写出健壮可靠的并发程序。

发表回复

您的邮箱地址不会被公开。 必填项已用 * 标注