java - 如何创建消费者生产者队列

Question

我有一个 Producer，它生成一个具有属性、类型的 POJO。只能有两种类型:“A”和“B”。我有一个消费者线程池。每当我从生产者处收到类型“B”的消息时，在继续执行之前，我需要确保池中的所有其他线程都已完成执行(目前默认为 Thread.sleep)。然后消费者线程应该获取类型“B”的消息并运行它。在该线程运行之前，无法从队列中弹出任何消息。

示例:

class POJO_Message{

String type; //This will contain the type of message "A" or "B"

}

Answer 1

您可以使用LinkedBlockingDeque 。一个例子:

public class ProducerConsumer {

    public static void main(String[] args) {

        final LinkedBlockingDeque<Message> queue = new LinkedBlockingDeque<>(10);

        final AtomicLong id = new AtomicLong(0);
        final Timer producer = new Timer(true);
        producer.scheduleAtFixedRate(new TimerTask() {
            @Override
            public void run() {
               queue.add(new Message(  String.format("msg: %s"  , id.incrementAndGet() ) ) );
            }
        }, 10, 10);

        // consume
        for(;;) {
            try {
                Message msg  = queue.take();
                System.out.println( msg );
            } catch (InterruptedException e) {
                Thread.currentThread().interrupt();
            }
        }

    }

    private static class Message {
        private final String id;

        public Message(String id) {
            this.id = id;
        }

        public String getId() {
            return id;
        }

        @Override
        public String toString() {
            return String.format("Message [id=%s]", id);
        }

    }

}