Java并发编程：使用BlockingQueue构建健壮的生产者-消费者系统

// 假设这是高频触发的消息接收方法
public static void onMessage(String record) {
    globalQueue.add(record); // 将消息加入全局队列
    // 错误示例：手动管理线程启动
    if (!threadStatus[0]) { // 检查线程0是否“未启动”
        threadStatus[0] = true; // 标记为“已启动”
        new Thread(new MessageProcessor(0), "0").start(); // 启动线程
    }
    // ... 对其他线程索引重复类似逻辑
}

// 错误示例：线程的run方法
public void run() {
    int threadNum = Integer.parseInt(Thread.currentThread().getName());
    long startTime = System.currentTimeMillis();
    long endTime = startTime + 60 * 1000; // 线程运行1分钟后自动退出

    while (System.currentTimeMillis() < endTime) {
        if (!globalQueue.isEmpty()) {
            String recordToUse = globalQueue.remove();
            System.out.println("Successful removal: Thread-" + threadNum);
            // 执行更多业务操作...
        } else {
            // 队列为空时，继续检查
            continue;
        }
    }
    threadStatus[threadNum] = false; // 线程退出时，标记为“未启动”
    return;
}

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

public class MessageProcessorSystem {
    // 共享的阻塞队列，用于存储待处理的消息
    private static final BlockingQueue<String> messageQueue = new LinkedBlockingQueue<>();
    // 用于管理消费者线程的线程池
    private static ExecutorService consumerThreadPool;
    // 定义一个特殊的“毒丸”消息，用于通知消费者线程退出
    private static final String POISON_PILL = "POISON_PILL_SIGNAL";

    /**
     * 生产者方法：将消息添加到队列中
     * 这个方法会被高频触发
     *
     * @param record 待处理的消息
     */
    public static void onMessage(String record) {
        try {
            messageQueue.put(record); // 使用put()方法，队列满时会阻塞
            // System.out.println("Producer added: " + record);
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt(); // 重新设置中断标志
            System.err.println("Producer interrupted while adding message: " + e.getMessage());
        }
    }

    // ... 消费者线程类和主启动逻辑将在下面定义
}

class MessageConsumer implements Runnable {
    private final int consumerId;
    private final BlockingQueue<String> queue;

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

    @Override
    public void run() {
        System.out.println("Consumer " + consumerId + " started.");
        try {
            while (true) {
                String message = queue.take(); // 阻塞等待，直到队列中有消息
                if (MessageProcessorSystem.POISON_PILL.equals(message)) {
                    System.out.println("Consumer " + consumerId + " received poison pill, exiting.");
                    break; // 收到毒丸，退出循环
                }
                // 模拟消息处理
                System.out.println("Consumer " + consumerId + " processing: " + message);
                // 实际业务逻辑：对消息进行处理
                // Thread.sleep(50); // 模拟耗时操作
            }
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt(); // 重新设置中断标志
            System.err.println("Consumer " + consumerId + " interrupted: " + e.getMessage());
        } finally {
            System.out.println("Consumer " + consumerId + " finished.");
        }
    }
}

public class MessageProcessorSystem {
    // ... (messageQueue, consumerThreadPool, POISON_PILL 声明同上)

    public static void startConsumers(int numberOfConsumers) {
        consumerThreadPool = Executors.newFixedThreadPool(numberOfConsumers);
        for (int i = 0; i < numberOfConsumers; i++) {
            consumerThreadPool.submit(new MessageConsumer(i, messageQueue));
        }
        System.out.println(numberOfConsumers + " consumer threads started.");
    }

    /**
     * 优雅关闭消费者线程
     * 通过发送“毒丸”信号，通知所有消费者线程退出
     */
    public static void shutdownConsumers() {
        System.out.println("Initiating shutdown for consumers...");
        if (consumerThreadPool != null) {
            // 向队列中添加与消费者数量相同的毒丸，确保每个消费者都能收到
            for (int i = 0; i < 8; i++) { // 假设有8个消费者，或者根据实际启动的数量
                try {
                    messageQueue.put(POISON_PILL);
                } catch (InterruptedException e) {
                    Thread.currentThread().interrupt();
                    System.err.println("Error adding poison pill: " + e.getMessage());
                }
            }
            consumerThreadPool.shutdown(); // 停止接受新任务
            try {
                // 等待所有任务执行完毕，最多等待5秒
                if (!consumerThreadPool.awaitTermination(5, TimeUnit.SECONDS)) {
                    consumerThreadPool.shutdownNow(); // 立即关闭，中断正在执行的任务
                    System.err.println("Consumer threads did not terminate in time, forced shutdown.");
                }
            } catch (InterruptedException e) {
                consumerThreadPool.shutdownNow();
                Thread.currentThread().interrupt();
                System.err.println("Shutdown interrupted, forced shutdown.");
            }
        }
        System.out.println("Consumer threads shutdown complete.");
    }

    public static void main(String[] args) throws InterruptedException {
        int numConsumers = 8; // 启动8个消费者线程
        startConsumers(numConsumers);

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

        // 模拟运行一段时间
        Thread.sleep(2000); // 运行2秒后开始关闭

        // 关闭消费者线程
        shutdownConsumers();

        // 确保所有消息都被处理
        System.out.println("Remaining messages in queue: " + messageQueue.size());
        if (!messageQueue.isEmpty()) {
            System.out.println("Warning: Queue still contains messages after shutdown attempt.");
            // 在实际应用中，可能需要将剩余消息持久化或进行其他处理
        }
    }
}