欢迎您访问程序员文章站本站旨在为大家提供分享程序员计算机编程知识!
您现在的位置是: 首页

多线程程序一段问题代码分析(生产者消费者)

程序员文章站 2022-07-15 15:23:04
...

问题在《一个java生产者消费者代码的问题》,解答一下,只解释原因,其他不考虑。

 

作者要的是一个生产者生成,接着必须有一个消费者消费,那这不是需要单线程吗?或者使用1个大小的阻塞队列。所以只谈论问题本身,不谈论好不好。

 

具体代码:

import java.util.concurrent.locks.Condition;

import java.util.concurrent.locks.Lock;

import java.util.concurrent.locks.ReentrantLock;


//生产/消费者模式

public class Basket {

    Lock lock = new ReentrantLock();


    // 产生Condition对象

    Condition produced = lock.newCondition();

    Condition consumed = lock.newCondition();

    boolean available = false;


    public void produce() throws InterruptedException {

        lock.lock();


        try {

            if (available) {

                produced.await(); // 放弃lock进入睡眠

            }


            System.out.println("Apple produced.");


            available = true;


            consumed.signal(); // 发信号唤醒等待这个Condition的线程

        } finally {

            lock.unlock();

        }

    }


    public void consume() throws InterruptedException {

        lock.lock();


        try {

            if (!available) {

                consumed.await(); // 放弃lock进入睡眠

            }


            /* 吃苹果 */

            System.out.println("Apple consumed.");


            available = false;


            produced.signal(); // 发信号唤醒等待这个Condition的线程

        } finally {

            lock.unlock();

        }

    }

}

 

import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

//测试用类
public class ConditionTester {

    public static void main(String[] args) throws InterruptedException {
        final Basket basket = new Basket();

        // 定义一个producer
        Runnable producer = new Runnable() {
            public void run() {
                try {
                    basket.produce();
                } catch (InterruptedException ex) {
                    ex.printStackTrace();
                }
            }
        };

        // 定义一个consumer
        Runnable consumer = new Runnable() {
            public void run() {
                try {
                    basket.consume();
                } catch (InterruptedException ex) {
                    ex.printStackTrace();
                }
            }
        };

        // 各产生10个consumer和producer
        ExecutorService service = Executors.newCachedThreadPool();

        for (int i = 0; i < 4; i++)
            service.submit(consumer);

        Thread.sleep(2000 * 2);

        for (int i = 0; i < 4; i++)
            service.submit(producer);

        service.shutdown();
    }
}

 

原因分析:

1、假设前面有2个producer(此时available=true)

1.1、一个在等待lock

1.2、一个await

2、consumer生成内容后,available=false,produced.signal(); 最后lock.unlock();

3.1、因为lock.unlock所以会触发一个lock获取到锁(虽然signal也会触发等待这个条件的其他线程,但是多线程大家都知道什么时候触发这是不确定的),如果此时正好是[1.1]那么因为available=false,执行完释放锁

3.2、produced.signal()所以会触发一个await的producer;

 

解决方案:

只要保证[3.1]还是需要await即可解决问题

 

所以加一个 AtomicInteger producedAwaitCounter = new AtomicInteger(0); 统计当前等待的生产者,如果当前available=false,但已经有生产者生成了内容,那么先等待消费者消费了再说

 

            if (available || producedAwaitCounter.get() > 0) {

                producedAwaitCounter.incrementAndGet();

                produced.await(); // 放弃lock进入睡眠

                producedAwaitCounter.decrementAndGet();

            }

 

当然最简单的是使用:自旋,原理可以自己分析下:

            while (available) {

                produced.await(); // 放弃lock进入睡眠

            }

 

 

package com.sishuok.es.test;

import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.locks.Condition;

import java.util.concurrent.locks.Lock;

import java.util.concurrent.locks.ReentrantLock;


//生产/消费者模式

public class Basket {

    Lock lock = new ReentrantLock(true);


// 产生Condition对象

    Condition produced = lock.newCondition();

    Condition consumed = lock.newCondition();

    boolean available = false;
    AtomicInteger producedAwaitCounter = new AtomicInteger(0);


    public void produce() throws InterruptedException {

        lock.lock();

        try {

            if (available || producedAwaitCounter.get() > 0) {
                producedAwaitCounter.incrementAndGet();
                produced.await(); // 放弃lock进入睡眠
                producedAwaitCounter.decrementAndGet();
            }

            System.out.println("Apple produced.");


            available = true;


            consumed.signal(); // 发信号唤醒等待这个Condition的线程

        } finally {
            lock.unlock();
        }

    }


    public void consume() throws InterruptedException {

        lock.lock();


        try {

            if (!available) {
                consumed.await(); // 放弃lock进入睡眠
            }

            /* 吃苹果 */

            System.out.println("Apple consumed.");


            available = false;

            produced.signal(); // 发信号唤醒等待这个Condition的线程
        } finally {
            lock.unlock();
        }

    }

}

 

在回答里还一个类似的,不过还是不太一样。

http://blog.csdn.net/liguogangde/article/details/9103501