I just wanted to confirm if this example I created would qualify as a good example of a multithreaded producer consumer. I would like any review changes on improving this example. I found that static fields are essential for synchronization of the threads. The wait-timeout mechanism along with the available flag functions as good as the wait/notifyAll mechanism. Synchronizing small blocks of code with private final static lock objects gives finer grained concurrency and prevents deadlocking.
package concurrency;
import java.util.ArrayList;
import java.util.List;
/*
* A Producer Consumer example scenario. There are MAX_THREAD number of producers and consumers.
* each thread produces and consumes a single element.
* Field transferCollection functions as the intermediatory collection between the producer and consumer.
* Boolean field available acts as the signaling mechanism between the producer and consumer.
* switched to true at the time when maximum number of threads have produced elements.
*
*/
public class ProduceConsume {
static int counter = 0;
static List<Integer> transferCollection = new ArrayList<Integer>();
static volatile boolean available = false;
static int producer_sum, CONSUMER_SUM, MAX_THREADS = 1000;
private static final Object addLock = new Object();
private static final Object getLock = new Object();
public static void main(String[] args) throws Exception {
List<Runnable> producers = new ArrayList<Runnable>();
Runnable producer = null;
for (int i = 0; i < MAX_THREADS; i++) {
producer = new Runnable() {
ProduceConsume pc = new ProduceConsume();
@Override
public void run() {
pc.produce();
}
};
producers.add(producer);
}
for (Runnable x : producers)
new Thread(x).start();
List<Runnable> consumers = new ArrayList<Runnable>();
Runnable consumer = null;
for (int i = 0; i < MAX_THREADS; i++) {
consumer = new Runnable() {
ProduceConsume pc = new ProduceConsume();
@Override
public void run() {
pc.consume();
}
};
consumers.add(consumer);
}
for (Runnable x : consumers)
new Thread(x).start();
}
/*
* Consumer threads invoke this method when MAX_THREADS of the producer are finished producing and adding to the list the same
* number of items.
* <b>getLock</b>
* A private final static lock object which threads wait on until the producer signals that elements are available on the list.
* <b>addLock</b>
* A private final static lock object which threads wait on to sum up elements in the transferCollection.
*/
private void consume() {
synchronized (getLock) {
while (!available) {
try {
getLock.wait(200);
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
}
}
}
if (counter > 0) {
int size0 = transferCollection.size() - 1;
synchronized (addLock) {
int idx = --counter;
CONSUMER_SUM += transferCollection.get(size0 - idx);
}
}
if (counter == 0) {
System.out.println("sum of consumer elements : " + CONSUMER_SUM);
available=false;
}
}
/*
* Producer threads invoke this method till number of threads has reached MAX_THREADS which have produced same
* number of items.
* <b>addLock</b>
* A private final static lock object that allows thread to sequentially add elements to the collection and
* produce a checksum at the end.
*/
private void produce() {
synchronized (addLock) {
transferCollection.add(++counter);
producer_sum += counter;
}
if (counter == MAX_THREADS) {
System.out.println("sum of producer elements : " + producer_sum);
available = true;
}
}
}
