Java ThreadPoolExecutor任务优雅停止策略：避免无限运行

import java.math.BigInteger;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

// 原始的PrimeProducer实现，存在问题
public class PrimeProducer implements Runnable {
    private final BlockingQueue<BigInteger> queue;

    PrimeProducer(BlockingQueue<BigInteger> queue) {
        this.queue = queue;
    }

    @Override
    public void run() {
        try {
            BigInteger p = BigInteger.ONE;
            // 期望通过中断停止循环
            while (!Thread.currentThread().isInterrupted()) {
                queue.put(p = p.nextProbablePrime());
            }
        } catch (InterruptedException e) {
            // 捕获中断异常，通常在此处清理资源或退出
            System.out.println("PrimeProducer interrupted.");
        }
    }

    // 尝试取消任务的方法
    public void cancel() {
        // 错误：这里中断的是调用cancel()方法的线程，而不是执行run()的工作线程
        Thread.currentThread().interrupt();
    }

    public void get() {
        for (BigInteger i : queue) {
            System.out.println(i.toString());
        }
    }

    public static void main(String[] args) {
        PrimeProducer generator = new PrimeProducer(new ArrayBlockingQueue<>(10));
        ExecutorService exec = Executors.newFixedThreadPool(1);
        exec.execute(generator); // 任务由线程池的工作线程执行

        try {
            Thread.sleep(1000); // 运行1秒
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt(); // 重新中断当前线程
        } finally {
            generator.cancel(); // 尝试取消任务
        }

        exec.shutdown(); // 关闭线程池
        System.out.println("Main thread exiting.");
    }
}

import java.math.BigInteger;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit; // 引入TimeUnit

public class PrimeProducer implements Runnable {

    private final BlockingQueue<BigInteger> queue;
    private volatile boolean cancelled; // 使用volatile标志

    PrimeProducer(BlockingQueue<BigInteger> queue) {
        this.queue = queue;
    }

    @Override
    public void run() {
        try {
            BigInteger p = BigInteger.ONE;
            // 循环检查cancelled标志
            while (!cancelled) {
                // 如果任务被取消，且put操作阻塞，InterruptedException会抛出
                queue.put(p = p.nextProbablePrime());
            }
        } catch (InterruptedException e) {
            // 当put操作因线程中断而抛出异常时，任务可以终止
            System.out.println("PrimeProducer interrupted during put operation.");
        } finally {
            System.out.println("PrimeProducer stopped.");
        }
    }

    // 设置取消标志
    public void cancel() {
        cancelled = true;
    }

    public void get() {
        for (BigInteger i : queue) {
            System.out.println(i.toString());
        }
    }

    public static void main(String[] args) {
        PrimeProducer generator = new PrimeProducer(new ArrayBlockingQueue<>(10));
        ExecutorService exec = Executors.newFixedThreadPool(1);
        exec.execute(generator);

        try {
            Thread.sleep(1000); // 运行1秒
        } catch (InterruptedException e) {
            System.out.println("Main thread interrupted.");
            Thread.currentThread().interrupt();
        } finally {
            generator.cancel(); // 设置取消标志
            System.out.println("Cancellation signal sent.");
        }

        // 尝试优雅关闭线程池，等待任务完成，最多等待5秒
        exec.shutdown();
        try {
            if (!exec.awaitTermination(5, TimeUnit.SECONDS)) {
                System.out.println("ThreadPool did not terminate in time, forcing shutdown...");
                exec.shutdownNow(); // 强制关闭
            }
        } catch (InterruptedException e) {
            System.out.println("Awaiting termination interrupted.");
            exec.shutdownNow();
            Thread.currentThread().interrupt();
        }
        System.out.println("Main thread exiting.");
    }
}