Java实现生产者/消费者模式

描述：生产者一次生产N个产品，池中达到M就等待，通知等待的消费者。消费者有产品就消费，到没有时就通知生产者，然后等待。

实现效果：





















代码：


1.生产者<span style="font-size:14px;">package com.yealink.consume_product;

import java.util.List;

/**
* @author xioasheng
*
* 生产者.<br/>
* 默认产品池大小M=20,产品梯阶大小N=5.在生产过程中,池的大小会超过20,但池中最大应该是M+N-1.
*/
public class Producer implements Runnable {

/**
* 池默认大小
*/
public static final int DEFALUT_SIZE = 20;
/**
* 默认一次生产的数量
*/
public static final int DEFALUT_STEP_SIZE = 5;

private static int PRODUCER_ID = 0; //生产者号

private List<Product> pool = null;
private int size = DEFALUT_SIZE;
private int stepSize = DEFALUT_STEP_SIZE;

private String name = "Producer_"+(++PRODUCER_ID); //生产者名

private boolean isRun = true;

/**
* 默认产品池大小20, 默认产品增长梯阶大小5
*/
public Producer(List<Product> pool) {
this.pool = pool;
}

/**
* @param pool
* @param size 池大小
*/
public Producer(List<Product> pool, int size) {
this.pool = pool;
this.size = size;
}

/**
* @param pool
* @param size 池大小
* @param stepSize 一次生产多少
*/
public Producer(List<Product> pool, int size, int stepSize) {
this.pool = pool;
this.size = size;
this.stepSize = stepSize;
}

public void run() {
// TODO 生产者线程
//int pi = 0;
while(isRun) {//&& pi<10
//pi++;
synchronized (pool) { //同步产品池
if(pool.size() >= size) {
try {
System.out.println(name+" 等待!");
pool.wait(); //同步什么就等待什么,否则抛出java.lang.IllegalMonitorStateException
} catch (InterruptedException e) {
isRun = false;
System.out.println(name+" thread interrupt!");
}
} else {

for(int i=0; i<stepSize; i++) { //一次生产stepSize个产品
pool.add(product()); //生产产品
}
System.out.println("产品池中有: "+pool.size());
pool.notifyAll(); //通知等待的线程(主要用来通知消费者, 但生产者线程也会通知到)
}
}

try {
System.out.println(name+" 休息1秒!");
Thread.sleep(1000); //调试用
} catch (InterruptedException e) {
System.out.println(name+" sleep 1s thread interrupt");
}
}
System.out.println(name+" end! pool size: "+pool.size());
}

private static int P_ID = 0;
/**
* 生产产品
* @return 产品
*/
private Product product() {
String name = "product_"+(++P_ID);
System.out.println(this.name+" 生产了: "+name);
return new Product(name);
}

}</span>


2.消费者

package com.yealink.consume_product;

import java.util.List;

/**
* @author xiaosheng
*
* 消费者
*/
public class Consumer implements Runnable {

private static int C_ID = 0; //消费者ID

private List<Product> pool = null;
private String name = "Consumer_"+(++C_ID);
private boolean isRun = true;
public Consumer(List<Product> pool) {
this.pool = pool;
}

public void run() {
// TODO 消费者线程
//int pi = 0;
while(isRun) {//&& pi<10
//pi++;
synchronized (pool) {
if(pool.size() < 1) {
try {
System.out.println(name+" 等待!");
pool.notifyAll(); //通知线程(主要是消费者,但也会通知到生产者线程)
pool.wait();
} catch (InterruptedException e) {
isRun = false;
System.out.println(name+" thread interrupt!");
}
} else {
Product p = pool.remove(0); //消费
printProduct(p);

}
}
try {
Thread.sleep(1000); //调试用
} catch (InterruptedException e) {

System.out.println(name+" sleep 1s thread interrupt");
}
}
System.out.println(name+" end! pool size: "+pool.size());
}

private void printProduct(Product p) {
System.out.println(name+" 消费了: "+p.getName());
}
}
3.Product

package com.yealink.consume_product;

public class Product {
String name;

public Product(String name){
this.name=name;
}

public String getName() {
return name;
}

public void setName(String name) {
this.name = name;
}

}


4.测试Demo

package com.yealink.consume_product;

import java.util.LinkedList;
import java.util.List;

/**
* @author xiaosheng
*
*/
public class Sale {

public static void main(String[] args) {
//链表产品池
List<Product> pool = new LinkedList<Product>();
//两个生产者
Producer p1 = new Producer(pool);
Producer p2 = new Producer(pool);

Thread tp1 = new Thread(p1);
Thread tp2 = new Thread(p2);

tp1.start();
tp2.start();

//两个消费者
Consumer c1 = new Consumer(pool);
Consumer c2 = new Consumer(pool);

Thread tc1 = new Thread(c1);
Thread tc2 = new Thread(c2);

tc1.start();
tc2.start();

}

}



注意：等待时候要用pool.wait()因为同步的是pool。否则会抛出java.lang.IllegalMonitorStateException