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

Question

我正在尝试在 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) 循环)不会查看该项目。但是当我添加 Thread.sleep(x)之前while(true)循环，即使 x=1 ms ，它起作用了，并成功抓取了该项目。

有什么问题吗？理论上，while(true)循环应该不是问题!为什么看不到和peek该项目？!

Answer 1

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

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

Thread.sleep() 是有效的，因为当唤醒时，线程必须从 RAM 中获取所有内容，而 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.通知()

在制作人中:

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();
    }
}