Copy the code code as follows:
package condition;
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;
/**
* Use Lock and Condition to implement the producer-consumer model
* @author will
*
*/
public class ProducerConsumerDemo {
public static void main(String[] args) {
int producerCount = 10;
int consumerCount = 15;
final ProducerConsumerDemo pcd = new ProducerConsumerDemo(5); // Buffer size is 5
Thread[] producerThreads = new Thread[producerCount];
for(int i = 0; i < producerCount; i++) {
producerThreads[i] = new Thread("producer" + (i+1)) {
@Override
public void run() {
pcd.produce();
}
};
}
Thread[] consumerThreads = new Thread[consumerCount];
for(int j = 0; j < consumerCount; j++) {
consumerThreads[j] = new Thread("consumer" + (j+1)) {
@Override
public void run() {
pcd.consume();
}
};
}
//Start the producer consumer thread
for(int i = 0; i < producerCount; i++) {
producerThreads[i].start();
}
for(int j = 0; j < consumerCount; j++) {
consumerThreads[j].start();
}
}
private static final int DEFAULT_BUFFER_SIZE = 10;
private int bufferSize; // buffer size
private List<Object> bufferList;
private final Lock lock = new ReentrantLock(true);
private final Condition condition = lock.newCondition();
public ProducerConsumerDemo(int bufferSize) {
this.bufferSize = bufferSize > 0 ? bufferSize : DEFAULT_BUFFER_SIZE;
bufferList = new ArrayList<Object>(bufferSize);
}
// Production
public void produce() {
lock.lock(); // lock
try {
while(bufferList.size() == bufferSize) { // The buffer is full
System.out.println("Producer wait, thread: " + Thread.currentThread().getName());
condition.await();
}
// Production
bufferList.add(new Object());
System.out.println("Producer produce one, now buffer size: "
+ bufferList.size() + ", and thread: " + Thread.currentThread().getName());
condition.signalAll(); // Notify consumers
} catch(InterruptedException e) {
e.printStackTrace();
} finally {
lock.unlock();
}
}
// Consumption
public void consume() {
lock.lock(); // lock
try {
while(bufferList.isEmpty()) { // The buffer is empty
System.out.println("Consumer wait, thread: " + Thread.currentThread().getName());
condition.await();
}
// Consumption
bufferList.remove(0); // Remove one from the head of the linked list
System.out.println("Consumer consumer one, now buffer size: "
+ bufferList.size() + ", and thread: " + Thread.currentThread().getName());
condition.signalAll();
} catch(InterruptedException e) {
e.printStackTrace();
} finally {
lock.unlock();
}
}
}