java | 生产者/消费者

这里说一下，生产者-消费者和 保护性暂停 的区别。

• 与保护性暂停不同，不需要产生结果和消费结果一一对应
• 消费队列可以平衡生产和消费的线程资源
• 生产者仅负责产生结果，消费者仅负责处理结果
• 消息队列是有容量限制的，满时不会再加入数据，空时不会再消耗数据
• JDK 中各种堵塞队列，采用该模式
• 异步的意思是，消息产生了，会先放在队列中，可能不会立刻消费

代码如下

package com.redisc;

import lombok.extern.slf4j.Slf4j;

import java.util.*;


@Slf4j(topic = "c.Run")
public class Run {

    public static void main(String[] args) throws Exception {
        MessageQueue queue = new MessageQueue(2);
        for (int i = 0; i < 3; i++) {
            int value = i;
            new Thread(() -> {
                queue.put(new Message(value, "值" + value));
            }, "生产者" + i).start();
        }

        new Thread(() -> {
            try {
                while (true) {
                    Thread.sleep(1000);
                    Message message = queue.take();
                }
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }, "消费者").start();
    }

}

// 消费队列类， java 线程之间的通信
@Slf4j(topic = "c.MessageQueue")
class MessageQueue {
    // 消息队列集合
    private LinkedList<Message> list = new LinkedList<>();
    //队列容量
    private int capcity;

    public MessageQueue(int capcity) {
        this.capcity = capcity;
    }

    public Message take() {
        synchronized (list) {
            // 检查对象是否为空
            while (list.isEmpty()) {
                try {
                    log.debug("队列为空，消费者线程等待");
                    list.wait();
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
            // 从头部获取消息
            Message message = list.removeFirst();
            log.debug("已消费消息,{}", message.toString());
            //考虑这样的情况 因为当队列满的时候，队列会 wait，如果 remove 之后，队列就不满了，所以，需要唤醒队列
            list.notifyAll();
            return message;
        }
    }

    // 存放消息
    public void put(Message message) {
        synchronized (list) {
            //检查队列是否满了
            while (list.size() == capcity) {
                try {
                    log.debug("队列满了，生产者线程等待");
                    list.wait();
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
            log.debug("已生产消息,{}", message.toString());
            // 将消息加入到队列尾部
            list.addLast(message);
            // 唤醒消息
            list.notifyAll();
        }
    }

}

// final 弄成线程安全的
final class Message {
    private int id;
    private Object value;

    public Message(int id, Object value) {
        this.value = value;
        this.id = id;
    }

    public int getId() {
        return id;
    }

    public Object getValue() {
        return value;
    }

    @Override
    public String toString() {
        return "Message{" +
                "id=" + id +
                ",value=" + value +
                "}";
    }
}

输出

17:50:44.184 [生产者0] DEBUG c.MessageQueue - 已生产消息,Message{id=0,value=值0}
17:50:44.187 [生产者2] DEBUG c.MessageQueue - 已生产消息,Message{id=2,value=值2}
17:50:44.187 [生产者1] DEBUG c.MessageQueue - 队列满了，生产者线程等待
17:50:45.185 [消费者] DEBUG c.MessageQueue - 已消费消息,Message{id=0,value=值0}
17:50:45.186 [生产者1] DEBUG c.MessageQueue - 已生产消息,Message{id=1,value=值1}
17:50:46.188 [消费者] DEBUG c.MessageQueue - 已消费消息,Message{id=2,value=值2}
17:50:47.191 [消费者] DEBUG c.MessageQueue - 已消费消息,Message{id=1,value=值1}
17:50:48.192 [消费者] DEBUG c.MessageQueue - 队列为空，消费者线程等待