正确管理和重启Java线程：基于BlockingQueue的生产者-消费者模型

while(System.currentTimeMillis() < endThreadTime) {
    if(!global.isEmpty()) {
        recordToUse = global.remove();
        System.out.println("Successful removal: Thread-"+ thread_num);
    } else {
        continue; // If the queue is empty, keep checking until it is not empty.
    }
    // ... more operations
}

import java.util.concurrent.BlockingQueue;
import java.util.concurrent.LinkedBlockingQueue;

public class MessageProcessor {
    // 使用LinkedBlockingQueue作为消息队列，它是一个无界队列，也可以指定容量
    private static final BlockingQueue<String> messageQueue = new LinkedBlockingQueue<>();

    // 定义一个特殊的“毒丸”消息，用于通知消费者线程退出
    public static final String POISON_PILL = "POISON_PILL_SHUTDOWN";

    // ... 其他组件
}

// 在你的Spring Boot服务中
public class BusinessService {

    public static void onMessage(String record) {
        try {
            messageQueue.put(record); // 使用put()方法，队列满时会阻塞
            System.out.println("Producer added message: " + record);
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt(); // 重新设置中断状态
            System.err.println("Producer was interrupted while adding message: " + e.getMessage());
        }
    }

    // ... 其他业务逻辑
}

import java.util.concurrent.BlockingQueue;

public class ConsumerWorker implements Runnable {
    private final BlockingQueue<String> queue;
    private final int workerId;

    public ConsumerWorker(BlockingQueue<String> queue, int workerId) {
        this.queue = queue;
        this.workerId = workerId;
    }

    @Override
    public void run() {
        System.out.println("ConsumerWorker-" + workerId + " started.");
        try {
            while (true) {
                String record = queue.take(); // 队列为空时，线程会在此阻塞

                // 检查是否是“毒丸”消息，用于优雅关闭
                if (MessageProcessor.POISON_PILL.equals(record)) {
                    System.out.println("ConsumerWorker-" + workerId + " received poison pill, shutting down.");
                    break; // 退出循环，线程结束
                }

                System.out.println("ConsumerWorker-" + workerId + " successfully removed and processing: " + record);
                // 模拟消息处理时间
                Thread.sleep(50 + (long)(Math.random() * 100)); // 模拟处理耗时
                // 这里可以添加更多对消息进行的操作
            }
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt(); // 重新设置中断状态
            System.err.println("ConsumerWorker-" + workerId + " was interrupted: " + e.getMessage());
        } finally {
            System.out.println("ConsumerWorker-" + workerId + " stopped.");
        }
    }
}

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

public class ApplicationRunner {
    private static ExecutorService executorService;
    private static final int NUM_CONSUMERS = 8; // 定义消费者线程的数量

    public static void main(String[] args) throws InterruptedException {
        // 初始化线程池
        executorService = Executors.newFixedThreadPool(NUM_CONSUMERS);

        // 启动消费者线程
        for (int i = 0; i < NUM_CONSUMERS; i++) {
            executorService.submit(new ConsumerWorker(MessageProcessor.messageQueue, i));
        }

        System.out.println(NUM_CONSUMERS + " consumer workers started.");

        // 模拟生产者不断生成消息
        for (int i = 0; i < 100; i++) {
            BusinessService.onMessage("Message-" + i);
            Thread.sleep(50); // 模拟消息产生间隔
        }

        // 模拟运行一段时间后，进行优雅关闭
        Thread.sleep(5000); // 让生产者继续产生一些消息

        // 优雅关闭：发送“毒丸”消息给每个消费者
        System.out.println("Initiating graceful shutdown...");
        for (int i = 0; i < NUM_CONSUMERS; i++) {
            MessageProcessor.messageQueue.put(MessageProcessor.POISON_PILL);
        }

        // 关闭ExecutorService
        shutdownAndAwaitTermination(executorService);
    }

    // 优雅关闭ExecutorService的方法
    private static void shutdownAndAwaitTermination(ExecutorService pool) {
        pool.shutdown(); // 启动有序关闭，不再接受新任务
        try {
            // 等待所有任务执行完毕，最多等待60秒
            if (!pool.awaitTermination(60, TimeUnit.SECONDS)) {
                pool.shutdownNow(); // 立即关闭，尝试停止所有正在执行的任务
                // 再次等待，以防万一
                if (!pool.awaitTermination(60, TimeUnit.SECONDS))
                    System.err.println("Pool did not terminate");
            }
        } catch (InterruptedException ie) {
            // (重新)取消
            pool.shutdownNow();
            // 保留中断状态
            Thread.currentThread().interrupt();
        }
    }
}