Java中的信号量。生产者消费者问题

Question

我正在通过典型的生产者-消费者问题测试信号量的使用，其中我只有一个生产者和一个消费者。生产者一次添加一个产品，消费者可以同时取出几个。

为了执行测试，生产者和消费者从 10 个元素的数组中存储并删除数字，其中 0 表示没有产品，任何其他数字表示产品。对存储和检索项目的访问集中在一个名为 Data 的类中。如果我们有多个线程同时工作，我会使用互斥锁有序地使用向量。

执行时，我观察到权限数根据线程执行的操作是不正确的。应用程序显示错误，因为生产者的信号量说它有权限，但数据向量已满。

package producer.consumer;

import java.io.IOException;

public class ProducerConsumer {

    public static void main(String[] args) throws IOException {
        final int MAX = 10;

        Data data = new Data(MAX);
        Consumer consumer = new Consumer(data);
        Producer producer = new Producer(data);

        consumer.start();
        producer.start();
    }   
}

package producer.consumer;

public class Producer extends Thread{
    private final Data data;

    public Producer(Data data) {
         this.data = data;
    }

    @Override
    public void run() {
        while (true) {
            try {
                data.add((int) (Math.random() * data.getLength()) + 1);
            } catch (InterruptedException ex) {
                System.out.println(ex.getMessage());
            }
        }
    }  
}

package producer.consumer;

import java.util.logging.Level;
import java.util.logging.Logger;

public class Consumer extends Thread{
    private final Data data;

    public Consumer(Data data) {
        this.data = data;
    }

    @Override
    public void run() {
        while (true) {
            try {
                data.remove((int) (Math.random() * data.getLength()) + 1);
            } catch (InterruptedException ex) {
               Logger.getLogger(Consumer.class.getName()).log(Level.SEVERE, null, ex);
            }
        }
    }
}

package producer.consumer;

import java.io.IOException;
import java.util.Arrays;
import java.util.concurrent.Semaphore;

public class Data {
    private final int[] data;
    private final Semaphore mutex = new Semaphore(1);
    private final Semaphore semProducer, semConsumer;

    public Data(int MAX) throws IOException {
        data = new int[MAX];
        semProducer = new Semaphore(MAX);
        semConsumer = new Semaphore(0);
    }

    public int getLength() {
        return data.length;
    }

    public void add(int number) throws InterruptedException {

        semProducer.acquire();

        mutex.acquire();
        System.out.println("trying to add a product");

        int i = 0;
        while (data[i] != 0) {
            i++;
        }
        data[i] = number;

        int permits = semConsumer.availablePermits() + 1;

        System.out.println("data added in " + i + " " + Arrays.toString(data)
            + " Resources consumer " + permits
            + " Resources producer  " + semProducer.availablePermits());
        mutex.release();

        semConsumer.release();
    }

    public void remove(int numberElements) throws InterruptedException {

        semConsumer.acquire(numberElements);

        mutex.acquire();
        System.out.println("trying to withdraw " + numberElements);

        for (int i = 0; i < numberElements; i++) {
            if (data[i] != 0) {
                data[i] = 0;
            }
        }

        int permisos = semProducer.availablePermits() + 1;
        System.out.println(" Retired " + numberElements + " " + Arrays.toString(data)
            + " Resources consumer " + semConsumer.availablePermits()
            + " Resources producer " + permisos);
        mutex.release();

        semProducer.release(numberElements);
    }
}

非常感谢您的帮助。

Answer 1

您的消费者并不总是消费它声称消费的东西。

    for (int i = 0; i < numberElements; i++) {
        if (data[i] != 0) {
            data[i] = 0;
        }
    }

假设 numberElements 为 3，并且我们在 data[7]、data[8]、data[9] 中正好有 3 个可用元素。

循环以 i == 3 结束，没有任何内容被删除，但生产者信号量仍将“增加”3。

在消费者中，如果使用 i 作为数组索引，它需要覆盖整个数组，并且需要一个单独的计数器，用于“删除元素”。

即使生产者首先填充这些数据槽，可用元素也不会总是位于编号最小的数据槽中。考虑生产者设法生产至少 5 个元素的时间序列，然后消费者运行以消耗 2，然后立即再次运行以消耗 3，然后再生产更多元素。 data[0] 和 data[1] 将在消费者第二次运行时为空，我们会遇到我描述的场景。

Answer 2

EDIT 获取和释放许可证似乎是正确的； 但是你需要确保消费者实际上会清除正确数量的元素。

例如，编辑Data 类

public void remove(int numberElements) throws InterruptedException {
    semConsumer.acquire(numberElements);
    mutex.acquire();
    System.out.println("remove: num-elem=" + numberElements);

    int consumed=0;
    for (int i = 0; consumed<numberElements; i++) {
        if (data[i] != 0) {
            data[i] = 0;
            consumed++;
        }
    }
    System.out.println(
            " Retired " + numberElements + " " + Arrays.toString(data)  );
    mutex.release();
    semProducer.release(numberElements);
}

---

还要注意，这种实现效率不是很高（在插入和删除项目时，您需要遍历整个数组，当 MAX 很大时，这可能会很昂贵..）