JAVA线程中的生产者和消费者问题

谁借莪１个温暖的怀抱￠ 2021-09-25 15:48 219阅读 0赞

生产者消费者问题是研究多线程时绕不开的问题，描述的是有一块生产者和消费者共享的有界缓冲区，生产者往缓冲区放入产品，消费者从缓冲区取走产品，这个过程可以无休止的执行，不能因缓冲区满生产者放不进产品而终止，也不能因缓冲区空消费者无产品可取而终止。  
  
解决生产者消费者问题的方法有两种，一种是采用某种机制保持生产者和消费者之间的同步，一种是在生产者和消费者之间建立一个管道。前一种有较高的效率并且可控制性较好，比较常用，后一种由于管道缓冲区不易控制及被传输数据对象不易封装等原因，比较少用。  
  
同步问题的核心在于，CPU是按时间片轮询的方式执行程序，我们无法知道某一个线程是否被执行、是否被抢占、是否结束等，因此生产者完全可能当缓冲区已满的时候还在放入产品，消费者也完全可能当缓冲区为空时还在取出产品。**现在同步问题的解决方法一般是采用信号或者加锁机制**，即生产者线程当缓冲区已满时放弃自己的执行权，进入等待状态，并通知消费者线程执行。消费者线程当缓冲区已空时放弃自己的执行权，进入等待状态，并通知生产者线程执行。这样一来就保持了线程的同步，并避免了线程间互相等待而进入死锁状态。  
         
JAVA语言提供了独立于平台的线程机制，保持了write once, run anywhere的特色。同时也提供了对同步机制的良好支持。  
         
JAVA中一共有四种方法支持同步，其中三个是同步方法，一个是管道方法。  
①方法wait()/notify()  
②方法await()/signal()  
③阻塞队列方法BlockingQueue  
④管道方法PipedInputStream/PipedOutputStream  
  
  
**一、方法wait()/notify()**  
wait()和notify()是根类Object的两个方法，也就意味着所有的JAVA类都具有这个两个方法，可以认为所有的对象默认都具有一个锁，虽然看不到也无法直接操作，但它是存在的。  
wait()方法表示：当缓冲区已满或空时，生产者或消费者线程停止自己的执行，放弃锁，使自己处于等待状态，让另一个线程开始执行。  
notify()方法表示：当生产者或消费者对缓冲区放入或取出一个产品时，向另一个线程发出可执行通知，同时放弃锁，使自己处于等待状态。  
  
import java.util.LinkedList;  
public class Sycn1  
\{  
private LinkedList<Object> myList = new LinkedList<Object>();  
private int MAX = 10;  
  
public void start()  
\{  
new Thread(new Producer()).start();  
new Thread(new Consumer()).start();  
\}  
  
public static void main(String\[\] args) throws Exception  
\{  
Sycn1 s1 = new Sycn1();  
s1.start();  
\}  
  
class Producer implements Runnable  
\{  
public void run()  
\{  
while (true)  
\{  
synchronized (myList)  
\{  
try  
\{  
while (myList.size() == MAX)  
\{  
System.out.println("warning: it's full!");  
myList.wait();  
\}  
Object o = new Object();  
if (myList.add(o))  
\{  
System.out.println("Producer: " + o);  
myList.notify();  
\}  
\}  
catch (InterruptedException ie)  
\{  
System.out.println("producer is interrupted!");  
\}  
\}  
\}  
\}  
\}  
  
class Consumer implements Runnable  
\{  
public void run()  
\{  
while (true)  
\{  
synchronized (myList)  
\{  
try  
\{  
while (myList.size() == 0)  
\{  
System.out.println("warning: it's empty!");  
myList.wait();  
\}  
Object o = myList.removeLast();  
System.out.println("Consumer: " + o);  
myList.notify();  
\}  
catch (InterruptedException ie)  
\{  
System.out.println("consumer is interrupted!");  
\}  
\}  
\}  
\}  
\}  
  
\}  
  
**二、方法await()/signal()**  
在JDK5.0以后，JAVA提供了新的更加健壮的线程处理机制，包括了同步、锁定、线程池等等，可以实现更小粒度上的控制。await()和signal()就是其中用来同步的两种方法，功能基本上和wait()/notify()相同，完全可以取代它们，但是它们和新引入的锁定机制Lock直接挂钩，具有更大的灵活性。  
  
import java.util.LinkedList;  
import java.util.concurrent.locks.\*;  
public class Sycn2  
\{  
private LinkedList<Object> myList = new LinkedList<Object>();  
private int MAX = 10;  
private final Lock lock = new ReentrantLock();  
private final Condition full = lock.newCondition();  
private final Condition empty = lock.newCondition();  
  
public void start()  
\{  
new Thread(new Producer()).start();  
new Thread(new Consumer()).start();  
\}  
  
public static void main(String\[\] args) throws Exception  
\{  
Sycn2 s2 = new Sycn2();  
s2.start();  
\}  
  
class Producer implements Runnable  
\{  
public void run()  
\{  
while (true)  
\{  
lock.lock();  
try  
\{  
while (myList.size() == MAX)  
\{  
System.out.println("warning: it's full!");  
full.await();  
\}  
Object o = new Object();  
if (myList.add(o))  
\{  
System.out.println("Producer: " + o);  
empty.signal();  
\}  
\}  
catch (InterruptedException ie)  
\{  
System.out.println("producer is interrupted!");  
\}  
finally  
\{  
lock.unlock();  
\}  
\}  
\}  
\}  
  
class Consumer implements Runnable  
\{  
public void run()  
\{  
while (true)  
\{  
lock.lock();  
try  
\{  
while (myList.size() == 0)  
\{  
System.out.println("warning: it's empty!");  
empty.await();  
\}  
Object o = myList.removeLast();  
System.out.println("Consumer: " + o);  
full.signal();  
\}  
catch (InterruptedException ie)  
\{  
System.out.println("consumer is interrupted!");  
\}  
finally  
\{  
lock.unlock();  
\}  
\}  
\}  
\}  
\}  
  
**三、阻塞队列方法BlockingQueue**  
BlockingQueue也是JDK5.0的一部分，是一个已经在内部实现了同步的队列，实现方式采用的是第2种await()/signal()方法。它可以在生成对象时指定容量大小。它用于阻塞操作的是put()和take()方法。  
put()方法类似于上面的生产者线程，容量最大时自动阻塞。  
take()方法类似于上面的消费者线程，容量为0时自动阻塞。  
  
import java.util.concurrent.\*;  
public class Sycn3  
\{  
private LinkedBlockingQueue<Object> queue   
= new LinkedBlockingQueue<Object>(10);  
private int MAX = 10;  
  
public void start()  
\{  
new Thread(new Producer()).start();  
new Thread(new Consumer()).start();  
\}  
  
public static void main(String\[\] args) throws Exception  
\{  
Sycn3 s3 = new Sycn3();  
s3.start();  
\}  
  
class Producer implements Runnable  
\{  
public void run()  
\{  
while (true)  
\{  
try  
\{  
if (queue.size() == MAX)  
System.out.println("warning: it's full!");  
Object o = new Object();  
queue.put(o);  
System.out.println("Producer: " + o);  
\}  
catch (InterruptedException e)  
\{  
System.out.println("producer is interrupted!");  
\}  
\}  
\}  
\}  
  
class Consumer implements Runnable  
\{  
public void run()  
\{  
while (true)  
\{  
try  
\{  
if (queue.size() == 0)  
System.out.println("warning: it's empty!");  
Object o = queue.take();  
System.out.println("Consumer: " + o);  
\}  
catch (InterruptedException e)  
\{  
System.out.println("producer is interrupted!");  
\}  
\}  
\}  
\}  
  
\}  
运行一下代码发现问题：  
warning: it's full!  
Producer: java.lang.object@4526e2a  
可能这是因为put()和System.out.println()之间没有同步造成的。**真正的原因**是因为当缓冲区已满，生产者在put()操作时，put()内部调用了await()方法放弃了线程的执行，然后消费者线程执行，调用take()方法，take()内部调用了signal()方法，通知生产者线程可以执行，致使在消费者的println()还没运行的情况下生产者的println()先被执行，所以有了上面的输出。run()中的synchronized其实并没有起什么作用。对于BlockingQueue大家可以放心使用，这可不是它的问题，只是在和别的对象之间的同步有问题。对于这种多重嵌套同步的问题以后再谈。  
  
  
**四、管道方法PipedInputStream/PipedOutputStream**  
这个类位于java.io包中，是解决同步问题的最简单的办法，一个线程将数据写入管道，另一个线程从管道读取数据，这样便构成了一种生产者/消费者的缓冲区编程模式。  
在下面的代码中没有使用Object对象，而是简单的读写字节值，这是因为PipedInputStream/PipedOutputStream不允许传输对象，这是JAVA本身的一个bug,具体的大家可以看sun的解释：http://bugs.sun.com/bugdatabase/view\_bug.do?bug\_id=4131126import java.io.\*;  
  
public class Sycn4  
\{  
private PipedOutputStream pos;  
private PipedInputStream pis;  
  
public Sycn4()  
\{  
try  
\{  
pos = new PipedOutputStream();  
pis = new PipedInputStream(pos);  
\}  
catch (IOException e)  
\{  
System.out.println(e);  
\}  
\}  
  
public void start()  
\{  
new Producer().start();  
new Consumer().start();  
\}  
  
public static void main(String\[\] args) throws Exception  
\{  
Sycn4 s4 = new Sycn4();  
s4.start();  
\}  
  
class Producer extends Thread  
\{  
public void run()  
\{  
try  
\{  
while (true)  
\{  
int b = (int) (Math.random() \* 255);  
System.out.println("Producer: a byte, the value is " + b);  
pos.write(b);  
pos.flush();  
\}  
\}  
catch (Exception e)  
\{  
e.printStackTrace();  
\}  
finally  
\{  
try  
\{  
pos.close();  
pis.close();  
\}  
catch (IOException e)  
\{  
System.out.println(e);  
\}  
\}  
\}  
\}  
  
class Consumer extends Thread  
\{  
public void run()  
\{  
try  
\{  
while (true)  
\{  
int b = pis.read();  
System.out.println("Consumer: a byte, the value is " +   
                                        String.valueOf(b));  
\}  
\}  
catch (Exception e)  
\{  
e.printStackTrace();  
\}  
finally  
\{  
try  
\{  
pos.close();  
pis.close();  
\}  
catch (IOException e)  
\{  
System.out.println(e);  
\}  
\}  
\}  
\}  
\}  
原帖地址：http://blog.csdn.net/jhj735412/article/details/6931135