生产者和消费者线程意外工作

Question

主线程

 public static void main(String[] args) {
        List<Integer> taskQueue = new ArrayList<Integer>();
        int MAX_CAPACITY = 5;
        Producer pt=new Producer(taskQueue,MAX_CAPACITY);
        Consumer ct=new Consumer(taskQueue);
        pt.start();
        ct.start();
    }

生产者线程

public class Producer<T> extends Thread {

    private final List<Integer> taskQueue;
    private final int Max_CAPACITY;

    public Producer(List<Integer> taskQueue,int size) {
        this.taskQueue = taskQueue;
        this.Max_CAPACITY=size;
    }

    @Override
    public void run() {
        int count = 0;
        while (true) {
            try {
                produce(count++);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
    }

    public void produce(int i) throws InterruptedException {

        synchronized (taskQueue){
            while(taskQueue.size()==Max_CAPACITY){
                System.out.println("taskQueue is full wait some time for removal");
                taskQueue.wait();
            }

            Thread.sleep(100);
            taskQueue.add(i);

            System.out.println("produced "+i);
            taskQueue.notifyAll();
        }
    }
}

消费者话题

public class Consumer<T> extends Thread {

    private final List<Integer> taskQueue;

    Consumer(List<Integer> taskQueue) {
        this.taskQueue = taskQueue;

    }

    @Override
    public void run() {

        while (true) {
            try {
                consume();
            } catch (InterruptedException ie) {
                ie.printStackTrace();
            }
        }
    }

    private void consume() throws InterruptedException {
        synchronized (taskQueue) {

            while(taskQueue.isEmpty()){
                System.out.println("taskQueue is empty..wait for adding items");
                taskQueue.wait();
            }
            Thread.sleep(100);
           int removal= taskQueue.remove(0);
            System.out.println("consumed.. "+removal);
            taskQueue.notifyAll();


        }
    }
}

根据概念：

在对象上调用Wait()和notify()方法。这里taskqueue是通用对象。 假设Producer 线程首先获得机会并获得taskqueue 对象的锁。在taskqueue 中添加值时，它调用notifyAll()。但是在Consumer 线程上没有调用wait()（因为队列不为空）所以notifyAll() 对消费者没有用处。同样，假设消费者将恢复并从taskqueue 中删除元素并调用notifyAll()，现在producer 线程没有使用消费者notifyAll()，因为生产者也没有调用wait()。那么这个生产者-消费者是如何工作的呢？

简而言之，如果生产者没有调用wait()，那么在消费者线程调用notifyAll()之后，生产者线程如何恢复工作。如果 Producer 中没有调用 wait()，生产者如何获得有关 Consumer 线程的通知？

输出：

produced 0
produced 1
produced 2
produced 3
produced 4
taskQueue is full wait some time for removal
consumed.. 0
consumed.. 1
consumed.. 2
consumed.. 3
consumed.. 4
taskQueue is empty..wait for adding items
produced 5
produced 6
produced 7
produced 8
produced 9
taskQueue is full wait some time for removal
consumed.. 5
consumed.. 6
consumed.. 7
consumed.. 8
consumed.. 9

它会一直这样......

更新：

假设生产者线程首先获得了机会。然后它打印（生产）

produced 0
produced 1
produced 2
produced 3
produced 4

之后，taskequeue 已满，wait() 被调用生产者， 所以现在消费者线程有机会，一旦消费者线程消费元素 consume 0 。 notify() 被调用，所以现在处于等待状态的生产者线程得到通知并恢复。所以现在制片人应该 生产 5 生产 6 个，以此类推。 但消费者线程仍在运行和打印

consume 2
consume 3

这是怎么回事？希望现在很清楚我的意思。 我只想要这个输出的理由。打印效果如何

Answer 1

notifyAll 总是立即返回。 （就此而言，notify 也是如此）。

notifyAll 它将“打破”在同一对象上调用wait 的任何内容，但这并不意味着这些线程会立即继续运行obj.wait() 调用之后的代码。如果没有这样的线程，好吧，那么什么都不会发生。

notify 是同一个东西，只不过它不会破坏所有这些，而是​​破坏任意一个，如果没有，则什么也不会发生。

其他线程没有立即开始运行的原因是它们首先需要重新获取锁（它们在 synchronized(obj) 块中；当等待开始时它们“放弃”该锁，它们需要重新获取- 在等待结束之前获取它。