并发设计模式实战系列(5)：生产者/消费者

// Java中的LinkedBlockingQueue核心实现
public void put(E e) throws InterruptedException {
    final ReentrantLock putLock = this.putLock; // 分离读写锁
    putLock.lockInterruptibly();
    try {
        while (count.get() == capacity) {
            notFull.await(); // 队列满时自动阻塞
        }
        enqueue(e);
        if (count.getAndIncrement() + 1 < capacity)
            notFull.signal(); // 仅唤醒生产者
    } finally {
        putLock.unlock();
    }
}

import java.util.concurrent.*;
public class ProducerConsumerDemo {
    // 有界阻塞队列（Array实现更严格的控制）
    private final BlockingQueue<String> buffer = new ArrayBlockingQueue<>(20);
    // 生产者配置
    class Producer implements Runnable {
        private final String[] menu = {"鱼香肉丝", "宫保鸡丁", "水煮鱼"};
        @Override
        public void run() {
            try {
                while (!Thread.currentThread().isInterrupted()) {
                    String dish = menu[ThreadLocalRandom.current().nextInt(menu.length)];
                    buffer.put(dish); // 自动阻塞直到有空位
                    System.out.println("[厨师] 制作完成：" + dish + " | 窗口存量：" + buffer.size());
                    Thread.sleep(500); // 模拟烹饪时间
                }
            } catch (InterruptedException e) {
                Thread.currentThread().interrupt();
            }
        }
    }
    // 消费者配置
    class Consumer implements Runnable {
        @Override
        public void run() {
            try {
                while (!Thread.currentThread().isInterrupted()) {
                    String dish = buffer.take(); // 自动阻塞直到有菜品
                    System.out.println("[服务员] 送出：" + dish + " | 剩余：" + buffer.size());
                    Thread.sleep(800); // 模拟送餐时间
                }
            } catch (InterruptedException e) {
                Thread.currentThread().interrupt();
            }
        }
    }
    // 启动系统
    public void startSystem() {
        ExecutorService producers = Executors.newFixedThreadPool(3);
        ExecutorService consumers = Executors.newCachedThreadPool();
        // 启动3个生产者（3个厨师）
        for (int i = 0; i < 3; i++) {
            producers.submit(new Producer());
        }
        // 启动动态消费者（根据队列负载）
        new Thread(() -> {
            while (true) {
                int idealConsumers = Math.max(1, buffer.size() / 5); // 每5盘菜1个服务员
                adjustConsumerCount(idealConsumers, consumers);
                try {
                    Thread.sleep(3000); // 每3秒动态调整
                } catch (InterruptedException e) {
                    break;
                }
            }
        }).start();
    }
    // 动态调整消费者数量
    private void adjustConsumerCount(int target, ExecutorService pool) {
        int current = ((ThreadPoolExecutor) pool).getActiveCount();
        if (target > current) {
            for (int i = current; i < target; i++) {
                pool.submit(new Consumer());
            }
        }
    }
    public static void main(String[] args) {
        new ProducerConsumerDemo().startSystem();
    }
}

// 生产者线程池：固定数量（对应物理灶台数量）
Executors.newFixedThreadPool(3); 
// 消费者线程池：动态调整（根据队列负载）
Executors.newCachedThreadPool();
// 队列选择：ArrayBlockingQueue vs LinkedBlockingQueue
// - ArrayBlockingQueue：固定容量，更严格的内存控制
// - LinkedBlockingQueue：节点链接，适合频繁插入删除

// 批量获取提高吞吐量
List<String> batch = new ArrayList<>(10);
int count = buffer.drainTo(batch, 10); // 一次性取10个
if (count > 0) {
    processBatch(batch);
}

// 使用优先级队列（需实现Comparator）
BlockingQueue<Order> queue = new PriorityBlockingQueue<>(20, 
    (o1, o2) -> o2.getVIPLevel() - o1.getVIPLevel());

// 设置超时时间的插入/获取
boolean success = buffer.offer(dish, 1, TimeUnit.SECONDS);
if (!success) {
    log.warn("菜品无法及时制作：{}", dish);
}

// 生产者配置
Properties producerProps = new Properties();
producerProps.put("bootstrap.servers", "kafka1:9092,kafka2:9092");
producerProps.put("key.serializer", StringSerializer.class.getName());
producerProps.put("value.serializer", StringSerializer.class.getName());
KafkaProducer<String, String> producer = new KafkaProducer<>(producerProps);
producer.send(new ProducerRecord<>("orders", "order-123", "VIP订单内容"));
// 消费者配置
Properties consumerProps = new Properties();
consumerProps.put("group.id", "order-processors");
consumerProps.put("bootstrap.servers", "kafka1:9092");
consumerProps.put("key.deserializer", StringDeserializer.class.getName());
consumerProps.put("value.deserializer", StringDeserializer.class.getName());
KafkaConsumer<String, String> consumer = new KafkaConsumer<>(consumerProps);
consumer.subscribe(Collections.singletonList("orders"));
while (true) {
    ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(100));
    for (ConsumerRecord<String, String> record : records) {
        processOrder(record.value());
    }
}

class RateLimiter {
    private final Semaphore tokens;
    private final ScheduledExecutorService refiller;
    RateLimiter(int permitsPerSecond) {
        this.tokens = new Semaphore(permitsPerSecond);
        this.refiller = Executors.newSingleThreadScheduledExecutor();
        refiller.scheduleAtFixedRate(() -> {
            int current = tokens.availablePermits();
            if (current < permitsPerSecond) {
                tokens.release(permitsPerSecond - current);
            }
        }, 0, 1, TimeUnit.SECONDS);
    }
    void acquire() throws InterruptedException {
        tokens.acquire();
    }
}
// 在生产线程中使用
rateLimiter.acquire();
buffer.put(item);

// 主工作队列
BlockingQueue<Message> mainQueue = new LinkedBlockingQueue<>();
// 死信队列
BlockingQueue<Message> dlq = new LinkedBlockingQueue<>();
class Consumer implements Runnable {
    @Override
    public void run() {
        while (!Thread.currentThread().isInterrupted()) {
            Message msg = mainQueue.take();
            try {
                process(msg);
            } catch (Exception e) {
                if (msg.getRetryCount() < 3) {
                    msg.incRetryCount();
                    mainQueue.put(msg);
                } else {
                    dlq.put(msg); // 进入死信队列
                }
            }
        }
    }
}
// 单独的死信处理线程
new Thread(() -> {
    while (true) {
        Message failedMsg = dlq.take();
        alertAdmin(failedMsg);
        archive(failedMsg);
    }
}).start();

# 启动JFR记录
java -XX:StartFlightRecording=duration=60s,filename=recording.jfr \
     -jar your-application.jar
# 分析队列状态
使用JDK Mission Visualizer查看：
- jdk.BlockingQueue$Take 事件
- jdk.BlockingQueue$Offer 事件
- 线程池的活跃线程数统计

// 三级处理流水线
BlockingQueue<RawData> inputQueue = new LinkedBlockingQueue<>();
BlockingQueue<ProcessedData> stage1Queue = new LinkedBlockingQueue<>();
BlockingQueue<FinalResult> stage2Queue = new LinkedBlockingQueue<>();
// 第一级：数据清洗
new Thread(() -> {
    while (true) {
        RawData data = inputQueue.take();
        ProcessedData cleaned = cleanData(data);
        stage1Queue.put(cleaned);
    }
}).start();
// 第二级：业务计算
new Thread(() -> {
    while (true) {
        ProcessedData data = stage1Queue.take();
        FinalResult result = calculate(data);
        stage2Queue.put(result);
    }
}).start();
// 第三级：结果存储
new Thread(() -> {
    while (true) {
        FinalResult result = stage2Queue.take();
        saveToDatabase(result);
    }
}).start();