由生产者/消费者问题看JAVA多线程
2012-09-29 17:56
731 查看
转: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()方法表示:当生产者或消费者对缓冲区放入或取出一个产品时,向另一个线程发出可执行通知,同时放弃锁,使自己处于等待状态。
下面是一个例子代码:
2. 方法await()/signal()
在JDK5.0以后,JAVA提供了新的更加健壮的线程处理机制,包括了同步、锁定、线程池等等,它们可以实现更小粒度上的控制。await()和signal()就是其中用来做同步的两种方法,它们的功能基本上和wait()/notify()相同,完全可以取代它们,但是它们和新引入的锁定机制Lock直接挂钩,具有更大的灵活性。
下面是一个例子代码:
3. 阻塞队列方法BlockingQueue
注:BlockingQueue 可以安全地与多个生产者和多个使用者一起使用。
BlockingQueue也是JDK5.0的一部分,它是一个已经在内部实现了同步的队列,实现方式采用的是我们的第2种await()/signal()方法。它可以在生成对象时指定容量大小。
它用于阻塞操作的是put()和take()方法。
put()方法类似于我们上面的生产者线程,容量最大时,自动阻塞。
take()方法类似于我们上面的消费者线程,容量为0时,自动阻塞。
下面是一个例子代码:
你发现这个例子中的问题了吗?
如果没有,我建议你运行一下这段代码,仔细观察它的输出,是不是有下面这个样子的?为什么会这样呢?
…
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
5.使用信号量的方式实现,转:http://blog.csdn.net/a442180673/article/details/7309954
生产者消费者问题是研究多线程程序时绕不开的问题,它的描述是有一块生产者和消费者共享的有界缓冲区,生产者往缓冲区放入产品,消费者从缓冲区取走产品,这个过程可以无休止的执行,不能因缓冲区满生产者放不进产品而终止,也不能因缓冲区空消费者无产品可取而终止。
解决生产者消费者问题的方法有两种,一种是采用某种机制保持生产者和消费者之间的同步,一种是在生产者和消费者之间建立一个管道。前一种有较高的效率并且可控制性较好,比较常用,后一种由于管道缓冲区不易控制及被传输数据对象不易封装等原因,比较少用。
同步问题的核心在于,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
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
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 */ }
相关文章推荐
- java多线程之生产者消费者经典问题
- java多线程总结六:经典生产者消费者问题实现
- 黑马程序员-java多线程生产者消费者问题
- JAVA基础再回首(二十五)——Lock锁的使用、死锁问题、多线程生产者和消费者、线程池、匿名内部类使用多线程、定时器、面试题
- java 多线程学习之多生产者多消费者产生的线程安全问题分析与解决:Lock和Condition
- 多线程经典问题之生产者消费者问题的JAVA实现
- 由生产者/消费者问题看JAVA多线程
- Java多线程的一个案例-生产者消费者问题
- java多线程之生产者消费者经典问题
- JAVA--多线程生产者消费者问题详解
- java多线程之生产者消费者问题
- JAVA学习第二十七课(多线程(六))- 多生产者多消费者问题(JDK1.5新特性)
- java 多线程 详解 生产者消费者 问题解决
- java多线程之生产者消费者经典问题
- Java多线程学习之生产者消费者问题(一)
- java多线程之生产者消费者经典问题
- 由生产者/消费者问题看JAVA多线程
- 用JAVA中的多线程示例生产者和消费者问题
- Java基础_线程_多线程_生产者消费者问题
- Java 多线程:生产者消费者问题