问题在《一个java生产者消费者代码的问题》,解答一下,只解释原因,其他不考虑。
?
作者要的是一个生产者生成,接着必须有一个消费者消费,那这不是需要单线程吗?或者使用1个大小的阻塞队列。所以只谈论问题本身,不谈论好不好。
?
具体代码:
class="java" name="code">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
?