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;

/**
 * 利用Lock、Condition實(shí)現(xiàn)生產(chǎn)者消費(fèi)者模式
 * @author will
 *
 */
public class ProducerConsumerDemo {

 public static void main(String[] args) {
  int producerCount = 10;
  int consumerCount = 15;

  final ProducerConsumerDemo pcd = new ProducerConsumerDemo(5);   // 緩沖區(qū)大小為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();
    }
   };
  }

  // 啟動(dòng)生產(chǎn)者消費(fèi)者線(xiàn)程
  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;   // 緩沖區(qū)大小
 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);
 }

 // 生產(chǎn)
 public void produce() {
  lock.lock();   // 加鎖
  try {
   while(bufferList.size() == bufferSize) {   // 緩沖區(qū)滿(mǎn)了
    System.out.println("Producer wait, thread: " + Thread.currentThread().getName());
    condition.await();
   }

   // 生產(chǎn)
   bufferList.add(new Object());
   System.out.println("Producer produce one, now buffer size: "
     + bufferList.size() + ", and thread: " + Thread.currentThread().getName());
   condition.signalAll();   // 通知消費(fèi)者
  } catch(InterruptedException e) {
   e.printStackTrace();
  } finally {
   lock.unlock();
  }
 }

 // 消費(fèi)
 public void consume() {
  lock.lock();   // 加鎖
  try {
   while(bufferList.isEmpty()) {   // 緩沖區(qū)空了
    System.out.println("Consumer wait, thread: " + Thread.currentThread().getName());
    condition.await();
   }

   // 消費(fèi)
   bufferList.remove(0);   // 從鏈表頭部移除一個(gè)
   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();
  }
 }

}