生产者消费者

package function.thread;

 

import java.util.ArrayList;

import java.util.Arrays;

import java.util.List;

import java.util.Random;

import java.util.concurrent.locks.Condition;

import java.util.concurrent.locks.Lock;

import java.util.concurrent.locks.ReentrantLock;

/**

功能：三个生产者，往容量最大为5的容器里put数据

三个消费者从容器中取数据。只要容器不为空，就可以取数据，

只要容器不满，就可以存放数据

容器里不能拥有相同的数据

*/

public class ConditionTest {

 

public static void main(String args[]) {

 

final String goonFlag[] = {"true"};

final BoundedBuffer bf = new BoundedBuffer();

System.out.println("**************************************** main thread begin********************************");

//刚开始是抢占式的，take和put线程都可能先执行

for (int i = 0; i < 3; i++) {

new Thread(" put thread "+i) {

public void run() {

try {

//这里是匿名的内部类，所以要使用final变量

while(goonFlag[0].equals("true")){

bf.put(new Random().nextInt(100));

}

} catch (InterruptedException e) {

e.printStackTrace();

}

}

}.start();

 

new Thread(" take thread "+i) {

public void run() {

try {

while(goonFlag[0].equals("true")){

try {

bf.take();

} catch (Exception e) {

e.printStackTrace();

}

}

} catch (Exception e) {

e.printStackTrace();

}

}

}.start();

}//for block,new thread

 

try {

System.out.println("**************************************** main thread sleep********************************");

Thread.sleep(15000);

goonFlag[0]= "false";

} catch (InterruptedException e) {

e.printStackTrace();

}

}//main method

}

 

class BoundedBuffer {

final Lock lock = new ReentrantLock();// 锁对象

final Condition putCond = lock.newCondition();// 队列未满，写线程可以满足执行条件

final Condition takeCond = lock.newCondition();// 读线程条件

 

final List items = new ArrayList();// 缓存队列

 

public void put(Object x) throws InterruptedException {

String threadName = Thread.currentThread().getName();

System.out.println("############### 【"+threadName+"】 come in put method...... ");

lock.lock();

 

System.out.println("############### 【"+threadName+"】 get lock and sleep 500 ms ");

Thread.sleep(500);

System.out.println("############### 【"+threadName+"】wake up。。。。。。 ");

try {

if (items.size()>=5){

System.out.println("############### 【"+threadName+"】 stack is full and await ");

takeCond.signalAll();//队列已经满了，读线程可以进行读取了

// 阻塞写线程

//一个线程被阻塞，从业务角度理解，它的使命已经完成

//如果要重新参与作业，就要重新参与线程的竞争

putCond.await();

}else{

if(!items.contains(x)){

items.add(x);// 赋值

System.out.println("############### 【"+threadName+"】 put: "+x);

System.out.println("############### 【"+threadName+"】 list length "+items.size()+", list content:"+items.toString());

takeCond.signalAll();//队列不为空，读线程可以进行读取了

}else{

System.out.println("############### 【"+threadName+"】 put same: "+x);

System.out.println("############### 【"+threadName+"】 list content:"+items.toString());

}

 

}

 

} finally {

System.out.println("############### 【"+threadName+"】 release the lock ");

lock.unlock();

}

}

 

public Object take() throws Exception {

String threadName = Thread.currentThread().getName();

System.out.println("############### 【"+threadName+"】 come in take method...... ");

lock.lock();

//int listSize = items.size();不能使用局部变量，因为如果使用，得到的可能是await之前的值

//await期间，该值可能已经被修改

System.out.println("**************** 【"+threadName+"】 get a lock and sleep 500ms ");

Thread.sleep(500);

System.out.println("**************** 【"+threadName+"】 wake up。。。。。。 ");

try {

if (items.size() == 0){

System.out.println("**************** 【"+threadName+"】 stack is empty and await ");

putCond.signalAll();// 队列为空，就可以继续写入，也就是可以唤醒写线程

//当前线程调用了condition，被打上了相应的标签

//阻塞读线程。这里需要注意，被贴上了参与共享资源标签的线程在被唤醒后,

//可能会和还未打上标签的线程竞争。还有一种就是被打上了标签之后的多个线程同时竞争

takeCond.await();

return null;

}else{

Object x = items.get(items.size()-1);// 取值

System.out.println("**************** 【"+threadName+"】 get the value: "+x);

items.remove(x);

System.out.println("**************** 【"+threadName+"】 remain list : "+items.toString());

putCond.signalAll();// 队列未满，就可以继续写入，也就是可以唤醒写线程

return x;

}

 

 

}catch(Exception e){

e.printStackTrace();

return null;

}finally {

System.out.println("**************** 【"+threadName+"】 release the lock ");

lock.unlock();

}

}

}