?生产者消费者 环形缓冲模式
?---------------------------------------------------------
/**
?* 环形缓冲区
?* @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();
}
?
}
?