Producer/Consumer threads using a Queue
Solution 1
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
.
Solution 2
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<Runnable>(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)
Solution 3
I have extended cletus proposed answer to working code example.
- One
ExecutorService
(pes) acceptsProducer
tasks. - One
ExecutorService
(ces) acceptsConsumer
tasks. - Both
Producer
andConsumer
sharesBlockingQueue
. - Multiple
Producer
tasks generates different numbers. - Any of
Consumer
tasks can consume number generated byProducer
Code:
import java.util.concurrent.*;
public class ProducerConsumerWithES {
public static void main(String args[]){
BlockingQueue<Integer> sharedQueue = new LinkedBlockingQueue<Integer>();
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<Integer> sharedQueue;
private int threadNo;
public Producer(BlockingQueue<Integer> sharedQueue,int threadNo) {
this.threadNo = threadNo;
this.sharedQueue = sharedQueue;
}
@Override
public void run() {
for(int i=1; i<= 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<Integer> sharedQueue;
private int threadNo;
public Consumer (BlockingQueue<Integer> 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.
Solution 4
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.
Solution 5
This is a very simple code.
import java.util.*;
// @author : rootTraveller, June 2017
class ProducerConsumer {
public static void main(String[] args) throws Exception {
Queue<Integer> queue = new LinkedList<>();
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<Integer> queue;
private int queueSize ;
public Producer (Queue<Integer> 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<Integer> queue;
private int queueSize;
public Consumer(Queue<Integer> 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
}
}
}
Comments
-
Akshay almost 2 years
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<Object> readQ = new Queue<Object>(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<Object> readQ; private Consumer consumer; private Producer producer; public static void main (String args[]) { readQ = new Queue<Object>(100); consumer = new Consumer(readQ); producer = new Producer(readQ); } }
Consumer class-
public class Consumer implements Runnable { Queue<Object> queue; public Consumer(Queue<Object> 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<Object> queue; public Producer(Queue<Object> 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<Object> { 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!