java多线程 消费者-生产者

java多线程一般都会讲消费者-生产者模型

生产者与消费者模型中，要保证以下几点：
1 同一时间内只能有一个生产者生产
2 同一时间内只能有一个消费者消费
3 生产者生产的同时消费者不能消费
4 消息队列满时生产者不能继续生产
5 消息队列空时消费者不能继续消费

参考了下网上一个代码实例http://www.talented.com.cn/archives/2007/5/16_141.html  发现作者写得有问题 修改了一些代码  现在ok了

----------------------------------------------------Message类

package com.example.test;

public class Message {
 
 public static int id;
 public String content;

 public String getContent() {
  return content;
 }

 public void setContent(String content) {
  this.content = content;
 }

 public int getId() {
  return id;
 }

 public void setId(int id) {
  Message.id = id;
 }
}

----------------------------------------------------Queue类

package com.example.test;

import java.util.ArrayList;
import java.util.List;

public class Queue {
 List<Message> queue = new ArrayList<Message>();

 /** 队列中message对象的最大值,默认为5 */
 int maxMessageNum = 5;

 public synchronized void produce(Message message) {

  this.notifyAll();
  while (queue.size() == maxMessageNum) {
   System.out.println(Thread.currentThread().getName()
     + "  队列满！等待中。。。");
   try {
    this.wait();
   } catch (InterruptedException e) {
    e.printStackTrace();
   }
  }
  queue.add(message);
  System.out.println(Thread.currentThread().getName() + "正在生产"
    + message.getContent() + "。。。  ,当前个数:" + getCount());

 }

 public synchronized void consume() {

  this.notifyAll();
  while (queue.size() == 0) {
   System.out.println(Thread.currentThread().getName()
     + "  队列空！等待中。。。");
   try {
    System.out.println("begin!");
    wait();
    System.out.println("end!");
   } catch (InterruptedException e) {
    e.printStackTrace();
   }
  }

  Message message = queue.get(0);
  queue.remove(0);
  System.out.println(Thread.currentThread().getName() + "正在消费"
    + message.getContent() + "。。。 ,当前个数: " + getCount());

 }

 public synchronized int getCount() {
  return queue.size();
 }
}


----------------------------------------------------Test类

package com.example.test;

public class Test {

 public static void main(String[] args) {

  Queue Q = new Queue();

  Producer wQ1 = new Producer(Q);
  Producer wQ2 = new Producer(Q);
 
  Consumer rQ1 = new Consumer(Q);
  Consumer rQ2 = new Consumer(Q);
  Consumer rQ3 = new Consumer(Q);
 
  Thread threadWQ1 = new Thread(wQ1, "thread-wQ1");
  Thread threadWQ2 = new Thread(wQ2, "thread-wQ2");

  Thread threadRQ1 = new Thread(rQ1, "thread-rQ1");
  Thread threadRQ2 = new Thread(rQ2, "thread-rQ2");
  Thread threadRQ3 = new Thread(rQ3, "thread-rQ3");
 
  threadWQ1.start();
  threadWQ2.start();
 
  threadRQ1.start();
  threadRQ2.start();
  threadRQ3.start();
 }
}

class Producer extends Thread {

 private Queue queue;

 Producer(Queue queue) {
  this.queue = queue;
 }

 public void run() {
 
  while (true) {
   Message message = new Message();
   message.setId(++Message.id);
   message.setContent("food"+Message.id);
   queue.produce(message);
   try {
    sleep(100);
   } catch (Exception e) {
   }
  }

 }
}

class Consumer extends Thread {
 private Queue queue;

 Consumer(Queue queue) {
  this.queue = queue;
 }

 public void run() {
  while (true) {
   queue.consume();
   try {
    sleep(100);
   } catch (Exception e) {
   }

  }
 }
}

评论

17 楼 newko 2010-03-27  

sw1982 写道

LinkedBolockQueue

生产者和消费者（BolockingQueue Implements）
http://compous-cn.iteye.com/blog/625888

16 楼 banfry 2010-03-16  

超级潜水艇 写道

asialee 写道

liwenso 写道

banfry 写道

问下各位，“  public synchronized void produce(Message message) {”
该方法锁定的资源是“message”还是“queue”？

锁定的是这个produce方法

这个应该是锁定的是Queue对象吧，要执行produce方法，必须要拿到Queue对象的锁。

锁对象，有人还没有搞清楚同步的对象是哪个哦！

锁定"queue"，会不会将message也附带锁上呢，毕竟message存放在queue里面？

15 楼 sw1982 2010-03-15  

LinkedBolockQueue

14 楼 tianmo2008 2010-03-15  

liwenso 写道

banfry 写道

问下各位，“  public synchronized void produce(Message message) {”
该方法锁定的资源是“message”还是“queue”？

锁定的是这个produce方法

锁对象的,锁方法的话,java的同步就没意义了.

13 楼 超级潜水艇 2010-03-15  

asialee 写道

liwenso 写道

banfry 写道

问下各位，“  public synchronized void produce(Message message) {”
该方法锁定的资源是“message”还是“queue”？

锁定的是这个produce方法

这个应该是锁定的是Queue对象吧，要执行produce方法，必须要拿到Queue对象的锁。

锁对象，有人还没有搞清楚同步的对象是哪个哦！

12 楼 asialee 2010-03-15  

liwenso 写道

banfry 写道

问下各位，“  public synchronized void produce(Message message) {”
该方法锁定的资源是“message”还是“queue”？

锁定的是这个produce方法

这个应该是锁定的是Queue对象吧，要执行produce方法，必须要拿到Queue对象的锁。

11 楼 liwenso 2010-03-15  

banfry 写道

问下各位，“  public synchronized void produce(Message message) {”
该方法锁定的资源是“message”还是“queue”？

锁定的是这个produce方法

10 楼 asialee 2010-03-15  

没有代码高亮，也欢迎大家指正。

9 楼 asialee 2010-03-15  

下面的是我写的：
import java.util.ArrayList;
import java.util.List;


public class ProductQueue {
private List<Product> products;

private int maxSize;

public ProductQueue(int maxSize){
this.maxSize = maxSize;
products = new ArrayList<Product>(maxSize);
}

public synchronized void addProdcut(Product product){
while(isFull()){
try {
wait();
} catch (InterruptedException e) {
e.printStackTrace();
}
}
products.add(product);
System.out.println("Produce:" + product.getId() + " " + product.getMadeDate());
notifyAll();
}

public synchronized void removeProduct(){
while(isEmpty()){
try {
wait();
} catch (InterruptedException e) {
e.printStackTrace();
}
}
Product product = products.get(products.size() - 1);
products.remove(product);
System.out.println("Consume:" + product.getId() + " " + product.getMadeDate());
notifyAll();
}



public synchronized boolean isFull(){
boolean isFull =  products.size() == maxSize;
if(isFull){
System.out.println("The queue is full.");
}
return isFull;
}

public synchronized boolean isEmpty(){
boolean isEmpty = products.size() <= 0;
if(isEmpty){
System.out.println("The queue is empty.");
}
return isEmpty;
}
}



public class Consumer implements Runnable{
private ProductQueue queue;

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

@Override
public void run() {
while(true){
queue.removeProduct();
try {
Thread.sleep(100l);
} catch (InterruptedException e) {
e.printStackTrace();
}
}

}

}


import java.util.Date;


public class Producer implements Runnable{
private static int id;
private ProductQueue queue;

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

@Override
public void run() {
while(true){
Product product = new Product(id++,new Date());
queue.addProdcut(product);
try {
Thread.sleep(100);
} catch (InterruptedException e) {
e.printStackTrace();
}
}

}

}


import java.util.Date;


public class Product {
private int id;
private Date madeDate;

public Product(int id, Date madeDate){
this.id = id;
this.madeDate = madeDate;
}

public int getId() {
return id;
}

public void setId(int id) {
this.id = id;
}

public Date getMadeDate() {
return madeDate;
}

public void setMadeDate(Date madeDate) {
this.madeDate = madeDate;
}

}



public class ThreadMain {
public static void main(String[] args) {
ProductQueue queue = new ProductQueue(50);

for(int i = 0; i < 10; i++){
Producer producer = new Producer(queue);
new Thread(producer).start();
}

for(int i = 0; i < 10; i++){
Consumer consumer = new Consumer(queue);
new Thread(consumer).start();
}
}
}

8 楼 banfry 2010-03-15  

问下各位，“  public synchronized void produce(Message message) {”
该方法锁定的资源是“message”还是“queue”？

7 楼 p_x1984 2010-03-15  

谢谢各位的提醒！

6 楼 lzmhehe 2010-03-15  

看看jdk的BlockQueue
两年前我也自己实现生产消费
后来发现有现成的东西

5 楼 hmilylxs 2010-03-14  

asialee 写道

为什么在consume方法和produce方法开始的时候要调用 this.notifyAll(); 这个应该是生产者在生产完产品后调用通知其他线程，同样消费者在消费完产品后也要调用 this.notifyAll();方法来通知其他线程，为什么一上来调用它呢？

楼上有道理，我认为在consume和produce的结尾分别获得producer和consumer的锁，然后notify它们。在方法一开始唤醒不知道有什么意义。

4 楼 nishizhutou 2010-03-13  

<div class="quote_title">p_x1984 写道</div>
<div class="quote_div">
<p>java多线程一般都会讲消费者-生产者模型<br><br>生产者与消费者模型中，要保证以下几点：<br>1 同一时间内只能有一个生产者生产<br>2 同一时间内只能有一个消费者消费<br>3 生产者生产的同时消费者不能消费<br>4 消息队列满时生产者不能继续生产<br>5 消息队列空时消费者不能继续消费</p>
<p> </p>
<p>---------------------------------------------</p>
<p>除了4和5,其他都不是必要条件.实际上,更多的生产者消费者应用都不遵循1,2,3.<br></p>
</div>
<p> </p>

3 楼 asialee 2010-03-13  

为什么在consume方法和produce方法开始的时候要调用 this.notifyAll(); 这个应该是生产者在生产完产品后调用通知其他线程，同样消费者在消费完产品后也要调用 this.notifyAll();方法来通知其他线程，为什么一上来调用它呢？

2 楼 mercyblitz 2010-03-13  

不错，不过synchronized 关键字范围过大

1 楼 agao1985 2010-03-13  

做开发一年了 没写过一个线程类