java - Java 中的生产者/消费者 - 生产者阻止我的消费者

Question

通常 SO existent 主题可以帮助我解决问题，但现在我发现自己陷入了困境。

我想在 Java 中使用并发实现 Prod/Cons。不使用现有 API，因为是出于学习目的。

我的生产者正在阻止消费者使用队列 (Holder) 中的消息，但我希望生产者和消费者同时使用队列。

您可以运行我的示例，您会看到，当生产者正在添加时，消费者正在等待锁定。但我希望消费者在添加消息后立即完成他的工作，而不是在生产者告诉他时。

令我惊讶的是，我发现搜索 P/C 模式的所有这些示例都与我的一样有效(生产者阻止了消费者，这对我来说没有意义)

import java.util.LinkedList;
import java.util.Queue;
import java.util.Random;
import java.util.concurrent.Executor;
import java.util.concurrent.Executors;

class Holder<T> {

    private int capacity;
    private Queue<T> items = new LinkedList<T>();

    public Holder(int capacity) {
        this.capacity = capacity;
    }

    public synchronized void addItem(T item) throws InterruptedException {
        Thread.sleep(new Random().nextInt(2000));
        while (isFull()) {
            System.out.println("Holder FULL. adding operation is waiting... [" + item + "]");
            this.wait();
        }
        System.out.println(items.size() + "  -- holder +++ added " + item);
        items.add(item);
        this.notifyAll();
    }

    public T getItem() throws InterruptedException {
        synchronized (this) {
            while (isEmpty()) {
                System.out.println("Holder EMPTY. getting operation is waiting...");
                this.wait();
            }
            T next = items.poll();
            System.out.println(items.size() + "  -- holder --- removed " + next + " - remaining: " + items.size());
            this.notifyAll();
            return next;
        }
    }

    private synchronized boolean isEmpty() {
        return items.isEmpty();
    }

    private synchronized boolean isFull() {
        return items.size() >= capacity;
    }

}

class Producer implements Runnable {

    public static final int GENERATED_ITEMS_COUNT = 10;
    private int id;
    private Holder<String> holder;

    public Producer(int id, Holder<String> holder) {
        this.id = id;
        this.holder = holder;
    }

    @Override
    public void run() {
        try {
            for (int i = 0; i < GENERATED_ITEMS_COUNT; i++) {
                String produced = "Message " + i + " from [P" + id + "] " + System.nanoTime();
                holder.addItem(produced);
            }
        } catch (InterruptedException e) {
            e.printStackTrace();

        }
    }
}

class Consumer implements Runnable {

    private Holder<String> holder;

    public Consumer(Holder<String> hodler) {
        this.holder = hodler;
    }

    @Override
    public void run() {
        while (true) {
            try {
                String consumed = holder.getItem();
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
    }
}

public class ConsumerProducerApp {

    public static void main(String[] args) throws InterruptedException {
        Holder<String> coada = new Holder<String>(10);

        Thread consumer = new Thread(new Consumer(coada));
        consumer.start();

        Executor executor = Executors.newCachedThreadPool();
        for (int i = 1; i <= 9; i++) {
            executor.execute(new Producer(i, coada));
        }
    }
}

编辑:所以假设我们从这个等式中排除了 Thread.sleep。如果我有 100000 个生产者，并且每个生产者都生产消息怎么办？他们不是在阻止我的 Consumer 吗？因为 Holder 上的普通锁。不是任何方式，也许是另一种模式让我的消费者单独完成他的工作？据我目前的了解，我的实现是正确的，我可能会尝试实现不可能的事情？

Answer 1

为了线程安全，消费者和生产者不能同时使用队列。但是从队列中添加或删除应该是超快的。在一个现实的例子中，需要时间的是生成项目(例如获取网页)和使用它(例如解析它)。

您的 sleep() 调用应该在同步块(synchronized block)之外:

• 避免在生产者不使用队列时阻塞消费者；
• 避免在生产者未使用队列时阻塞其他生产者。

.

public void addItem(T item) throws InterruptedException {
    // simulating long work, not using the queue
    Thread.sleep(new Random().nextInt(2000));

    // long work done, now use the queue
    synchronized (this) {
        while (isFull()) {
            System.out.println("Holder FULL. adding operation is waiting... [" + item + "]");
            this.wait();
        }
        System.out.println(items.size() + "  -- holder +++ added " + item);
        items.add(item);
        this.notifyAll();
    }
}