前往小程序,Get更优阅读体验!
立即前往
首页
学习
活动
专区
工具
TVP
发布
社区首页 >专栏 >由生产者/消费者问题看JAVA多线程

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

作者头像
全栈程序员站长
发布2021-05-19 11:10:49
5690
发布2021-05-19 11:10:49
举报
文章被收录于专栏:全栈程序员必看

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

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

同步问题的核心在于,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()方法表示:当生产者或消费者对缓冲区放入或取出一个产品时,向另一个线程发出可执行通知,同时放弃锁,使自己处于等待状态。

下面是一个例子代码:

代码语言:javascript
复制
import
 java.util.LinkedList;
由生产者/消费者问题看JAVA多线程
由生产者/消费者问题看JAVA多线程
public
 
class
 Sycn1

{

由生产者/消费者问题看JAVA多线程    private LinkedList<Object> myList =new LinkedList<Object>();
由生产者/消费者问题看JAVA多线程    private int MAX = 10;
由生产者/消费者问题看JAVA多线程    
由生产者/消费者问题看JAVA多线程    public Sycn1(){

由生产者/消费者问题看JAVA多线程    }
由生产者/消费者问题看JAVA多线程    
由生产者/消费者问题看JAVA多线程    public void start(){

由生产者/消费者问题看JAVA多线程            new Producer().start();
由生产者/消费者问题看JAVA多线程            new Consumer().start();
由生产者/消费者问题看JAVA多线程    }
由生产者/消费者问题看JAVA多线程    
由生产者/消费者问题看JAVA多线程    public static void main(String[] args) throws Exception{

由生产者/消费者问题看JAVA多线程        Sycn1 s1 = new Sycn1();
由生产者/消费者问题看JAVA多线程        s1.start();
由生产者/消费者问题看JAVA多线程    }
由生产者/消费者问题看JAVA多线程    
由生产者/消费者问题看JAVA多线程    class Producer extends Thread{        
由生产者/消费者问题看JAVA多线程        public void run(){

由生产者/消费者问题看JAVA多线程            while(true){

由生产者/消费者问题看JAVA多线程                synchronized(myList){

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

由生产者/消费者问题看JAVA多线程                        while(myList.size() == MAX){

由生产者/消费者问题看JAVA多线程                            System.out.println(“warning: it’s full!“);
由生产者/消费者问题看JAVA多线程                            myList.wait();
由生产者/消费者问题看JAVA多线程                        }
由生产者/消费者问题看JAVA多线程                        Object o = new Object();
由生产者/消费者问题看JAVA多线程                        if(myList.add(o)){

由生产者/消费者问题看JAVA多线程                            System.out.println(“Producer: “ + o);
由生产者/消费者问题看JAVA多线程                            myList.notify();
由生产者/消费者问题看JAVA多线程                        }
由生产者/消费者问题看JAVA多线程                    }catch(InterruptedException ie){

由生产者/消费者问题看JAVA多线程                        System.out.println(“producer is interrupted!“);
由生产者/消费者问题看JAVA多线程                    }
由生产者/消费者问题看JAVA多线程                }
由生产者/消费者问题看JAVA多线程            }
由生产者/消费者问题看JAVA多线程        }
由生产者/消费者问题看JAVA多线程    }
由生产者/消费者问题看JAVA多线程    
由生产者/消费者问题看JAVA多线程    class Consumer extends Thread{

由生产者/消费者问题看JAVA多线程        public void run(){

由生产者/消费者问题看JAVA多线程            while(true){

由生产者/消费者问题看JAVA多线程                synchronized(myList){

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

由生产者/消费者问题看JAVA多线程                        while(myList.size() == 0){

由生产者/消费者问题看JAVA多线程                            System.out.println(“warning: it’s empty!“);
由生产者/消费者问题看JAVA多线程                            myList.wait();
由生产者/消费者问题看JAVA多线程                        }
由生产者/消费者问题看JAVA多线程                        Object o = myList.removeLast();
由生产者/消费者问题看JAVA多线程                        System.out.println(“Consumer: “ + o);
由生产者/消费者问题看JAVA多线程                        myList.notify();
由生产者/消费者问题看JAVA多线程                    }catch(InterruptedException ie){

由生产者/消费者问题看JAVA多线程                        System.out.println(“consumer is interrupted!“);
由生产者/消费者问题看JAVA多线程                    }
由生产者/消费者问题看JAVA多线程                }
由生产者/消费者问题看JAVA多线程            }
由生产者/消费者问题看JAVA多线程        }
由生产者/消费者问题看JAVA多线程    }
由生产者/消费者问题看JAVA多线程    
由生产者/消费者问题看JAVA多线程}

2. 方法await()/signal()

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

下面是一个例子代码:

代码语言:javascript
复制
import
 java.util.LinkedList;
由生产者/消费者问题看JAVA多线程
由生产者/消费者问题看JAVA多线程
import
 java.util.concurrent.locks.
*
;
由生产者/消费者问题看JAVA多线程
由生产者/消费者问题看JAVA多线程
public
 
class
 Sycn2

{

由生产者/消费者问题看JAVA多线程    private LinkedList<Object> myList = new LinkedList<Object>();
由生产者/消费者问题看JAVA多线程    private int MAX = 10;
由生产者/消费者问题看JAVA多线程    private final Lock lock = new ReentrantLock();
由生产者/消费者问题看JAVA多线程    private final Condition full = lock.newCondition();
由生产者/消费者问题看JAVA多线程    private final Condition empty = lock.newCondition();
由生产者/消费者问题看JAVA多线程    
由生产者/消费者问题看JAVA多线程    public Sycn2(){

由生产者/消费者问题看JAVA多线程    }
由生产者/消费者问题看JAVA多线程    
由生产者/消费者问题看JAVA多线程    public void start(){

由生产者/消费者问题看JAVA多线程            new Producer().start();
由生产者/消费者问题看JAVA多线程            new Consumer().start();
由生产者/消费者问题看JAVA多线程    }
由生产者/消费者问题看JAVA多线程    
由生产者/消费者问题看JAVA多线程    public static void main(String[] args) throws Exception{

由生产者/消费者问题看JAVA多线程        Sycn2 s2 = new Sycn2();
由生产者/消费者问题看JAVA多线程        s2.start();
由生产者/消费者问题看JAVA多线程    }
由生产者/消费者问题看JAVA多线程    
由生产者/消费者问题看JAVA多线程    class Producer extends Thread{        
由生产者/消费者问题看JAVA多线程        public void run(){

由生产者/消费者问题看JAVA多线程            while(true){

由生产者/消费者问题看JAVA多线程                lock.lock();
由生产者/消费者问题看JAVA多线程                try{

由生产者/消费者问题看JAVA多线程                    while(myList.size() == MAX){

由生产者/消费者问题看JAVA多线程                        System.out.println(“warning: it’s full!“);
由生产者/消费者问题看JAVA多线程                        full.await();
由生产者/消费者问题看JAVA多线程                    }
由生产者/消费者问题看JAVA多线程                    Object o = new Object();
由生产者/消费者问题看JAVA多线程                    if(myList.add(o)){

由生产者/消费者问题看JAVA多线程                        System.out.println(“Producer: “ + o);
由生产者/消费者问题看JAVA多线程                        empty.signal();
由生产者/消费者问题看JAVA多线程                    }
由生产者/消费者问题看JAVA多线程                }catch(InterruptedException ie){

由生产者/消费者问题看JAVA多线程                    System.out.println(“producer is interrupted!“);
由生产者/消费者问题看JAVA多线程                }finally{

由生产者/消费者问题看JAVA多线程                    lock.unlock();
由生产者/消费者问题看JAVA多线程                }
由生产者/消费者问题看JAVA多线程            }
由生产者/消费者问题看JAVA多线程        }
由生产者/消费者问题看JAVA多线程    }
由生产者/消费者问题看JAVA多线程    
由生产者/消费者问题看JAVA多线程    class Consumer extends Thread{

由生产者/消费者问题看JAVA多线程        public void run(){

由生产者/消费者问题看JAVA多线程            while(true){

由生产者/消费者问题看JAVA多线程                lock.lock();
由生产者/消费者问题看JAVA多线程                try{

由生产者/消费者问题看JAVA多线程                    while(myList.size() == 0){

由生产者/消费者问题看JAVA多线程                        System.out.println(“warning: it’s empty!“);
由生产者/消费者问题看JAVA多线程                        empty.await();
由生产者/消费者问题看JAVA多线程                    }
由生产者/消费者问题看JAVA多线程                    Object o = myList.removeLast();
由生产者/消费者问题看JAVA多线程                    System.out.println(“Consumer: “ + o);
由生产者/消费者问题看JAVA多线程                    full.signal();
由生产者/消费者问题看JAVA多线程                }catch(InterruptedException ie){

由生产者/消费者问题看JAVA多线程                    System.out.println(“consumer is interrupted!“);
由生产者/消费者问题看JAVA多线程                }finally{

由生产者/消费者问题看JAVA多线程                    lock.unlock();
由生产者/消费者问题看JAVA多线程                }
由生产者/消费者问题看JAVA多线程            }
由生产者/消费者问题看JAVA多线程        }
由生产者/消费者问题看JAVA多线程    }
由生产者/消费者问题看JAVA多线程    
由生产者/消费者问题看JAVA多线程}

3. 阻塞队列方法BlockingQueue

BlockingQueue也是JDK5.0的一部分,它是一个已经在内部实现了同步的队列,实现方式采用的是我们的第2种await()/signal()方法。它可以在生成对象时指定容量大小。

它用于阻塞操作的是put()和take()方法。

put()方法类似于我们上面的生产者线程,容量最大时,自动阻塞。

take()方法类似于我们上面的消费者线程,容量为0时,自动阻塞。

下面是一个例子代码:

代码语言:javascript
复制
import
 java.util.concurrent.
*
;
由生产者/消费者问题看JAVA多线程
由生产者/消费者问题看JAVA多线程
public
 
class
 Sycn3

{

由生产者/消费者问题看JAVA多线程    private LinkedBlockingQueue<Object> queue = new LinkedBlockingQueue<Object>(10);
由生产者/消费者问题看JAVA多线程    private int MAX = 10;
由生产者/消费者问题看JAVA多线程    
由生产者/消费者问题看JAVA多线程    public Sycn3(){

由生产者/消费者问题看JAVA多线程    }
由生产者/消费者问题看JAVA多线程    
由生产者/消费者问题看JAVA多线程    public void start(){

由生产者/消费者问题看JAVA多线程            new Producer().start();
由生产者/消费者问题看JAVA多线程            new Consumer().start();
由生产者/消费者问题看JAVA多线程    }
由生产者/消费者问题看JAVA多线程    
由生产者/消费者问题看JAVA多线程    public static void main(String[] args) throws Exception{

由生产者/消费者问题看JAVA多线程        Sycn3 s3 = new Sycn3();
由生产者/消费者问题看JAVA多线程        s3.start();
由生产者/消费者问题看JAVA多线程    }
由生产者/消费者问题看JAVA多线程    
由生产者/消费者问题看JAVA多线程    class Producer extends Thread{        
由生产者/消费者问题看JAVA多线程        public void run(){

由生产者/消费者问题看JAVA多线程            while(true){

由生产者/消费者问题看JAVA多线程                //synchronized(this){

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

由生产者/消费者问题看JAVA多线程                    if(queue.size() == MAX)
由生产者/消费者问题看JAVA多线程                        System.out.println(“warning: it’s full!“);
由生产者/消费者问题看JAVA多线程                    Object o = new Object();
由生产者/消费者问题看JAVA多线程                    queue.put(o);
由生产者/消费者问题看JAVA多线程                    System.out.println(“Producer: “ + o);
由生产者/消费者问题看JAVA多线程                    }catch(InterruptedException e){

由生产者/消费者问题看JAVA多线程                        System.out.println(“producer is interrupted!“);
由生产者/消费者问题看JAVA多线程                    }
由生产者/消费者问题看JAVA多线程                //}
由生产者/消费者问题看JAVA多线程            }
由生产者/消费者问题看JAVA多线程        }
由生产者/消费者问题看JAVA多线程    }
由生产者/消费者问题看JAVA多线程    
由生产者/消费者问题看JAVA多线程    class Consumer extends Thread{

由生产者/消费者问题看JAVA多线程        public void run(){

由生产者/消费者问题看JAVA多线程            while(true){

由生产者/消费者问题看JAVA多线程                //synchronized(this){

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

由生产者/消费者问题看JAVA多线程                    if(queue.size() == 0)
由生产者/消费者问题看JAVA多线程                        System.out.println(“warning: it’s empty!“);
由生产者/消费者问题看JAVA多线程                    Object o = queue.take();
由生产者/消费者问题看JAVA多线程                    System.out.println(“Consumer: “ + o);
由生产者/消费者问题看JAVA多线程                    }catch(InterruptedException e){

由生产者/消费者问题看JAVA多线程                        System.out.println(“producer is interrupted!“);
由生产者/消费者问题看JAVA多线程                    }
由生产者/消费者问题看JAVA多线程                //}
由生产者/消费者问题看JAVA多线程            }
由生产者/消费者问题看JAVA多线程        }
由生产者/消费者问题看JAVA多线程    }
由生产者/消费者问题看JAVA多线程    
由生产者/消费者问题看JAVA多线程}

你发现这个例子中的问题了吗?

如果没有,我建议你运行一下这段代码,仔细观察它的输出,是不是有下面这个样子的?为什么会这样呢?

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

代码语言:javascript
复制
import
 java.io.
*
;
由生产者/消费者问题看JAVA多线程
由生产者/消费者问题看JAVA多线程
public
 
class
 Sycn4

{

由生产者/消费者问题看JAVA多线程    private PipedOutputStream pos;
由生产者/消费者问题看JAVA多线程    private PipedInputStream pis;
由生产者/消费者问题看JAVA多线程    //private ObjectOutputStream oos;
由生产者/消费者问题看JAVA多线程    //private ObjectInputStream ois;
由生产者/消费者问题看JAVA多线程    
由生产者/消费者问题看JAVA多线程    public Sycn4(){

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

由生产者/消费者问题看JAVA多线程            pos = new PipedOutputStream();
由生产者/消费者问题看JAVA多线程            pis = new PipedInputStream(pos);
由生产者/消费者问题看JAVA多线程            //oos = new ObjectOutputStream(pos);
由生产者/消费者问题看JAVA多线程            //ois = new ObjectInputStream(pis);
由生产者/消费者问题看JAVA多线程        }catch(IOException e){

由生产者/消费者问题看JAVA多线程            System.out.println(e);
由生产者/消费者问题看JAVA多线程        }
由生产者/消费者问题看JAVA多线程    }
由生产者/消费者问题看JAVA多线程    
由生产者/消费者问题看JAVA多线程    public void start(){

由生产者/消费者问题看JAVA多线程        new Producer().start();
由生产者/消费者问题看JAVA多线程        new Consumer().start();
由生产者/消费者问题看JAVA多线程    }
由生产者/消费者问题看JAVA多线程    
由生产者/消费者问题看JAVA多线程    public static void main(String[] args) throws Exception{

由生产者/消费者问题看JAVA多线程        Sycn4 s4 = new Sycn4();
由生产者/消费者问题看JAVA多线程        s4.start();
由生产者/消费者问题看JAVA多线程    }
由生产者/消费者问题看JAVA多线程    
由生产者/消费者问题看JAVA多线程    class Producer extends Thread{

由生产者/消费者问题看JAVA多线程        public void run() {

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

由生产者/消费者问题看JAVA多线程                while(true){

由生产者/消费者问题看JAVA多线程                    int b = (int) (Math.random() * 255);
由生产者/消费者问题看JAVA多线程                    System.out.println(“Producer: a byte, the value is “ + b);
由生产者/消费者问题看JAVA多线程                    pos.write(b);
由生产者/消费者问题看JAVA多线程                    pos.flush();
由生产者/消费者问题看JAVA多线程                    //Object o = new MyObject();
由生产者/消费者问题看JAVA多线程                    //oos.writeObject(o);
由生产者/消费者问题看JAVA多线程                    //oos.flush();
由生产者/消费者问题看JAVA多线程                    //System.out.println(“Producer: ” + o);
由生产者/消费者问题看JAVA多线程                }
由生产者/消费者问题看JAVA多线程            }catch(Exception e){

由生产者/消费者问题看JAVA多线程                //System.out.println(e);
由生产者/消费者问题看JAVA多线程                e.printStackTrace();
由生产者/消费者问题看JAVA多线程            }finally{

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

由生产者/消费者问题看JAVA多线程                    pos.close();
由生产者/消费者问题看JAVA多线程                    pis.close();
由生产者/消费者问题看JAVA多线程                    //oos.close();
由生产者/消费者问题看JAVA多线程                    //ois.close();
由生产者/消费者问题看JAVA多线程                }catch(IOException e){

由生产者/消费者问题看JAVA多线程                    System.out.println(e);
由生产者/消费者问题看JAVA多线程                }
由生产者/消费者问题看JAVA多线程            }
由生产者/消费者问题看JAVA多线程        }
由生产者/消费者问题看JAVA多线程    }
由生产者/消费者问题看JAVA多线程    
由生产者/消费者问题看JAVA多线程    class Consumer extends Thread{

由生产者/消费者问题看JAVA多线程        public void run(){

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

由生产者/消费者问题看JAVA多线程                while(true){

由生产者/消费者问题看JAVA多线程                    int b = pis.read();
由生产者/消费者问题看JAVA多线程                    System.out.println(“Consumer: a byte, the value is “ + String.valueOf(b));
由生产者/消费者问题看JAVA多线程                    //Object o = ois.readObject();
由生产者/消费者问题看JAVA多线程                    //if(o != null)
由生产者/消费者问题看JAVA多线程                        //System.out.println(“Consumer: ” + o);
由生产者/消费者问题看JAVA多线程                }
由生产者/消费者问题看JAVA多线程            }catch(Exception e){

由生产者/消费者问题看JAVA多线程                //System.out.println(e);
由生产者/消费者问题看JAVA多线程                e.printStackTrace();
由生产者/消费者问题看JAVA多线程            }finally{

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

由生产者/消费者问题看JAVA多线程                    pos.close();
由生产者/消费者问题看JAVA多线程                    pis.close();
由生产者/消费者问题看JAVA多线程                    //oos.close();
由生产者/消费者问题看JAVA多线程                    //ois.close();
由生产者/消费者问题看JAVA多线程                }catch(IOException e){

由生产者/消费者问题看JAVA多线程                    System.out.println(e);
由生产者/消费者问题看JAVA多线程                }
由生产者/消费者问题看JAVA多线程            }
由生产者/消费者问题看JAVA多线程        }
由生产者/消费者问题看JAVA多线程    }
由生产者/消费者问题看JAVA多线程    
由生产者/消费者问题看JAVA多线程    //class MyObject implements Serializable {

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

发布者:全栈程序员栈长,转载请注明出处:https://javaforall.cn/100531.html原文链接:

本文参与 腾讯云自媒体同步曝光计划,分享自作者个人站点/博客。
原始发表:2021年5月6日 p,如有侵权请联系 cloudcommunity@tencent.com 删除

本文分享自 作者个人站点/博客 前往查看

如有侵权,请联系 cloudcommunity@tencent.com 删除。

本文参与 腾讯云自媒体同步曝光计划  ,欢迎热爱写作的你一起参与!

评论
登录后参与评论
0 条评论
热度
最新
推荐阅读
领券
问题归档专栏文章快讯文章归档关键词归档开发者手册归档开发者手册 Section 归档