How To create Consumer Producer Queue

Question

I have a Producer Which produces a POJO with a property, type. There can only be two types, "A" and "B". I have a thread pool for Consumer. Whenever I receive a message of type "B" from the Producer, Before I can proceed for execution, I need to make sure that all the other threads in the pool have completed execution(For now a default Thread.sleep). And then a consumer thread should pickup the message of type "B" and run it. Till this thread is running no message can be popped from the Queue.

Example:

class POJO_Message{

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

}

Answer 1

You can use ReadWriteLock to do you work.when the message type is 'B',try acquire write lock，other type message acquire read lock.one simple code like this.

public class ConsumerProducerQueue {
    ExecutorService executor = Executors.newFixedThreadPool(10);
    private ReadWriteLock lock = new ReentrantReadWriteLock();

    public void addMessage(Message message) {
        if ("B".equals(message.getType())) {
            lock.writeLock().lock();
            Future<?> result = executor.submit(new Task(message));
            try {
                result.get();
            } catch (Exception e) {

            } finally {
                lock.writeLock().unlock();
            }
        } else {
            lock.readLock().lock();
            Future<?> result = executor.submit(new Task(message));
            try {
                result.get();
            } catch (Exception e) {

            } finally {
                lock.readLock().unlock();
            }
        }
    }
}

the performance of this method is not good.

Answer 2

You can use LinkedBlockingDeque. An example:

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);
        }

    }

}