【机试题】队列并发消费问题

偏执的太偏执、 2024-02-19 11:35 29阅读 0赞

编写一个队列，生产者生产消息（消息内容是1,2,3,…,n），消费者消费消息，并将消息打印到控制台。  
要求：

1.  生产者以每秒10个的速率生产消息，队列满了后阻塞等待；
2.  队列长度为100；
3.  消费者以每秒1个的速率消费消息；  
    生产者和消费者在不同的线程；
4.  起20个消费者线程；
5.  消费者取不到消息的时候阻塞等待；
6.  不能重复消费；

import java.util.concurrent.*;
    
    class Message {
        
        private final int value;
    
        public Message(int value) {
        
            this.value = value;
        }
    
        public int getValue() {
        
            return value;
        }
    }
    
    class Producer implements Runnable {
        
        private final BlockingQueue<Message> queue;
    
        public Producer(BlockingQueue<Message> queue) {
        
            this.queue = queue;
        }
    
        @Override
        public void run() {
        
            try {
        
                for (int i = 1; i <= 1000; i++) {
        
                    Message message = new Message(i);
                    queue.put(message);
                    System.out.println("Produced: " + message.getValue());
                    Thread.sleep(100); // 生产者以每秒10个的速率生产消息
                    synchronized (this) {
        
                        while (queue.remainingCapacity() == 0) {
        
                            wait();
                        }
                    }
                }
            } catch (InterruptedException e) {
        
                e.printStackTrace();
            }
        }
    }
    
    class Consumer implements Runnable {
        
        private final BlockingQueue<Message> queue;
        private final Semaphore semaphore;
    
        public Consumer(BlockingQueue<Message> queue, Semaphore semaphore) {
        
            this.queue = queue;
            this.semaphore = semaphore;
        }
    
        @Override
        public void run() {
        
            try {
        
                while (true) {
        
                    Message message = queue.take(); // 取不到消息时阻塞等待
                    System.out.println("Consumed: " + message.getValue());
                    semaphore.acquire(); // 不能重复消费，使用信号量控制消费者数量
                    Thread.sleep(1000); // 消费者以每秒1个的速率消费消息
                }
            } catch (InterruptedException e) {
        
                e.printStackTrace();
            } finally {
        
                semaphore.release(); // 释放信号量，允许其他消费者继续消费
            }
        }
    }
    
    public class Main {
        
        public static void main(String[] args) {
        
            int numConsumers = 20; // 起20个消费者线程
            Semaphore semaphore = new Semaphore(numConsumers); // 控制消费者数量
            BlockingQueue<Message> queue = new LinkedBlockingQueue<>(100); // 队列长度为100
            Producer producer = new Producer(queue); // 生产者线程
            for (int i = 0; i < numConsumers; i++) {
         // 创建20个消费者线程并启动它们
                new Thread(new Consumer(queue, semaphore)).start(); // 消费者线程，取不到消息时阻塞等待，不能重复消费，每秒消费1个消息，最多同时有20个消费者线程在运行。
            } // 启动生产者线程，每秒生产10个消息，队列满后阻塞等待。生产者和消费者在不同的线程中运行。
        }
    }