Producer Consumer threads using a Queue

Question

I d like to create some sort of Producer Consumer threading app  But I m not sure what the best way to implement a queue between the two    So I ve some up with two ideas  both of which could be entirely wrong   I would like to know which would be better and if they both suck then what would be the best way to implement the queue  It s mainly my implementation of the queue in these examples that I m concerned about  I m extending a Queue class that is an in house class and is thread safe  Below are two examples with 4 classes each   Main class-  public class SomeApp       private Consumer consumer      private Producer producer       public static void main  String args                  consumer   new Consumer            producer   new Producer               Consumer class-  public class Consumer implements Runnable       public Consumer                 Thread consumer   new Thread this           consumer start               public void run                 while true                          get an object off the queue             Object object   QueueHandler dequeue                  do some stuff with the object                     Producer class-  public class Producer implements Runnable       public Producer                 Thread producer   new Thread this           producer start               public void run                 while true                          add to the queue some sort of unique object             QueueHandler enqueue new Object                         Queue class-  public class QueueHandler         This Queue class is a thread safe  written in house  class     public static Queue lt Object gt  readQ   new Queue lt Object gt  100        public static void enqueue Object object                  do some stuff         readQ add object              public static Object dequeue                   do some stuff         return readQ get              OR  Main class-  public class SomeApp       Queue lt Object gt  readQ      private Consumer consumer      private Producer producer       public static void main  String args                  readQ   new Queue lt Object gt  100           consumer   new Consumer readQ           producer   new Producer readQ              Consumer class-  public class Consumer implements Runnable       Queue lt Object gt  queue       public Consumer Queue lt Object gt  readQ                queue   readQ          Thread consumer   new Thread this           consumer start               public void run                 while true                          get an object off the queue             Object object   queue dequeue                  do some stuff with the object                     Producer class-  public class Producer implements Runnable       Queue lt Object gt  queue       public Producer Queue lt Object gt  readQ                queue   readQ          Thread producer   new Thread this           producer start               public void run                  while true                          add to the queue some sort of unique object             queue enqueue new Object                         Queue class-    the extended Queue class is a thread safe  written in house  class public class QueueHandler extends Queue lt Object gt            public QueueHandler int size                super size     All I m thinking about now is McDonalds             public void enqueue Object object                  do some stuff         readQ add               public Object dequeue                   do some stuff         return readQ get              And go

User · Accepted Answer

Java 5  has all the tools you need for this kind of thing  You will want to    Put all your Producers in one ExecutorService  Put all your Consumers in another ExecutorService  If necessary  communicate between the two using a BlockingQueue    I say  if necessary  for  3  because from my experience it s an unnecessary step  All you do is submit new tasks to the consumer executor service  So   final ExecutorService producers   Executors newFixedThreadPool 100   final ExecutorService consumers   Executors newFixedThreadPool 100   while     has more work         producers submit         producers shutdown    producers awaitTermination Long MAX VALUE  TimeUnit NANOSECONDS   consumers shutdown    consumers awaitTermination Long MAX VALUE  TimeUnit NANOSECONDS     So the producers submit directly to consumers

User · Answer

OK  as others note  the best thing to do is to use java util concurrent package  I highly recommend  Java Concurrency in Practice   It s a great book that covers almost everything you need to know   As for your particular implementation  as I noted in the comments  don t start Threads from Constructors -- it can be unsafe   Leaving that aside  the second implementation seem better  You don t want to put queues in static fields  You are probably just loosing flexibility for nothing   If you want to go ahead with your own implementation  for learning purpose I guess    supply a start   method at least  You should construct the object  you can instantiate the Thread object   and then call start   to start the thread     Edit  ExecutorService have their own queue so this can be confusing   Here s something to get you started     public class Main       public static void main String   args              The numbers are just silly tune parameters  Refer to the API            The important thing is  we are passing a bounded queue          ExecutorService consumer   new ThreadPoolExecutor 1 4 30 TimeUnit SECONDS new LinkedBlockingQueue lt Runnable gt  100               No need to bound the queue for this executor            Use utility method instead of the complicated Constructor          ExecutorService producer   Executors newSingleThreadExecutor             Runnable produce   new Produce consumer           producer submit produce               class Produce implements Runnable       private final ExecutorService consumer       public Produce ExecutorService consumer            this consumer   consumer              Override     public void run             Pancake cake   Pan cook            Runnable consume   new Consume cake           consumer submit consume            class Consume implements Runnable       private final Pancake cake       public Consume Pancake cake           this cake   cake              Override     public void run             cake eat              Further EDIT  For producer  instead of while true   you can do something like    Override public void run        while  Thread currentThread   isInterrupted               do stuff           This way you can shutdown the executor by calling  shutdownNow    If you d use while true   it won t shutdown     Also note that the Producer is still vulnerable to RuntimeExceptions  i e  one RuntimeException will halt the processing

User · Answer

Java code  BlockingQueue   which has synchronized put and get method   Java code  Producer    producer thread to produce data  Java code  Consumer    consumer thread to consume the data produced  Java code  ProducerConsumer Main   main function to start the producer and consumer thread    BlockingQueue java  public class BlockingQueue        int item      boolean available   false       public synchronized void put int value                 while  available    true                        try                                wait                  catch  InterruptedException e                                       item   value          available   true          notifyAll               public synchronized int get                 while available    false                        try                               wait                              catch InterruptedException e                                    available   false          notifyAll            return item            Consumer java  package com sukanya producer Consumer   public class Consumer extends Thread       blockingQueue queue      private int number      Consumer BlockingQueue queue int number                this queue   queue          this number   number             public void run                 int value   0           for  int i   0  i  lt  10  i                           value   queue get                System out println  Consumer      this number    got      value                       ProducerConsumer Main java  package com sukanya producer Consumer   public class ProducerConsumer Main        public static void main String args                  BlockingQueue queue   new BlockingQueue            Producer producer1   new Producer queue 1           Consumer consumer1   new Consumer queue 1           producer1 start            consumer1 start

User · Answer

This is a very simple code    import java util         author   rootTraveller  June 2017  class ProducerConsumer       public static void main String   args  throws Exception           Queue lt Integer gt  queue   new LinkedList lt  gt             Integer buffer   new Integer 10      Important buffer or queue size  change as per need           Producer producerThread   new Producer queue  buffer   PRODUCER            Consumer consumerThread   new Consumer queue  buffer   CONSUMER             producerThread start              consumerThread start                class Producer extends Thread       private Queue lt Integer gt  queue      private int queueSize        public Producer  Queue lt Integer gt  queueIn  int queueSizeIn  String ThreadName           super ThreadName           this queue   queueIn          this queueSize   queueSizeIn             public void run             while true               synchronized  queue                    while queue size      queueSize                       System out println Thread currentThread   getName       FULL           waiting    n                        try                          queue wait        Important                       catch  Exception ex                            ex printStackTrace                                                               queue empty then produce one  add and notify                   int randomInt   new Random   nextInt                     System out println Thread currentThread   getName       producing          randomInt                    queue add randomInt                    queue notifyAll       Important                 synchronized ends here   NOTE                    class Consumer extends Thread       private Queue lt Integer gt  queue      private int queueSize       public Consumer Queue lt Integer gt  queueIn  int queueSizeIn  String ThreadName           super  ThreadName           this queue   queueIn          this queueSize   queueSizeIn             public void run             while true               synchronized  queue                    while queue isEmpty                         System out println Thread currentThread   getName       Empty          waiting    n                        try                           queue wait       Important                       catch  Exception ex                            ex printStackTrace                                                               queue not empty then consume one and notify                 System out println Thread currentThread   getName       consuming          queue remove                     queue notifyAll                    synchronized ends here   NOTE

User · Answer

You are reinventing the wheel   If you need persistence and other enterprise features use JMS  I d suggest ActiveMq    If you need fast in-memory queues use one of the impementations of java s Queue   If you need to support java 1 4 or earlier  use Doug Lea s excellent concurrent package

User · Answer

I have extended cletus proposed answer to working code example     One ExecutorService  pes  accepts Producer tasks  One ExecutorService  ces  accepts Consumer tasks  Both Producer and Consumer shares BlockingQueue  Multiple Producer tasks generates different numbers  Any of Consumer tasks can consume number generated by Producer   Code   import java util concurrent     public class ProducerConsumerWithES       public static void main String args              BlockingQueue lt Integer gt  sharedQueue   new LinkedBlockingQueue lt Integer gt               ExecutorService pes   Executors newFixedThreadPool 2            ExecutorService ces   Executors newFixedThreadPool 2             pes submit new Producer sharedQueue 1             pes submit new Producer sharedQueue 2             ces submit new Consumer sharedQueue 1             ces submit new Consumer sharedQueue 2                shutdown should happen somewhere along with awaitTermination              https   stackoverflow com questions 36644043 how-to-properly-shutdown-java-executorservice 36644320 36644320             pes shutdown             ces shutdown            class Producer implements Runnable       private final BlockingQueue lt Integer gt  sharedQueue      private int threadNo      public Producer BlockingQueue lt Integer gt  sharedQueue int threadNo            this threadNo   threadNo          this sharedQueue   sharedQueue             Override     public void run             for int i 1  i lt   5  i                 try                   int number   i  10 threadNo                   System out println  Produced     number     by thread    threadNo                   sharedQueue put number                 catch  Exception err                    err printStackTrace                                     class Consumer implements Runnable      private final BlockingQueue lt Integer gt  sharedQueue      private int threadNo      public Consumer  BlockingQueue lt Integer gt  sharedQueue int threadNo            this sharedQueue   sharedQueue          this threadNo   threadNo             Override     public void run             while true               try                   int num   sharedQueue take                    System out println  Consumed     num     by thread   threadNo                 catch  Exception err                   err printStackTrace                                         output   Produced 11 by thread 1 Produced 21 by thread 2 Produced 22 by thread 2 Consumed  11 by thread 1 Produced 12 by thread 1 Consumed  22 by thread 1 Consumed  21 by thread 2 Produced 23 by thread 2 Consumed  12 by thread 1 Produced 13 by thread 1 Consumed  23 by thread 2 Produced 24 by thread 2 Consumed  13 by thread 1 Produced 14 by thread 1 Consumed  24 by thread 2 Produced 25 by thread 2 Consumed  14 by thread 1 Produced 15 by thread 1 Consumed  25 by thread 2 Consumed  15 by thread 1   Note  If you don t need multiple Producers and Consumers  keep single Producer and Consumer  I have added multiple Producers and Consumers to showcase capabilities of BlockingQueue among multiple Producers and Consumers

User · Answer

Use this typesafe pattern with poison pills  public sealed interface BaseMessage        final class ValidMessage lt T gt  implements BaseMessage             Nonnull         private final T value            public ValidMessage  Nonnull T value                this value   value                      Nonnull         public T getValue                 return value                      Override         public boolean equals Object o                if  this    o  return true              if  o    null    getClass      o getClass    return false              ValidMessage lt   gt  that    ValidMessage lt   gt   o              return value equals that value                       Override         public int hashCode                 return Objects hash value                       Override         public String toString                 return  quot ValidMessage value  s  quot  formatted value                        final class PoisonedMessage implements BaseMessage            public static final PoisonedMessage INSTANCE   new PoisonedMessage              private PoisonedMessage                         Override         public String toString                 return  quot PoisonedMessage   quot                      public class Producer implements Callable lt Void gt          Nonnull     private final BlockingQueue lt BaseMessage gt  messages       Producer  Nonnull BlockingQueue lt BaseMessage gt  messages            this messages   messages              Override     public Void call   throws Exception           messages put new BaseMessage ValidMessage lt  gt  1            messages put new BaseMessage ValidMessage lt  gt  2            messages put new BaseMessage ValidMessage lt  gt  3            messages put BaseMessage PoisonedMessage INSTANCE           return null           public class Consumer implements Callable lt Void gt          Nonnull     private final BlockingQueue lt BaseMessage gt  messages       private final int maxPoisons        public Consumer  Nonnull BlockingQueue lt BaseMessage gt  messages  int maxPoisons            this messages   messages          this maxPoisons   maxPoisons              Override     public Void call   throws Exception           int poisonsReceived   0          while  poisonsReceived  lt  maxPoisons  amp  amp   Thread currentThread   isInterrupted                  BaseMessage message   messages take                if  message instanceof BaseMessage ValidMessage lt   gt  vm                    Integer value    Integer  vm getValue                    System out println value                 else if  message instanceof BaseMessage PoisonedMessage                      poisonsReceived                else                   throw new IllegalArgumentException  quot Invalid BaseMessage type   quot    message                                   return null

[java] Producer/Consumer threads using a Queue

Examples related to java

Examples related to multithreading

Examples related to queue

Examples related to producer-consumer