异步模式之生产者/消费者

定义：

• 与前面的保护性暂停中的GuardObject不同，不需要产生结果和消费结果的线程一一对应
• 消费队列可以用来平衡生产和消费的线程资源
• 生产者仅负责产生结果数据，不关心数据如何处理，而消费者专心处理结果数据
• 消息队列是有容量限制的，满时不会再加入数据，空时不会再消耗数据
• JDK中各种阻塞队列，采用的就是这种模式

• 是一种异步模式，之所以称保护性暂停为同步模式，是因为一个线程产生结果，另一个线程会立刻获得结果。而生产者/消费者模式是异步模式的原因是，生产线程产生结果后会将结果放入消息队列中，消费者需要时才会从中取出结果。

  

实现：

要熟悉这种多线程编程思想，方法上加synchronized，实际上锁的就是调用该方法的对象，这样实现多个线程调用一个共享的锁对象时，可以同步。
这里默认为，当多个线程调用类中的方法时，使用的是该类的同一个实例对象。
对list、this上锁没有区别，因为list是成员变量，一个对象对应一个成员变量，只是粒度的区别。

import java.util.LinkedList;
public class Test21 {
}
class MessageQueue{
    //容器
    private LinkedList<Message> list = new LinkedList<>();
    //容器的容量
    private int capcity;
    public MessageQueue(int capcity){
        this.capcity=capcity;//容量通过构造方法传入
    }
    //获取消息
    public Message take() throws InterruptedException {
        //检查队列是否为空
        synchronized (list){
            while(list.isEmpty()){
                list.wait();
            }
            //从队列的头部获取元素返回
            Message message = list.removeFirst();
            list.notifyAll();
            return message;
        }
    }
    //存入消息
    public void put(Message message) throws InterruptedException {
        synchronized (list){
            //检查队列是否已满
            while(list.size()==capcity){
                list.wait();
            }
            //将新的消息加入队列尾部
            list.addLast(message);
            list.notifyAll();//唤醒线程
        }
    }
}
//消息类，作为消息队列中存储的对象，需要记录对象的id值以及真实值，可以看到该类是线程安全类，因为只涉及到对id值的读，没有写
class Message{
    private int id;
    private Object value;
    public Message(int id, Object value) {
        this.id = id;
        this.value = value;
    }
    public int getId(){
        return id;
    }
    public Object getValue(){
        return value;
    }
    @Override
    public String toString() {
        return super.toString();
    }
}

注意Message类内的方法只是对成员变量作读操作，而没有取操作，所以不存在线程安全问题，类是线程安全类。

测试：

    public static void main(String[] args) throws InterruptedException {
        MessageQueue queue = new MessageQueue(2);//容量为2，接下来多个线程共享一个MessageQueue对象
        for(int i=0;i<3;i++){
            int id=i;
            new Thread(()->{
                try {
                    queue.put(new Message(id,"值"+id));
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            },"生产者"+i).start();
        }
        new Thread(()->{
            while(true){
                try {
                    Thread.sleep(1000);
                    Message message = queue.take();
                    System.out.println("message内容"+message);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
        },"消费者").start();
    }