线程---生产者消费者问题

    在开始之前先简单回顾一下生产者消费者问题:一群生产者在生产消息，并将此消息提供给消费者去消费。它们中间设了具有N个缓存区的缓冲池，生产者每次可将生产的消息放入一个缓存区内，消费者每次可将一个缓存区内的消息拿出来消费。但这个过程有两个条件：任何一方操作一个缓冲区时不能有其它同时对该缓冲区进行操作；只有当缓冲区还有空余，生产者才能生产，只有当缓冲区至少有一个产品，消费者才能从中取出来消费。这里两个条件分别对应了互斥和同步。



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



正文：

首先先建一个使用多线程的写的生产者消费者的通用类,在生产方法中(produce())，如果队列中已经有五个大饼，就提示消息，等待消费者消费大饼，否则往队列中添加大饼。在消费方法中(consume()),如果队列中没有大饼，则提示消息，等待生产者生产大饼，否则就删除一条大饼消息。

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

public class Stack {   
    List<Message> stack = new ArrayList<Message>();
    private int   MAX_MESSAGE_NUM = 5;   
    public Stack(){   
        System.out.println("大饼生产基地！");    
    }   
       
    //加上互斥锁   
    public synchronized void produce(Message message){   
        while( stack.size() == MAX_MESSAGE_NUM ){   
            try{   
                System.out.println(Thread.currentThread().getName() + " 队列满！等待消费...");   
                this.wait(); 
            }catch(InterruptedException e){   
            }   
        }   
        stack.add(message);
        System.out.println(Thread.currentThread().getName() + " 正在生产" + message.getContent() + " 当前大饼个数:" + stack.size());
        this.notify();    
    }   
       
    //加上互斥锁   
    public synchronized void consume(){   
        while(stack.size() == 0){   
            try{   
                System.out.println("队列空，无大饼，请等待...");   
                this.wait();  
            }catch(InterruptedException e){   
            }   
        }   
        Message message = stack.get(0);
        stack.remove(0);
        System.out.println(Thread.currentThread().getName() + " 正在消费"
       + message.getContent() + " ,当前大饼个数: " + stack.size());
        
        this.notify();    
    }   
       
}   

创建个消息对象。

public class Message {
	public static int id;
	public String content;
	public static int getId() {
		return id;
	}
	public static void setId(int id) {
		Message.id = id;
	}
	public String getContent() {
		return content;
	}
	public void setContent(String content) {
		this.content = content;
	}
	
}

生产者类，主要产生消息，注意继承了Thread并实现了run，并且有一个对Stack实例的引用，主要调用了Stack的produce方法

class Producer extends Thread {
 
 private Stack stack;

 public Producer(Stack s){
  this.stack = s;
 }

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

消费者类，主要消费消息，跟生产者是对称的，主要调用了Stack的consume方法。

class Consumer extends Thread {
 private Stack stack;
 
  Consumer(Stack stack) {
    this.stack = stack;
   }

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

    }
   }
}

主类，主要构造出一个Stack并启动N个生产者和消费者。

public class ThreadTest {

 public static void main(String[] args) {

  Stack s = new Stack();

  Producer p1 = new Producer(s);
  Producer p2 = new Producer(s);
  Producer p3 = new Producer(s);
  
  Consumer c1 = new Consumer(s);
  Consumer c2 = new Consumer(s);
  Consumer c3 = new Consumer(s);
  
  Thread threadP1 = new Thread(p1, "thread-生产者--A");
  Thread threadP2 = new Thread(p2, "thread-生产者--B");
  Thread threadP3 = new Thread(p3, "thread-生产者--C");

  Thread threadC1 = new Thread(c1, "thread-消费者甲");
  Thread threadC2 = new Thread(c2, "thread-消费者乙");
  Thread threadC3 = new Thread(c3, "thread-消费者丙");
  
  threadP1.start();
  threadP2.start();
  threadP3.start();
  
  threadC1.start();
  threadC2.start();
  //threadC3.start();
 }
}

从以上这个例子，可以看出来，通过一个Stack,我们将一个对象的创建和执行分离开了（在生产者中创建，在消费者中执行,这种分离带来了极大的好处。比如将费时的执行分开之后提高了创建线程的响应性，并且它们之间的顺序实现了可控性，不用创建完了马上就执行,它们的分离使得可以分布式做创建和分离，从而出现了类似BMQ的消息中间件。