由生产者/消费者问题看JAVA多线程

青旅半醒 2022-05-25 14:06 157阅读 0赞

转:[http://blog.csdn.net/coutcin/article/details/1213834][http_blog.csdn.net_coutcin_article_details_1213834]

生产者消费者问题是研究多线程程序时绕不开的问题，它的描述是有一块生产者和消费者共享的有界缓冲区，生产者往缓冲区放入产品，消费者从缓冲区取走产品，这个过程可以无休止的执行，不能因缓冲区满生产者放不进产品而终止，也不能因缓冲区空消费者无产品可取而终止。

解决生产者消费者问题的方法有两种，一种是采用某种机制保持生产者和消费者之间的同步，一种是在生产者和消费者之间建立一个管道。前一种有较高的效率并且可控制性较好，比较常用，后一种由于管道缓冲区不易控制及被传输数据对象不易封装等原因，比较少用。

同步问题的核心在于，CPU是按时间片轮询的方式执行程序，我们无法知道某一个线程是否被执行、是否被抢占、是否结束等，因此生产者完全可能当缓冲区已满的时候还在放入产品，消费者也完全可能当缓冲区为空时还在取出产品。

现在同步问题的解决方法一般是采用信号或者加锁机制，即生产者线程当缓冲区已满时放弃自己的执行权，进入等待状态，并通知消费者线程执行。消费者线程当缓冲区已空时放弃自己的执行权，进入等待状态，并通知生产者线程执行。这样一来就保持了线程的同步，并避免了线程间互相等待而进入死锁状态。

JAVA语言提供了独立于平台的线程机制，保持了”write once, run anywhere”的特色。同时也提供了对同步机制的良好支持。

在JAVA中，一共有四种方法支持同步，其中三个是同步方法，一个是管道方法。

1.        方法 wait()/notify()

2.        方法await()/signal()

3.        阻塞队列方法BlockingQueue

4.        管道方法PipedInputStream/PipedOutputStream

下面我们看各个方法的实现：

1.        方法wait()/notify()

wait()和notify()是根类Object的两个方法，也就意味着所有的JAVA类都会具有这个两个方法，为什么会被这样设计呢？我们可以认为所有的对象默认都具有一个锁，虽然我们看不到，也没有办法直接操作，但它是存在的。

wait()方法表示：当缓冲区已满或空时，生产者或消费者线程停止自己的执行，放弃锁，使自己处于等待状态，让另一个线程开始执行；

notify()方法表示：当生产者或消费者对缓冲区放入或取出一个产品时，向另一个线程发出可执行通知，同时放弃锁，使自己处于等待状态。

下面是一个例子代码：

import java.util.LinkedList;
    import java.util.List;
    import java.util.Random;
    //使用wait和notify来实现生产者和消费者的关系
    public class Sycn1 {
    	private List<Integer> list = new LinkedList<Integer>();
    	private int MaxNum = 10;
    	public static void main(String[] args) {
    		new Sycn1().start();
    	}
    	private void start(){
    		new Product().start();
    		new Consumer().start();
    	}
    	private static int getInt(){
    		try {
    			Thread.sleep(500);
    		} catch (InterruptedException e) {e.printStackTrace();}
    		return new Random().nextInt(9)+1;
    	}
    	class Product extends Thread{
    		@Override
    		public void run() {
    			while(true){
    				synchronized(list){
    					while(list.size() == MaxNum){//使用while比使用if要安全点,因为可能消费异常但notify
    						try {
    							System.out.println("生产已满.等待消费.....");
    							list.wait();
    						} catch (InterruptedException e) {e.printStackTrace();}
    					}
    					int i = getInt();
    					if(list.add(i)){
    						System.out.println("生产元素为:"+i+"#当前元素列表:"+list);
    						list.notify();
    					}
    				}
    			}
    		}
    	}
    	class Consumer extends Thread{
    		@Override
    		public void run() {
    			while(true){
    				synchronized(list){
    					while(list.size()==0){
    						try {
    							System.out.println("消费已空.等待生产.....");
    							list.wait();
    						} catch (InterruptedException e) {e.printStackTrace();}
    					}
    					int i = list.remove(list.size()-1);
    					System.out.println("消费元素为:"+i+"#当前元素列表:"+list);
    					list.notify();
    				}
    			}
    		}
    	}
    	/**
    	 * 运行结果:
    	 *  生产元素为:3#当前元素列表:[3]
    		生产元素为:9#当前元素列表:[3, 9]
    		生产元素为:9#当前元素列表:[3, 9, 9]
    		消费元素为:9#当前元素列表:[3, 9]
    		消费元素为:9#当前元素列表:[3]
    		消费元素为:3#当前元素列表:[]
    		消费已空.等待生产.....
    		生产元素为:3#当前元素列表:[3]
    		消费元素为:3#当前元素列表:[]
    		消费已空.等待生产.....
    		生产元素为:2#当前元素列表:[2]
    	 */
    }

2.        方法await()/signal()

在JDK5.0以后，JAVA提供了新的更加健壮的线程处理机制，包括了同步、锁定、线程池等等，它们可以实现更小粒度上的控制。await()和signal()就是其中用来做同步的两种方法，它们的功能基本上和wait()/notify()相同，完全可以取代它们，但是它们和新引入的锁定机制Lock直接挂钩，具有更大的灵活性。

下面是一个例子代码：

import java.util.LinkedList;
    import java.util.Random;
    import java.util.concurrent.locks.Condition;
    import java.util.concurrent.locks.Lock;
    import java.util.concurrent.locks.ReentrantLock;
    public class Sycn2 {
    	private Lock lock = new ReentrantLock();
    	private Condition full = lock.newCondition();
    	private Condition empty = lock.newCondition();
    	private int MaxNum = 5;
    	private LinkedList<Integer> list = new LinkedList<Integer>();
    	public static void main(String[] args) {
    		new Sycn2().start();
    	}
    	public void start(){
    		new Product().start();
    		new Consumer().start();
    	}
    	private static void mysleep(){
    		try {
    			Thread.sleep(500);
    		} catch (InterruptedException e) {
    			e.printStackTrace();
    		}
    	}
    	class Product extends Thread{
    		@Override
    		public void run(){
    			while(true){
    				try{
    					lock.lock();
    					while(list.size() == MaxNum){
    						System.out.println("product---await----");
    						full.await();
    					}
    					int rand = new Random().nextInt(100);
    					list.add(rand);
    					System.out.println("Product@"+list+"#"+rand);
    					empty.signal();
    					mysleep();
    				}catch(Exception e){
    					e.printStackTrace();
    				}finally{
    					lock.unlock();
    				}
    			}
    		}
    	}
    	class Consumer extends Thread{
    		@Override
    		public void run(){
    			while(true){
    				try{
    					lock.lock();
    					while(list.size() == 0){
    						System.out.println("consumer---await----");
    						empty.await();
    					}
    					System.out.println("Consumer@"+list+"#"+list.removeLast());
    					full.signal();
    					mysleep();
    				}catch(Exception e){
    					e.printStackTrace();
    				}finally{
    					lock.unlock();
    				}
    			}
    		}
    	}
    	/**
    	 * 运行结果：
    	    Product@[60]#60
    		Product@[60, 58]#58
    		Product@[60, 58, 81]#81
    		Product@[60, 58, 81, 41]#41
    		Product@[60, 58, 81, 41, 0]#0
    		product---await----
    		Consumer@[60, 58, 81, 41, 0]#0
    		Consumer@[60, 58, 81, 41]#41
    		Consumer@[60, 58, 81]#81
    		Consumer@[60, 58]#58
    		Consumer@[60]#60
    		consumer---await----
    		Product@[27]#27
    	 */
    }

3.        阻塞队列方法BlockingQueue

注：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 Sycn3(){
        }
        
        public void start(){
                new Producer().start();
                new Consumer().start();
        }
        
        public static void main(String[] args) throws Exception{
            Sycn3 s3 = new Sycn3();
            s3.start();
        }
        
        class Producer extends Thread{        
            public void run(){
                while(true){
                    //synchronized(this){
                    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 extends Thread{
            public void run(){
                while(true){
                    //synchronized(this){
                    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()之间没有同步造成的，我也这样认为，我也这样认为，但是你把run()中的synchronized前面的注释去掉，重新编译运行，有改观吗？没有。为什么？

这是因为，当缓冲区已满，生产者在put()操作时，put()内部调用了await()方法，放弃了线程的执行，然后消费者线程执行，调用take()方法，take()内部调用了signal()方法，通知生产者线程可以执行，致使在消费者的println()还没运行的情况下生产者的println()先被执行，所以有了上面的输出。run()中的synchronized其实并没有起什么作用。

对于BlockingQueue大家可以放心使用，这可不是它的问题，只是在它和别的对象之间的同步有问题。

对于这种多重嵌套同步的问题，以后再谈吧，欢迎大家讨论啊！

4.        管道方法PipedInputStream/PipedOutputStream

这个类位于java.io包中，是解决同步问题的最简单的办法，一个线程将数据写入管道，另一个线程从管道读取数据，这样便构成了一种生产者/消费者的缓冲区编程模式。

管道流可以用于线程间通信，当程序中存在多线程，并且某个线程的运行需要得到另一个线程的信息时， 可以用管道流作为媒介来传输信息。管道流属于线程间通信的一种

下面是一个例子代码，在这个代码我没有使用Object对象，而是简单的读写字节值，这是因为PipedInputStream/PipedOutputStream不允许传输对象，这是JAVA本身的一个bug,具体的大家可以看sun的解释： [http://bugs.sun.com/bugdatabase/view\_bug.do?bug\_id=4131126][http_bugs.sun.com_bugdatabase_view_bug.do_bug_id_4131126]

import java.io.IOException;
    import java.io.PipedInputStream;
    import java.io.PipedOutputStream;
    import java.util.Random;
    public class Sycn4 {
    	private PipedOutputStream pos;
    	private PipedInputStream pis;
    	public Sycn4(){
    		pos = new PipedOutputStream();
    		try {
    			pis = new PipedInputStream(pos);
    		} catch (IOException e) {
    			e.printStackTrace();
    		}
    	}
    	public static void main(String[] args) {
    		new Sycn4().start();
    	}
    	public void start(){
    		new Product().start();
    		new Consumer().start();
    	}
    	private static void mysleep(){
    		try {
    			Thread.sleep(500);
    		} catch (InterruptedException e) {
    			e.printStackTrace();
    		}
    	}
    	class Product extends Thread{
    		@Override
    		public void run(){
    			try {
    				while(true){
    					int rand = new Random().nextInt(100);
    					pos.write(rand);
    					pos.flush();
    					System.out.println("product@"+rand);
    					mysleep();
    				}
    			} catch (IOException e) {
    				e.printStackTrace();
    			}finally{
    				try {
    					pos.close();
    					pis.close();
    				} catch (IOException e) {
    					e.printStackTrace();
    				}
    			}
    		}
    	}
    	class Consumer extends Thread{
    		@Override
    		public void run(){
    			try {
    				while(true){
    					System.out.println("Consumer@"+pis.read());
    					mysleep();
    				}
    			} catch (IOException e) {
    				e.printStackTrace();
    			}finally{
    				try {
    					pos.close();
    					pis.close();
    				} catch (IOException e) {
    					e.printStackTrace();
    				}
    			}
    		}
    	}
    	/**
    	 * 运行结果：
    	 *  product@37
    		Consumer@37
    		product@98
    		Consumer@98
    		product@20
    		Consumer@20
    		product@70
    		Consumer@70
    	 */
    }

5.使用信号量的方式实现，转:[http://blog.csdn.net/a442180673/article/details/7309954][http_blog.csdn.net_a442180673_article_details_7309954]

import java.util.Random;
    import java.util.concurrent.Semaphore;
    /**
     * 生产者与消费者的问题
     * 
     */
    class Signs {
    	static Semaphore empty = new Semaphore(10); // 信号量：记录仓库空的位置，代表可以生产的最大个数
    	static Semaphore full = new Semaphore(0); // 信号量：记录仓库满的位置
    	static Semaphore mutex = new Semaphore(1); // 临界区互斥访问信号量， 相当于互斥锁。
    }
    
    // 生产一个产品
    class Producer implements Runnable {
    	Random r = new Random(1000);
    
    	public void run() {
    		try {
    			while (true) {
    				Signs.empty.acquire(); // 递减仓库空信号量
    				Signs.mutex.acquire(); // 进入临界区
    				SemaphoreTest.COUNT++;
    				System.out.println("生产一个产品放入仓库,剩余：" + SemaphoreTest.COUNT);
    
    				Signs.mutex.release(); // 离开临界区
    				Signs.full.release(); // 递增仓库满信号量
    				Thread.currentThread().sleep(100);
    			}
    		} catch (InterruptedException e) {
    			e.printStackTrace();
    		}
    	}
    }
    
    // 消费一个产品
    class Consumer implements Runnable {
    
    	public void run() {
    		try {
    			while (true) {
    				Signs.full.acquire(); // 递减仓库满信号量
    				Signs.mutex.acquire();
    				SemaphoreTest.COUNT--;
    				System.out.println("C1拿出一个产品消费，剩余：" + SemaphoreTest.COUNT);
    
    				Signs.mutex.release();
    				Signs.empty.release(); // 递增仓库空信号量
    				Thread.currentThread().sleep((1000));
    			}
    		} catch (InterruptedException e) {
    			e.printStackTrace();
    		}
    	}
    }
    
    class Consumer2 implements Runnable {
    
    	public void run() {
    		try {
    			while (true) {
    				Signs.full.acquire(); // 递减仓库满信号量
    				Signs.mutex.acquire();
    				SemaphoreTest.COUNT--;
    				System.out.println("C2拿出一个产品消费，剩余：" + SemaphoreTest.COUNT);
    
    				Signs.mutex.release();
    				Signs.empty.release(); // 递增仓库空信号量
    				Thread.currentThread().sleep((500));
    			}
    		} catch (InterruptedException e) {
    			e.printStackTrace();
    		}
    	}
    }
    
    public class SemaphoreTest {
    	public static int COUNT = 0;
    	// 产品数目
    	public static void main(String[] args) {
    		new Thread(new Producer()).start();
    		new Thread(new Consumer()).start();
    		new Thread(new Consumer2()).start();
    	}
    	/**
    	 * 运行结果：
    	 *  生产一个产品放入仓库,剩余：1
    		C2拿出一个产品消费，剩余：0
    		生产一个产品放入仓库,剩余：1
    		C1拿出一个产品消费，剩余：0
    		生产一个产品放入仓库,剩余：1
    		生产一个产品放入仓库,剩余：2
    		生产一个产品放入仓库,剩余：3
    		C2拿出一个产品消费，剩余：2
    		生产一个产品放入仓库,剩余：3
    		生产一个产品放入仓库,剩余：4
    		生产一个产品放入仓库,剩余：5
    		生产一个产品放入仓库,剩余：6
    		生产一个产品放入仓库,剩余：7
    		C2拿出一个产品消费，剩余：6
    		生产一个产品放入仓库,剩余：7
    		C1拿出一个产品消费，剩余：6
    		生产一个产品放入仓库,剩余：7
    		生产一个产品放入仓库,剩余：8
    		生产一个产品放入仓库,剩余：9
    		生产一个产品放入仓库,剩余：10
    		C2拿出一个产品消费，剩余：9
    	 */
    }

[http_blog.csdn.net_coutcin_article_details_1213834]: http://blog.csdn.net/coutcin/article/details/1213834
[http_bugs.sun.com_bugdatabase_view_bug.do_bug_id_4131126]: http://bugs.sun.com/bugdatabase/view_bug.do?bug_id=4131126
[http_blog.csdn.net_a442180673_article_details_7309954]: http://blog.csdn.net/a442180673/article/details/7309954