消费者在 Java 中的简单生产者/消费者/队列代码中不起作用答案

【问题标题】：Consumer doesn't work in my simple producer/consumer/queue code in Java消费者在 Java 中的简单生产者/消费者/队列代码中不起作用
【发布时间】：2018-12-14 20:18:56
【问题描述】：

我正在尝试在 Java 11 中实现一个简单的生产者/消费者系统。基本上，我为每个线程使用两个线程，外加一个全局队列，如下所示：

一个全局优先级队列。
第一个线程生产者运行 HTTP 服务器，监听传入的 http 消息，并在收到消息后，pushes 将其作为队列的作业（queue.size 递增）
第二个线程，消费者，不断peeks 队列。如果有作业 (job ! = null)，则在某处提交 HTTP 请求，并在成功接收后，从队列中轮询它（queue.size() 递减）。

骨架如下：

主类：

public class Manager
{
    private Consumer consumer;
    private Producer producer;
    Queue queue;

    public static void main (String args[])
    {
        consumer = new Consumer();
        producer = new Producer();
    }
}

生产者类：

public class Producer implements Runnable
{
    public Producer()
    {
        Thread producer = new Thread(this);
        producer.start();
    }

    public void run()
    {
            //HTTP server starts, listens, and adds to the queue upon receiving a Job
            server.start();
            Manager.queue.add(new Job());
    }
}

消费类：

public class Consumer implements Runnable
{
    public Consumer()
    {
        Thread consumer = new Thread(this);
        consumer.start();
    }

    public void run()
    {
    // Thread.sleep(1);

        while(true)
        {
            //get an object off the queue
            Job job= Manager.queue.peek();
            //do some stuff with the object
        }
    }
}

Producer 和 queue 可以正常工作。但问题在于Consumer。上面的消费者代码（带有while(true) 循环）不会窥视该项目。但是当我在while(true) 循环之前添加Thread.sleep(x) 时，即使x=1 ms，它也可以工作，并成功抓取项目。

有什么问题？从理论上讲，while(true) 循环应该不是问题！为什么它看不到和peek这个项目？！

【问题讨论】：

你用什么初始化queue？
队列最初是空的。大小只有 2000。
不幸的是，这对我没有帮助。你用什么类来初始化queue？能贴出完整代码吗？
static PriorityQueue<Job> jobSubmissionQueue
PriorityQueue 不是线程安全的。您可以使用PriorityBlockingQueue 代替它，它是线程安全的。

标签： java multithreading queue polling producer-consumer

【解决方案1】：

问题原因：队列读写不同步。

这里发生的情况是，运行在不同 CPU 内核上的两个线程都使用自己的队列副本工作，因此生产者可能正在添加内容，这些更改甚至可能传播到 RAM 中，但消费者从不检查任何内容在 RAM 中，因为它有自己的队列缓存副本，所以女巫保持为空。

Thread.sleep() 起作用了，因为在唤醒时，线程必须从 RAM 中获取所有的东西，而它可能已经改变了。

正确的做法是只访问队列，同步时如下：

在生产者中：

synchronized(Manager.queue) {
     Manager.queue.add(new Job());
}

在消费者中：

boolean continue = true;
while (continue) {
    synchronized(Manager.queue) {
        Job job=Manager.queue.pop();
    }
}

最后一点：while (true) 的效率非常低，你可以使用 Object.wait() 和 Object.notify() 来做一些事情

在生产者中：

synchronized(Manager.queue) {
     Manager.queue.add(new Job());
     Manager.queue.notify();
}

在消费者中：

boolean continue = true;
while (continue) {
    synchronized(Manager.queue) {
        while (Manager.queue.peek() == null) {
            Manager.queue.wait();
        }
        Job job=Manager.queue.pop();
    }
}

【讨论】：

while(true) 真正增加了多少开销？是不是超级差？！
是的，它将一个 cpu 核心固定在 100%，导致您的系统有效地失去一个 CPU 核心并迫使 cpu 全速运行，从而导致不必要的功耗和热量。如果你不想处理Object.notify()，至少做Thread.sleep(10)，让cpu大部分时间空闲。
我这里有问题。第二个poll 永远不会起作用。只有第一个有效。我正在尝试在您的 while 循环中打印一些内容，但 while (Manager.queue.peek() == null) { System.out.println("111111111111111"); 没有为第二次投票打印任何内容。如何解决？
我不知道你说的poll是什么意思。如果您想知道为什么 while 循环从不循环，那是因为 notify() 和 wait() 逻辑应该防止这种情况发生，理论上您可以使用 if 如果您真的想要，但是这个 while 循环捕获任何编码错误和可能干扰的自发唤醒。不要使用if。如果poll 你的意思是pop：也许新的Job 是在消费者寻找新的之前添加的，所以没有等待，while 循环甚至没有执行一次。
是的，看起来逻辑在某个地方失败了。消费者处于循环中，打印（“1111”），等待。同时，制片人补充说，通知。消费者流行音乐。再次生产者添加，通知。但是消费者第二次没有去循环打印 1111。为什么？！

【解决方案2】：

PriorityQueue 不是线程安全的，而PriorityBlockingQueue 是。只要您不使用BlockingQueue 接口中定义的任何方法，这两种实现是可以互换的。只需将 PriorityQueue 更改为 PriorityBlockingQueue 即可解决您的问题。

【讨论】：

while(true) 真正增加了多少开销？是不是超级差？！