构建健壮的Java消息处理系统：使用BlockingQueue和线程池

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

// 全局消息队列
public class MessageQueueHolder {
    public static final BlockingQueue<String> globalMessageQueue = new LinkedBlockingQueue<>();
    // 用于优雅关闭的特殊消息，作为“毒丸”
    public static final String POISON_PILL = "POISON_PILL_SHUTDOWN";
}

public class MessageProducer {
    public void onMessage(String record) {
        try {
            // 将消息放入队列，如果队列满则阻塞等待
            MessageQueueHolder.globalMessageQueue.put(record);
            // System.out.println("Producer added: " + record);
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt(); // 重新设置中断标志
            System.err.println("Producer interrupted while adding message: " + e.getMessage());
        }
    }
}

import java.util.concurrent.BlockingQueue;

public 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 (MessageQueueHolder.POISON_PILL.equals(message)) {
                    System.out.println("Consumer " + consumerId + " received poison pill, shutting down.");
                    break; // 退出循环，线程结束
                }

                System.out.println("Consumer " + consumerId + " processing: " + message);
                // 模拟消息处理，例如：
                // Thread.sleep(100); 
                // ... 更多业务逻辑 ...
            }
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt(); // 重新设置中断标志
            System.err.println("Consumer " + consumerId + " interrupted: " + e.getMessage());
        } finally {
            System.out.println("Consumer " + consumerId + " finished.");
        }
    }
}

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

public class ApplicationMain {
    private static final int NUM_CONSUMERS = 8; // 消费者线程数量

    public static void main(String[] args) throws InterruptedException {
        // 创建一个固定大小的线程池来管理消费者
        ExecutorService consumerExecutor = Executors.newFixedThreadPool(NUM_CONSUMERS);

        // 启动消费者线程
        for (int i = 0; i < NUM_CONSUMERS; i++) {
            consumerExecutor.submit(new MessageConsumer(i, MessageQueueHolder.globalMessageQueue));
        }

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

        // 模拟一段时间后，准备关闭应用
        System.out.println("\n--- Application running for a while, preparing to shut down ---");
        Thread.sleep(2000); // 运行一段时间

        // 优雅关闭：向队列中放入“毒丸”消息，数量与消费者线程数相同
        for (int i = 0; i < NUM_CONSUMERS; i++) {
            MessageQueueHolder.globalMessageQueue.put(MessageQueueHolder.POISON_PILL);
        }

        // 关闭ExecutorService
        consumerExecutor.shutdown(); // 拒绝新任务，已提交任务会继续执行
        try {
            // 等待所有消费者线程完成任务，最长等待10秒
            if (!consumerExecutor.awaitTermination(10, TimeUnit.SECONDS)) {
                consumerExecutor.shutdownNow(); // 如果超时，则强制关闭
                System.err.println("Consumer threads did not terminate in time, forced shutdown.");
            }
        } catch (InterruptedException e) {
            consumerExecutor.shutdownNow();
            Thread.currentThread().interrupt();
            System.err.println("Main thread interrupted during shutdown: " + e.getMessage());
        }

        System.out.println("Application shut down gracefully.");
    }
}