生产者消费者模式 生产者消费者模式
生产者消费者 环形缓冲模式
---------------------------------------------------------
/**
* 环形缓冲区
* @author fengbin
*
*/
public class CircularBuf {
int NMAX = 1000;
int iput = 0; // 环形缓冲区的当前放人位置
int iget = 0; // 缓冲区的当前取出位置
int n = 0; // 环形缓冲区中的元素总数量
Object buffer[];
public CircularBuf() {
super();
buffer = new Object[NMAX];
}
public CircularBuf(int nmax) {
super();
NMAX = nmax;
buffer = new Object[NMAX];
}
/*
* 环形缓冲区的地址编号计算函数，，如果到达唤醒缓冲区的尾部，将绕回到头部。
*
* 环形缓冲区的有效地址编号为：0到(NMAX-1)
*
*/
public synchronized int addring(int i) {
return (i + 1) == NMAX ? 0 : i + 1;
}
/* 从环形缓冲区中取一个元素 */
public synchronized Object get() {
int pos;
System.out.println("------------数量:"+n);
if (n > 0) {
pos = iget;
iget = addring(iget);
n--;
// System.out.println("get-->" + buffer[pos]);
notifyAll();
return buffer[pos];
} else {
// System.out.println("Buffer is Empty");
try {
wait();
} catch (InterruptedException e) {
// TODO Auto-generated catch block
e.printStackTrace();
}
}
return null;
}
/* 向环形缓冲区中放人一个元素 */
public synchronized void put(Object z) {
if (n < NMAX) {
buffer[iput] = z;
System.out.println("put<--" + buffer[iput]);
iput = addring(iput);
n++;
notifyAll();
} else {
System.out.println("Buffer is full");
try {
wait();
put(z);//如果满了后,重新执行
System.out.println("rerun !");
} catch (InterruptedException e) {
// TODO Auto-generated catch block
e.printStackTrace();
}
}
}
}
-----------------------------------------------------------------
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import org.apache.commons.lang.StringUtils;
/**
* 环形缓冲区 测试
* @author fengbin
*
*/
public class CircularBufTest {
CircularBuf circularBuf = new CircularBuf(2);
class Producer implements Runnable{
String apple = "";
Producer(String str){
apple = str;
}
@Override
public void run() {
for(int j=0;j<5;j++){
circularBuf.put(apple+j);
System.out.println("!生产:"+apple+j+"结束");
}
}
}
class Consumer implements Runnable{
Consumer(){
}
@Override
public void run() {
while(true){
String str=(String) circularBuf.get();
if(StringUtils.isEmpty(str)){
continue;
}
System.out.println("#消费:"+str+"结束");
try {
Thread.sleep(500);
} catch (Exception e) {
System.out.println(e);
}
}
}
}
public void test() {
ExecutorService service = Executors.newCachedThreadPool();
Producer producer = new Producer("P");
Consumer consumer = new Consumer();
service.submit(producer);
service.submit(consumer);
}
public static void main(String[] args) {
CircularBufTest test = new CircularBufTest();
test.test();
}
}
=================================================
使用阻塞队列实现的生产者和消费者模式.
public class TestProducterConsumer {
class Producter extends Thread {
Queue q;
Producter(Queue q) {
this.q = q;
}
public void run() {
for (int i = 0; i < 10; i++) {
q.put(i);
System.out.println("producter :" + i);
}
}
}
class Consumer extends Thread{
Queue q;
Consumer(Queue q) {
this.q = q;
}
public void run() {
while (true) {
System.out.println("Consumer:" + q.get());
}
}
}
class Queue {
int value;
boolean bFull = false;
public synchronized void put(int i) {
if (!bFull) {
value = i;
bFull = true;
notify();// 必须用在synchronized
}
try {
wait();// 必须捕获异常
} catch (InterruptedException e) {
// TODO Auto-generated catch block
e.printStackTrace();
}
}
public synchronized int get() {
if (!bFull)
try {
wait();
} catch (InterruptedException e) {
// TODO Auto-generated catch block
e.printStackTrace();
}
bFull = false;
notify();
return value;
}
}
public static void main(String[] args) {
TestProducterConsumer con = new TestProducterConsumer();
con.test();
}
private void test() {
Queue q = new Queue();
Producter p = new Producter(q);
Consumer c = new Consumer(q);
p.start();
c.start();
}
}