1

所以我已经阅读了很多关于 Java 中的线程同步的内容。我目前正在尝试有界缓冲区问题。生产者将继续在缓冲区上生产产品，而消费者继续消费它。

如果缓冲区已满，生产者将等待，然后再生产另一个产品。如果缓冲区为空，消费者将等待。

但是，我的问题是生产者仅在缓冲区为空时才开始生产，直到缓冲区已满。消费者只有在缓冲区满时才开始消费，直到它为空。

示例（缓冲区大小：5）

Produced Product 1
Produced Product 2
Produced Product 3
Produced Product 4
Produced Product 5
Consumed Product 1
Consumed Product 2
Consumed Product 3
Consumed Product 4
Consumed Product 5
Produced Product 6
Produced Product 7
Produced Product 8
Produced Product 9
Produced Product 10
Consumed Product 6
Consumed Product 7
Consumed Product 8
Consumed Product 9
Consumed Product 10
Produced Product 11
Produced Product 12
Produced Product 13
Produced Product 14
Produced Product 15
Consumed Product 11
Consumed Product 12
Consumed Product 13
Consumed Product 14
Consumed Product 15
Produced Product 16
Produced Product 17
Produced Product 18
Produced Product 19
Produced Product 20
Consumed Product 16
Consumed Product 17
Consumed Product 18
Consumed Product 19
Consumed Product 20
Produced Product 21
Produced Product 22
Produced Product 23
Produced Product 24
Produced Product 25
Consumed Product 21
Consumed Product 22
Consumed Product 23
Consumed Product 24
Consumed Product 25
Produced Product 26
Produced Product 27
Produced Product 28
Produced Product 29
Produced Product 30
Consumed Product 26
Consumed Product 27
Consumed Product 28
Consumed Product 29
Consumed Product 30

我想要它以便生产者在缓冲区未满时生产，无论缓冲区是否为空。随后，我希望它让消费者消费只要缓冲区不为空，无论缓冲区是否已满。

我的代码有什么问题？

import java.util.LinkedList;
import java.util.Queue;
import java.util.Random;
import javax.swing.JOptionPane;

public class ProducerConsumer{

    final static Queue<Product> buffer = new LinkedList<>();
    private static int buffer_size, no_items, itemno = 1;
    private static Random r = new Random();
    public static void main(String[] args) {
        try{
            buffer_size = Integer.parseInt(JOptionPane.showInputDialog("Input Buffer size (default: 5)"));
            if(buffer_size<0){
                buffer_size = 5;
            }
        }catch(NumberFormatException nfe){
            buffer_size = 5;
        }
        try{
            no_items = Integer.parseInt(JOptionPane.showInputDialog("Input No. of Items (default: 10)"));
            if(no_items<0){
                no_items = 10;
            }
        }catch(NumberFormatException nfe){
            no_items = 10;
        }
        Producer producer = new Producer();
        Consumer consumer = new Consumer();
        producer.start();
        consumer.start();
    }

    static class Product {
        private String name = "Product X";
        private int productno;

        public Product(int productno) {
            this.productno = productno;
            this.name = "Product "+productno;
        }
        public int number() {
            return productno;
        }
        @Override
        public String toString() {
            return name;
        }
    }
    static class Producer extends Thread{
        public Producer(){
        }
        public void produce(){
            Product p = new Product(itemno++);
            try {
                this.sleep(r.nextInt(100));
            } catch (InterruptedException ex) {
                Thread.interrupted();
            }
            buffer.add(p);
            System.out.println("Produced "+p);
        }
        @Override
        public void run(){
            synchronized(buffer){
                while(itemno<=no_items){
                    while(buffer.size()==buffer_size){
                        try{
                            buffer.wait(100);
                        }catch(InterruptedException e){
                            Thread.interrupted();
                        }
                    }
                    produce();
                    buffer.notifyAll();
                }
            }
        }
    }
    static class Consumer extends Thread{
        public Consumer(){
        }
        public boolean consume(){
            try {
                this.sleep(r.nextInt(100));
            } catch (InterruptedException ex) {
                Thread.interrupted();
            }
            Product product = buffer.remove();
            System.out.println("Consumed "+product);
            return product.number()==no_items;
        }
        @Override
        public void run(){
            synchronized(buffer){
                boolean end = false;
                while(!end){
                    while(buffer.isEmpty()){
                        try{
                            buffer.wait(100);
                        }catch(InterruptedException e){
                            Thread.interrupted();
                        }
                    }
                    end = consume();
                    buffer.notifyAll();
                }
            }
        }
    }
}
4

2 回答 2

4

这是因为您buffer在 notifyAll 之前已经获取了监视器并生成了 N 次，并通过退出synchronized块释放了缓冲区监视器，所有这些都在 while 循环内。

尝试把块放在while外面synchronized，这样你就可以给另一个线程在并发生产/消费期间获取锁的机会。

请注意，在通知线程退出该区域notifyAll之前，它实际上不会影响其他线程synchronized

编辑：我在进行建议的更改后运行它，得到以下输出

Produced Product 1
Produced Product 2
Produced Product 3
Consumed Product 1
Consumed Product 2
Consumed Product 3
Produced Product 4
Produced Product 5
Consumed Product 4
Consumed Product 5
Produced Product 6
Consumed Product 6
Produced Product 7
Produced Product 8
Produced Product 9
Produced Product 10
Consumed Product 7
Consumed Product 8
Consumed Product 9
Consumed Product 10
于 2012-04-19T21:52:14.137 回答
0

由于两个线程都在缓冲区上同步，因此任何时候只有一个线程会运行。一个线程在调用 buffer.wait() 时向另一个发出信号以恢复运行并暂停自身。包含 buffer.wait() 调用 (buffer.size() == buffer_size) 和 (buffer.isEmpty()) 的 while 循环中的逻辑允许循环仅在其功能正常时运行（生产者或消费者）已完成，无法继续。

程序是这样执行的：生产者将产生 5 个项目，然后 while 循环条件为真，它会调用 buffer.wait() 以允许消费者运行。消费者将开始运行并消费，直到缓冲区为空，在这种情况下，while 条件为真，它将调用 buffer.wait() 并允许生产者再次运行。

要允许两个线程同时运行，这可能是您真正想要的，您应该仅在需要安全访问缓冲区时调用 synchronize(buffer)。一种方法是使用以下代码块：

static class Producer extends Thread {
    public Producer() {
    }

    public void produce() {
        Product p = new Product(itemno++);
        try {
            this.sleep(r.nextInt(100));
        } catch (InterruptedException ex) {
            Thread.interrupted();
        }
        synchronized(buffer) {
            buffer.add(p);
        }
        System.out.println("Produced " + p);
    }

    @Override
    public void run() {
        while (itemno <= no_items) {
            while (buffer.size() == buffer_size) {
                try {
                    sleep(100);
                } catch (InterruptedException e) {
                    Thread.interrupted();
                }
            }

            produce();
        }
    }
}

static class Consumer extends Thread {
    public Consumer() {
    }

    public boolean consume() {
        try {
            this.sleep(r.nextInt(100));
        } catch (InterruptedException ex) {
            Thread.interrupted();
        }

        Product product;
        synchronized (buffer) {
            product = buffer.remove();
        }

        System.out.println("Consumed " + product);
        return product.number() == no_items;
    }

    @Override
    public void run() {
        boolean end = false;
        while (!end) {
            while (buffer.isEmpty()) {
                try {
                    sleep(100);
                } catch (InterruptedException ex) {
                    Logger.getLogger(TrashMe.class.getName()).log(Level.SEVERE, null, ex);
                }
            }
            end = consume();
        }
    }
}
于 2012-04-19T22:11:03.963 回答