java 生产者消费者

共享数据的访问都要放在锁中进行。

1 synchronized方法：

public class ProduceConsumer {

public static void main(String[] args) {
Myqueue queue = new Myqueue();

Producer p1 = new Producer(queue);
Producer p2 = new Producer(queue);
Producer p3 = new Producer(queue);
Producer p4 = new Producer(queue);

Consumer c1 = new Consumer(queue);
Consumer c2 = new Consumer(queue);

new Thread(p1).start();
new Thread(p2).start();
new Thread(p3).start();
new Thread(p4).start();
new Thread(c1).start();
new Thread(c2).start();
}

}

class Myqueue {
private LinkedList<String> list;
private int maxSize = 10;
private int count = 0;
private int index = 0;

public Myqueue() {
list = new LinkedList<String>();
}

public Myqueue(int max) {
this();
maxSize = max;
}

public synchronized void put(String str) {
if (!this.isFull()) {
list.addLast(str + "--------编号" + index);

try {
Thread.sleep(500);
} catch (InterruptedException e) {
// TODO Auto-generated catch block
e.printStackTrace();
}
System.out.println(Thread.currentThread().getName() + "生产者：  "
+ str + "+++++编号" + index);
count++;
index++;
this.notifyAll();
}
if (this.isFull()) {               //必须判断
try {
this.wait();
} catch (InterruptedException e) {
// TODO Auto-generated catch block
e.printStackTrace();
}
}
}

public synchronized void get() {
if (!this.isEmpty()) {
try {
Thread.sleep(500);
} catch (InterruptedException e) {
// TODO Auto-generated catch block
e.printStackTrace();
}
System.out.println(Thread.currentThread().getName() + "消费者： "
+ list.pollLast());
count--;
this.notifyAll();
}
if (this.isEmpty()) {    //必须判断
try {
this.wait();
} catch (InterruptedException e) {
// TODO Auto-generated catch block
e.printStackTrace();
}
}
}

public boolean isFull() {
return count == maxSize;
}

public boolean isEmpty() {
return count == 0;
}

}

class Producer implements Runnable {
private Myqueue queue;

public Producer(Myqueue queue) {
this.queue = queue;
}

@Override
public void run() {
while (true) {
queue.put("烤鱼");
}

}

}

class Consumer implements Runnable {
private Myqueue queue;

public Consumer(Myqueue queue) {
this.queue = queue;
}

@Override
public void run() {
while (true) {
queue.get();
}
}

}

2 ReentrantLock实现

import java.util.LinkedList;

import java.util.concurrent.ArrayBlockingQueue;

import java.util.concurrent.BlockingQueue;

import java.util.concurrent.locks.Condition;

import java.util.concurrent.locks.ReentrantLock;

public class ReentrantLocTest {

/**
* @param args
*/
public static void main(String[] args) {
MyFoods foods = new MyFoods(8);
Productor pd = new Productor(foods);
MyConsumer cs = new MyConsumer(foods);
new Thread(pd).start();
new Thread(pd).start();
new Thread(pd).start();
new Thread(cs).start();
new Thread(cs).start();
new Thread(cs).start();
new Thread(cs).start();
}

}

class MyFoods {
LinkedList<String> foods;
int maxcount;
int index;
ReentrantLock lock;
private Condition noFull;
private Condition noEmpty;

public MyFoods(int count) {
foods = new LinkedList<String>();
maxcount = count;
lock = new ReentrantLock();
noFull = lock.newCondition();
noEmpty = lock.newCondition();
}

public void produce() throws InterruptedException {
lock.lock();
try {
while (foods.size() == maxcount) {
// 已满阻塞
noFull.await();
}
// 可以生产
index++;
foods.addFirst("产品" + index);
System.out.println(Thread.currentThread().getName()
+ "生产++++++++++++产品" + index);
noEmpty.signalAll();
} finally {
lock.unlock();
}
}

public void consume() throws InterruptedException {
lock.lock();
try {
while (foods.isEmpty()) {
// 已空
noEmpty.await();
}
String str = foods.pollLast();
System.out.println(Thread.currentThread().getName() + "消费——"
+ str);
noFull.signalAll();
} finally {
lock.unlock();
}
}

}

class Productor implements Runnable {
MyFoods foods;

public Productor(MyFoods f) {
foods = f;
}

public void run() {
while (true) {
try {
foods.produce();
Thread.sleep(200);
} catch (InterruptedException e) {
e.printStackTrace();
}
}
}

}

class MyConsumer implements Runnable {
MyFoods foods;

public MyConsumer(MyFoods f) {
foods = f;
}

public void run() {
while (true) {
try {
foods.consume();
Thread.sleep(200);
} catch (InterruptedException e) {
e.printStackTrace();
}
}
}

}

结果如下：