Java实现生产者——消费者模型

对多线程同步问题一直了解不是很深刻，找了生产者消费者模型来试验了一下。不知道正确与否，初步实验了一下好像是正确的，如有错误请指正。

生产者——消费者模型是这样的：

有M个生产者不断的想生产线上放置产品。而有N个消费者从生产线上将产品拿出来。实现这一过程。

代码如下所示：

package concurrency;

import java.util.Random;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

public class ProducerAndConsumer {
private final static int N=10;
private Object products;
private Object empties;                 //empties和products分别用于生产者消费者线程间的同步
private Object mutex;                   //用于对共享变量in/out之间的互斥访问
int slots = N;                          //记住流水线上的空槽数
int in;                                 //记住上一次生产者将产品放置的位置
int out;                                //记住消费者上次取出商品的位置
private String[] buffer = new String
;

public ProducerAndConsumer(){
products = new Object();
empties = new Object();
mutex = new Object();
in = 0;
out = 0;
}

class Producer implements Runnable{
int id;
Random rand;
public Producer(int id,int seed){
this.id = id;
rand = new Random(seed);
}
@Override
public void run() {
try {
while(true){
synchronized (empties) {
if(slots<=0)
empties.wait();
}
synchronized(mutex){
if(slots<=0)
continue;
System.out.println("Producer " + id + " produce item!");
buffer[in] = "Product " + in;
in = (in+1)%N;
slots--;
System.out.println("Left " + slots + " slots!");
}
synchronized (products) {
products.notifyAll();
}
Thread.sleep(rand.nextInt(1000)+500);
}
} catch (InterruptedException e) {
// TODO Auto-generated catch block
e.printStackTrace();
}

}
}
class Consumer implements Runnable{
int id;
Random rand;
public Consumer(int id,int seed){
this.id = id;
rand = new Random(seed);
}
@Override
public void run() {
try {
while(true){
synchronized (products) {
if(slots>=N)
products.wait();
}
synchronized (mutex) {
if(slots>=N)
continue;
System.out.println("Consumer " + id +
" take the product from " + out + " slot!");
buffer[out] = null;
out = (out+1)%N;

slots++;

System.out.println("Left " + slots +" slots!");
}
synchronized (empties) {
empties.notifyAll();
}
//empties.notifyAll();
Thread.sleep(rand.nextInt(1000)+1000);
}
} catch (InterruptedException e) {
e.printStackTrace();
}
}
}

public static void main(String[] args){
ProducerAndConsumer pac = new ProducerAndConsumer();
ExecutorService es = Executors.newFixedThreadPool(25);
Random rand = new Random(47);

for(int i = 0 ; i < 10; i++)
es.execute(pac.new Producer(i, rand.nextInt(100)));

for(int i = 0 ; i < 15; i++)
es.execute(pac.new Consumer(i, rand.nextInt(100)));
}
}

由于生产者数目较少，而消费者数目较多，因此设定生产产品的时间比消费产品的时间要短。结果如下所示

Producer 0 produce item!
Left 9 slots!
Producer 2 produce item!
Left 8 slots!
Producer 4 produce item!
Left 7 slots!
Producer 6 produce item!
Left 6 slots!
Producer 8 produce item!
Left 5 slots!
Producer 1 produce item!
Left 4 slots!
Producer 3 produce item!
Left 3 slots!
Producer 5 produce item!
Left 2 slots!
Producer 7 produce item!
Left 1 slots!
Consumer 2 take the product from 0 slot!
Left 2 slots!
Consumer 0 take the product from 1 slot!
Left 3 slots!
Consumer 4 take the product from 2 slot!
Left 4 slots!
Consumer 6 take the product from 3 slot!
Left 5 slots!
Consumer 8 take the product from 4 slot!
Left 6 slots!
Producer 9 produce item!
Left 5 slots!
Consumer 10 take the product from 5 slot!
Left 6 slots!
Consumer 12 take the product from 6 slot!
Left 7 slots!
Consumer 14 take the product from 7 slot!
Left 8 slots!
Consumer 7 take the product from 8 slot!
Left 9 slots!
Consumer 5 take the product from 9 slot!
Left 10 slots!
Producer 0 produce item!
Left 9 slots!
Consumer 9 take the product from 0 slot!
Left 10 slots!
Producer 9 produce item!
Left 9 slots!
Consumer 3 take the product from 1 slot!
Left 10 slots!
Producer 1 produce item!
Left 9 slots!
Consumer 1 take the product from 2 slot!
Left 10 slots!
Producer 7 produce item!
Left 9 slots!
Consumer 11 take the product from 3 slot!
Left 10 slots!
Producer 5 produce item!
Left 9 slots!
Consumer 13 take the product from 4 slot!
Left 10 slots!