生产者消费者问题是研究多线程程序时绕不开的问题它的描述是有一块生产者和消费者共享的有界缓沖区生产者往缓沖区放入产品消费者从缓沖区取走产品这个过程可以无休止的执行不能因缓沖区满生产者放不进产品而终止也不能因缓沖区空消费者无产品可取而终止
解决生产者消费者问题的方法有两种一种是采用某种机制保持生产者和消费者之间的同步一种是在生产者和消费者之间建立一个管道前一种有较高的效率并且可控制性较好比较常用后一种由于管道缓沖区不易控制及被传输数据对象不易封装等原因比较少用
同步问题的核心在于CPU是按时间片轮询的方式执行程序我们无法知道某一个线程是否被执行是否被抢占是否结束等因此生产者完全可能当缓沖区已满的时候还在放入产品消费者也完全可能当缓沖区为空时还在取出产品
现在同步问题的解决方法一般是采用信号或者加锁机制即生产者线程当缓沖区已满时放弃自己的执行权进入等待状态并通知消费者线程执行消费者线程当缓沖区已空时放弃自己的执行权进入等待状态并通知生产者线程执行这样一来就保持了线程的同步并避免了线程间互相等待而进入死锁状态
JAVA语言提供了独立于平台的线程机制保持了write once run anywhere的特色同时也提供了对同步机制的良好支持
在JAVA中一共有四种方法支持同步其中三个是同步方法一个是管道方法
方法wait()/notify()
方法await()/signal()
阻塞队列方法BlockingQueue
管道方法PipedInputStream/PipedOutputStream
下面我们看各个方法的实现
方法wait()/notify()
wait()和notify()是根类Object的两个方法也就意味着所有的JAVA类都会具有这个两个方法为什么会被这样设计呢？我们可以认为所有的对象默认都具有一个锁虽然我们看不到也没有办法直接操作但它是存在的
wait()方法表示当缓沖区已满或空时生产者或消费者线程停止自己的执行放弃锁使自己处于等待状态让另一个线程开始执行
notify()方法表示当生产者或消费者对缓沖区放入或取出一个产品时向另一个线程发出可执行通知同时放弃锁使自己处于等待状态
下面是一个例子代码
import javautilLinkedList;
public class Sycn{
private LinkedList<Object> myList =new LinkedList<Object>();
private int MAX = ;
public Sycn(){
}
public void start(){
new Producer()start();
new Consumer()start();
}
public static void main(String[] args) throws Exception{
Sycn s = new Sycn();
sstart();
}
class Producer extends Thread{
public void run(){
while(true){
synchronized(myList){
try{
while(myListsize() == MAX){
Systemoutprintln(warning: its full!);
myListwait();
}
Object o = new Object();
if(myListadd(o)){
Systemoutprintln(Producer: + o);
myListnotify();
}
}catch(InterruptedException ie){
Systemoutprintln(producer is interrupted!);
}
}
}
}
}
class Consumer extends Thread{
public void run(){
while(true){
synchronized(myList){
try{
while(myListsize() == ){
Systemoutprintln(warning: its empty!);
myListwait();
}
Object o = myListremoveLast();
Systemoutprintln(Consumer: + o);
myListnotify();
}catch(InterruptedException ie){
Systemoutprintln(consumer is interrupted!);
}
}
}
}
}
}
方法await()/signal()
在JDK以后JAVA提供了新的更加健壮的线程处理机制包括了同步锁定线程池等等它们可以实现更小粒度上的控制await()和signal()就是其中用来做同步的两种方法它们的功能基本上和wait()/notify()相同完全可以取代它们但是它们和新引入的锁定机制Lock直接挂钩具有更大的灵活性
下面是一个例子代码
import javautilLinkedList;
import ncurrentlocks*;
public class Sycn{
private LinkedList<Object> myList = new LinkedList<Object>();
private int MAX = ;
private final Lock lock = new ReentrantLock();
private final Condition full = locknewCondition();
private final Condition empty = locknewCondition();
public Sycn(){
}
public void start(){
new Producer()start();
new Consumer()start();
}
public static void main(String[] args) throws Exception{
Sycn s = new Sycn();
sstart();
}
class Producer extends Thread{
public void run(){
while(true){
locklock();
try{
while(myListsize() == MAX){
Systemoutprintln(warning: its full!);
fullawait();
}
Object o = new Object();
if(myListadd(o)){
Systemoutprintln(Producer: + o);
emptysignal();
}
}catch(InterruptedException ie){
Systemoutprintln(producer is interrupted!);
}finally{
lockunlock();
}
}
}
}
class Consumer extends Thread{
public void run(){
while(true){
locklock();
try{
while(myListsize() == ){
Systemoutprintln(warning: its empty!);
emptyawait();
}
Object o = myListremoveLast();
Systemoutprintln(Consumer: + o);
fullsignal();
}catch(InterruptedException ie){
Systemoutprintln(consumer is interrupted!);
}finally{
lockunlock();
}
}
}
}
}
阻塞队列方法BlockingQueue
BlockingQueue也是JDK的一部分它是一个已经在内部实现了同步的队列实现方式采用的是我们的第种await()/signal()方法它可以在生成对象时指定容量大小
它用于阻塞操作的是put()和take()方法
put()方法类似于我们上面的生产者线程容量最大时自动阻塞
take()方法类似于我们上面的消费者线程容量为时自动阻塞
下面是一个例子代码
import ncurrent*;
public class Sycn{
private LinkedBlockingQueue<Object> queue = new LinkedBlockingQueue<Object>();
private int MAX = ;
public Sycn(){
}
public void start(){
new Producer()start();
new Consumer()start();
}
public static void main(String[] args) throws Exception{
Sycn s = new Sycn();
sstart();
}
class Producer extends Thread{
public void run(){
while(true){
//synchronized(this){
try{
if(queuesize() == MAX)
Systemoutprintln(warning: its full!);
Object o = new Object();
queueput(o);
Systemoutprintln(Producer: + o);
}catch(InterruptedException e){
Systemoutprintln(producer is interrupted!);
}
//}
}
}
}
class Consumer extends Thread{
public void run(){
while(true){
//synchronized(this){
try{
if(queuesize() == )
Systemoutprintln(warning: its empty!);
Object o = queuetake();
Systemoutprintln(Consumer: + o);
}catch(InterruptedException e){
Systemoutprintln(producer is interrupted!);
}
//}
}
}
}
}
你发现这个例子中的问题了吗？
如果没有我建议你运行一下这段代码仔细观察它的输出是不是有下面这个样子的？为什么会这样呢？
…
warning: its full!
Producer: javalangobject@ea
…
你可能会说这是因为put()和Systemoutprintln()之间没有同步造成的我也这样认为我也这样认为但是你把run()中的synchronized前面的注释去掉重新编译运行有改观吗？没有为什么？
这是因为当缓沖区已满生产者在put()操作时put()内部调用了await()方法放弃了线程的执行然后消费者线程执行调用take()方法take()内部调用了signal()方法通知生产者线程可以执行致使在消费者的println()还没运行的情况下生产者的println()先被执行所以有了上面的输出run()中的synchronized其实并没有起什么作用
对于BlockingQueue大家可以放心使用这可不是它的问题只是在它和别的对象之间的同步有问题
对于这种多重嵌套同步的问题以后再谈吧欢迎大家讨论啊！
管道方法PipedInputStream/PipedOutputStream
这个类位于javaio包中是解决同步问题的最简单的办法一个线程将数据写入管道另一个线程从管道读取数据这样便构成了一种生产者/消费者的缓沖区编程模式
下面是一个例子代码在这个代码我没有使用Object对象而是简单的读写字节值这是因为PipedInputStream/PipedOutputStream不允许传输对象这是JAVA本身的一个bug具体的大家可以看sun的解释_bugdo?bug_id=
import javaio*;
public class Sycn{
private PipedOutputStream pos;
private PipedInputStream pis;
//private ObjectOutputStream oos;
//private ObjectInputStream ois;
public Sycn(){
try{
pos = new PipedOutputStream();
pis = new PipedInputStream(pos);
//oos = new ObjectOutputStream(pos);
//ois = new ObjectInputStream(pis);
}catch(IOException e){
Systemoutprintln(e);
}
}
public void start(){
new Producer()start();
new Consumer()start();
}
public static void main(String[] args) throws Exception{
Sycn s = new Sycn();
sstart();
}
class Producer extends Thread{
public void run() {
try{
while(true){
int b = (int) (Mathrandom() * );
Systemoutprintln(Producer: a byte the value is + b);
poswrite(b);
posflush();
//Object o = new MyObject();
//ooswriteObject(o);
//oosflush();
//Systemoutprintln(Producer: + o);
}
}catch(Exception e){
//Systemoutprintln(e);
eprintStackTrace();
}finally{
try{
posclose();
pisclose();
//oosclose();
//oisclose();
}catch(IOException e){
Systemoutprintln(e);
}
}
}
}
class Consumer extends Thread{
public void run(){
try{
while(true){
int b = pisread();
Systemoutprintln(Consumer: a byte the value is + StringvalueOf(b));
//Object o = oisreadObject();
//if(o != null)
//Systemoutprintln(Consumer: + o);
}
}catch(Exception e){
//Systemoutprintln(e);
eprintStackTrace();
}finally{
try{
posclose();
pisclose();
//oosclose();
//oisclose();
}catch(IOException e){
Systemoutprintln(e);
}
}
}
}
//class MyObject implements Serializable {
//}
}
