多线程模拟实现生产者／消费者模型 (借鉴)

在生产者／消费者模型中，生产者Producer负责生产数据，而消费者Consumer负责使用数据。多个生产者线程会在同一时间运行，生产数据，并放到内存中一个共享的区域。期间，多个消费者线程读取内存共享区，消费里面的数据。

分析

在下面Java应用程序中，生产者线程向一个线程安全的堆栈缓冲区中写（PUSH）数据，消费者从该堆栈缓冲区中读（POP）数据，这样，这个程序中同时运行的两个线程共享同一个堆栈缓冲区资源。

类Producer是生产者模型，其中的run方法中定义了生产者线程所做的操作，循环调用push()方法,将生产的100个字母送入堆栈中，每次执行完push操作后，调用sleep方法睡眠一段随机时间。

类Consumer是消费者模型，循环调用pop方法，从堆栈取出一个字母，一共取100次，每次执行完push操作后，调用sleep方法睡眠一段随机时间



同步堆栈类SynchronizedStack

package com.ailk.biapp.ci.ProducerAndConsumer;

public class SynchronizedStack {
private int index = 0;
private int size = 100;
//共享内存区
private char[] data;

public SynchronizedStack(int size){
System.out.println("栈被创建");
this.size = size;
data = new char[size];
}

/**
* 生产数据
*
* @param c
*/
public synchronized void push(char c){
while (index == size){
try{
System.err.println("生产数据满了");
this.wait();//等待，直到有数据出栈
}catch(InterruptedException e){
Thread.currentThread().interrupt();
e.printStackTrace();
}
}
data[index] = c;
index++;
this.notify();//通知其他线程把数据出栈
}

/**
* 消费数据
*
* @return
*/
public synchronized char pop(){
while (index == 0){
try{
System.err.println("栈空了");
this.wait();// 等待，直到有数据出栈
}catch(InterruptedException e){
Thread.currentThread().interrupt();
e.printStackTrace();
}
}
index --;//指针向下移动
char ch = data[index];
this.notify();//通知其他线程把数据入栈
return ch;
}

//显示堆栈内容
public synchronized void print(){
for(int i = 0; i < data.length; i++){
System.out.println(data[i]);
}
System.out.println();
this.notify();// 通知其它线程显示堆栈内容
}

}

生产者Product

package com.ailk.biapp.ci.ProducerAndConsumer;

public class Producer implements Runnable{
private SynchronizedStack stack;

public Producer(SynchronizedStack s){
stack = s;
}

public void run(){
char ch;
for(int i = 0; i< 100; i++){
//随机产生100个字符
ch = (char) (Math.random() * 26 + 'A');
stack.push(ch);
System.out.println("Produced:" + ch);
try{
//每一个字符线程就休眠一下
Thread.sleep((int) (Math.random() * 1000));
}catch (InterruptedException e) {
}
}
}
}

消费者Consumer

package com.ailk.biapp.ci.ProducerAndConsumer;

public class Consumer implements Runnable{
private SynchronizedStack stack;

public Consumer(SynchronizedStack s){
stack = s;
}

public void run() {
char ch;
for(int i = 0 ; i < 100; i++){
ch = stack.pop();
System.out.println("Consumed:" + ch);
}
try{
Thread.sleep((int) (Math.random() * 1000));
}catch(InterruptedException e){

}

}

}

测试:

package com.ailk.biapp.ci.ProducerAndConsumer;

public class ProductConsumerTest {
public static void main(String args[]){
// 下面的消费者类对象和生产者类对象所操作的是同一个同步堆栈对象
SynchronizedStack stack = new SynchronizedStack(5);
Runnable source = new Producer(stack);
Runnable sink = new Consumer(stack);

Thread t1 = new Thread(source);
Thread t2 = new Thread(sink);
t1.start();
t2.start();
}
}

借鉴于:http://www.cnblogs.com/linjiqin/archive/2011/04/15/2016820.html