Threads 101

Listing 14: Uses a counter to manage producer and consumer threads

import java.util.*;

class Counter
{
   private int count;

   public Counter()
   {
      count = 0;
   }

   public void increment()
   {
      ++count;
   }

   public void decrement()
   {
      --count;
   }

   public int get()
   {
      return count;
   }
}

class Queue
{
   // The shared buffer:
   static Queue store = new Queue();

   private LinkedList data = new LinkedList();
    
   public void put(Object o)
   {
      data.addFirst(o);
   }
   public Object get()
   {
      return data.removeLast();
   }
   public int size()
   {
      return data.size();
   }
}

class Producer extends Thread
{
   private static Random numbers = new Random();
   private Counter counter;

   Producer(String id, Counter counter)
   {
      super(id);
      this.counter = counter;
   }

   int genNumber()
   {
      // Generate a random number in (0, 1000)
      return (int)(numbers.nextDouble()*1000);
   }

   public void run()
   {
      counter.increment();
      
      // Generate some elements for the Queue
      for (int i = 0; i < 8; ++i)
      {
         int number = genNumber();
         System.out.println(getName() + " producing " + number);
         synchronized(Queue.store)
         {
            Queue.store.put(new Integer(number));
            Queue.store.notify();
         }
      }
      synchronized(Queue.store)
      {
         // Prevent deadlock for multiple consumers
         counter.decrement();
         Queue.store.notifyAll();
      }
      System.out.println("\t" + getName() + " finished");
   }
}

class Consumer extends Thread
{
   private Counter counter;

   Consumer(String id, Counter counter)
   {
      super(id);
      this.counter = counter;
   }

   public void run()
   {
      for (;;)
      {
         synchronized(Queue.store)
         {
            while (Queue.store.size() == 0)
            {
               if (counter.get() == 0)
               {
                  // Producers are done and queue is empty
                  System.out.println("\t" + getName()
                                     + " terminating");
                  return;
               }
               try
               {
                  Queue.store.wait();
               }
               catch (InterruptedException x)
               {}
            }
            System.out.println(getName() + " consuming "
                              + Queue.store.get());
         }
      }
   }
}

class CommTest2
{
   public static void main(String[] args)
   {
      // Start Producers
      Counter counter = new Counter();
      new Producer("Producer1", counter).start();
      new Producer("Producer2", counter).start();

      // Start Consumers
      new Consumer("Consumer1", counter).start();
      new Consumer("Consumer2", counter).start();
   }
}

/* Output:
Producer1 producing 386
Producer1 producing 240
Producer1 producing 982
Producer1 producing 453
Producer1 producing 878
Producer1 producing 228
Producer1 producing 245
Producer2 producing 189
Consumer1 consuming 386
Consumer2 consuming 240
Producer2 producing 740
Consumer1 consuming 982
Consumer2 consuming 453
Producer2 producing 264
Consumer1 consuming 878
Consumer2 consuming 228
Producer2 producing 686
Consumer1 consuming 189
Consumer2 consuming 740
Producer2 producing 761
Consumer1 consuming 264
Consumer2 consuming 686
Producer2 producing 586
Consumer1 consuming 761
Producer2 producing 847
Consumer1 consuming 586
Producer2 producing 161
Consumer1 consuming 847
Consumer1 consuming 161
        Producer2 finished
Consumer1 consuming 245
Producer1 producing 329
Consumer2 consuming 329
        Consumer1 terminating
        Consumer2 terminating
        Producer1 finished
*/
— End of Listing —