Java多线程 Producer and Consumer设计模式

public class ProducerThread extends Thread {

     private final static Random random = new Random(System.currentTimeMillis());

     private final static AtomicInteger counter = new AtomicInteger( 0 );

     private final MessageQueue messageQueue;

     public ProducerThread(MessageQueue messageQueue, int seq) {

         super ( "Producer-" + seq);

         this .messageQueue = messageQueue;

     }

     @Override

     public void run() {

         while ( true ) {

             try {

                 Message message = new Message( "Message-" + counter.getAndIncrement());

                 messageQueue.put(message);

                 System.out.println(Thread.currentThread().getName() + " put message " + message.getData());

                 Thread.sleep(random.nextInt( 1000 ));

             } catch (InterruptedException e) {

                 break ;

             }

         }

     }

}

public class ConsumerThread extends Thread {

     private final static Random random = new Random(System.currentTimeMillis());

     private final MessageQueue messageQueue;

     public ConsumerThread(MessageQueue messageQueue, int seq) {

         super ( "Consumer-" + seq);

         this .messageQueue = messageQueue;

     }

     @Override

     public void run() {

         while ( true ) {

             try {

                 Message message = messageQueue.take();

                 System.out.println(Thread.currentThread().getName() + " take a message " + message.getData());

                 Thread.sleep(random.nextInt( 1000 ));

             } catch (InterruptedException e) {

                 break ;

             }

         }

     }

}

public class Message {

     public Message(String data) {

         this .data = data;

     }

     private String data;

     public String getData() {

         return data;

     }

     public void setData(String data) {

         this .data = data;

     }

}

public class MessageQueue {

     private final static int DEFAULT_MAX_LIMIT = 100 ;

     private final LinkedList<Message> queue;

     private final int limit;

     public MessageQueue() {

         this (DEFAULT_MAX_LIMIT);

     }

     public MessageQueue( final int limit) {

         this .limit = limit;

         this .queue = new LinkedList<>();

     }

     public void put( final Message message) throws InterruptedException {

         synchronized (queue) {

             while (queue.size() > limit) {

                 queue.wait();

             }

             queue.addLast(message);

             queue.notifyAll();

         }

     }

     public Message take() throws InterruptedException {

         synchronized (queue) {

             while (queue.isEmpty()) {

                 queue.wait();

             }

             Message message = queue.removeFirst();

             queue.notifyAll();

             return message;

         }

     }

     public int getMaxLimit() {

         return this .limit;

     }

     public int getMessageSize() {

         synchronized (queue) {

             return queue.size();

         }

     }

}

public class ProducerAndConsumerClient {

     public static void main(String[] args) {

         final MessageQueue messageQueue = new MessageQueue();

         new ProducerThread(messageQueue, 1 ).start();

         new ProducerThread(messageQueue, 2 ).start();

         new ProducerThread(messageQueue, 3 ).start();

         new ConsumerThread(messageQueue, 1 ).start();

         new ConsumerThread(messageQueue, 2 ).start();

     }

}