生产者消费者问题一个生产者多个消费者java

Question

我正在为生产者消费者问题苦苦挣扎。当我运行一个生产者和多个消费者时存在问题。只有一个消费者消费缓冲区。据我所知，我的缓冲区实现可能存在问题。我该如何解决这个问题？如果它不是缓冲区，我做错了什么？


class Producer extends Thread {
    private final Buffer _buf;
    private final int maxSize;
    private final String name;

    public Producer(Buffer _buf, int maxSize, String name) {
        super(name);
        this._buf = _buf;
        this.maxSize = maxSize;
        this.name = name;
    }

    @Override
    public void run() {
        synchronized (_buf) {
            for (; ; ) {
                while (_buf.isFull()) {
                    try {
                        System.out.println("Buffer is full, " + "Producer thread waiting for " + "consumer to take something from buffer");
                        _buf.wait();
                    } catch (Exception ex) {
                        ex.printStackTrace();
                    }
                }
                Random random = new Random();
                int i = random.nextInt();
                System.out.println(this.name + " producing value " + i);
                _buf.put(i);
                _buf.notifyAll();
                try {
                    Thread.sleep(200);
                } catch (InterruptedException exception) {
                    exception.printStackTrace();
                }
            }
        }
    }
}

class Consumer extends Thread {
    private final Buffer _buf;
    private final int maxSize;
    private final String name;


    public Consumer(Buffer _buf, int maxSize, String name) {
        super(name);
        this._buf = _buf;
        this.maxSize = maxSize;
        this.name = name;
    }

    @Override
    public void run() {
        synchronized (_buf) {
            for (; ; ) {
                while (_buf.isEmpty()) {
                    System.out.println("Buffer is empty," + "Consumer thread is waiting" + " for producer thread to put something in buffer");
                    try {
                        _buf.wait();
                    } catch (Exception ex) {
                        ex.printStackTrace();
                    }
                }

                System.out.println(this.name + ": consuming value " + _buf.get());
                try {
                    Thread.sleep(1000);
                } catch (InterruptedException exception) {
                    exception.printStackTrace();
                }
                _buf.notifyAll();
            }
        }
    }
}


class Buffer {
    public synchronized void put(int i) {
        // check for queue overflow
        if (isFull()) {
            System.out.println("Overflow\nProgram Terminated");
            System.exit(1);
        }

        System.out.println("Inserting " + i);

        rear = (rear + 1) % capacity;
        arr[rear] = i;
        count++;
        notifyAll();

    }

    public synchronized int get() {
        if (isEmpty()) {
            System.out.println("Underflow\nProgram Terminated");
            System.exit(1);
        }
        int result = arr[front];

        System.out.println("Removing " + arr[front]);

        front = (front + 1) % capacity;
        count--;
        notifyAll();
        return result;
    }


    private final int[] arr;      // array to store queue elements
    private int front;      // front points to the front element in the queue
    private int rear;       // rear points to the last element in the queue
    private final int capacity;   // maximum capacity of the queue
    private int count;      // current size of the queue

    // Constructor to initialize a buffer queue
    Buffer(int size) {
        arr = new int[size];
        capacity = size;
        front = 0;
        rear = -1;
        count = 0;
    }

    public int size() {
        return count;
    }

    public Boolean isEmpty() {
        return (size() == 0);
    }

    public Boolean isFull() {
        return (size() == capacity);
    }
}

public class PKmain {

    public static void main(String[] args) {
        int maxSize = 100;
        Buffer buffer = new Buffer(10);

        Thread producer = new Producer(buffer, maxSize, "PRODUCER");
        Thread consumer1 = new Consumer(buffer, maxSize, "CONSUMER 1");
        Thread consumer2 = new Consumer(buffer, maxSize, "CONSUMER 2");
        Thread consumer3 = new Consumer(buffer, maxSize, "CONSUMER 3");
        Thread consumer4 = new Consumer(buffer, maxSize, "CONSUMER 4");
        Thread consumer5 = new Consumer(buffer, maxSize, "CONSUMER 5");
        Thread consumer6 = new Consumer(buffer, maxSize, "CONSUMER 6");

        producer.start();
        consumer1.start();
        consumer2.start();
        consumer3.start();
        consumer4.start();
        consumer5.start();
        consumer6.start();
    }
}

控制台输出如下：

Inserting -1893944
PRODUCER producing value 1150242252
Inserting 1150242252
PRODUCER producing value 957139043
Inserting 957139043
PRODUCER producing value -806406909
Inserting -806406909
PRODUCER producing value 1701947892
Inserting 1701947892
PRODUCER producing value -174867893
Inserting -174867893
PRODUCER producing value 1272708996
Inserting 1272708996
PRODUCER producing value -1522880833
Inserting -1522880833
PRODUCER producing value -1041643777
Inserting -1041643777
PRODUCER producing value 1741137093
Inserting 1741137093
Buffer is full, Producer thread waiting for consumer to take something from buffer
Removing -1893944
CONSUMER 6: consuming value -1893944
Removing 1150242252
CONSUMER 6: consuming value 1150242252
Removing 957139043
CONSUMER 6: consuming value 957139043
Removing -806406909
CONSUMER 6: consuming value -806406909
Removing 1701947892
CONSUMER 6: consuming value 1701947892
Removing -174867893
CONSUMER 6: consuming value -174867893
Removing 1272708996
CONSUMER 6: consuming value 1272708996
Removing -1522880833
CONSUMER 6: consuming value -1522880833
Removing -1041643777
CONSUMER 6: consuming value -1041643777
Removing 1741137093
CONSUMER 6: consuming value 1741137093
Buffer is empty,Consumer thread is waiting for producer thread to put something in buffer
Buffer is empty,Consumer thread is waiting for producer thread to put something in buffer
Buffer is empty,Consumer thread is waiting for producer thread to put something in buffer
Buffer is empty,Consumer thread is waiting for producer thread to put something in buffer
Buffer is empty,Consumer thread is waiting for producer thread to put something in buffer
Buffer is empty,Consumer thread is waiting for producer thread to put something in buffer
PRODUCER producing value -1656771306
Inserting -1656771306
PRODUCER producing value 146381233
Inserting 146381233
PRODUCER producing value -303301670
Inserting -303301670
...

谢谢！！！

Answer 1

您的消费者无法并行处理缓冲区的原因是 synchronized (_buf) 部分。所以当锁被单个Consumernone获取的时候，其他Consumer可以处理。作为一种解决方案，我建议在锁定内部缓冲区操作（这也存在于您的代码中）方面摆脱消费者和生产者方面的锁定。代码可能如下所示：

制作人部分：

    @Override
    public void run() {
        for (; ; ) {

            Random random = new Random();
            int i = random.nextInt();
            System.out.println(this.name + " producing value " + i);
            _buf.put(i);
            try {
                Thread.sleep(200);
            } catch (InterruptedException exception) {
                exception.printStackTrace();
            }
        }

    }

消费者部分：

    @Override
    public void run() {

        for (; ; ) {
            System.out.println(this.name + ": consuming value " + _buf.get());
            try {
                Thread.sleep(1000);
            } catch (InterruptedException exception) {
                exception.printStackTrace();
            }
        }
    }

缓冲部分：

     public synchronized void put(int i) {
        while (isFull()) {
            try {
                wait();
            } catch (Exception ex) {
                ex.printStackTrace();
            }
        }

        rear = (rear + 1) % capacity;
        arr[rear] = i;
        count++;
        notifyAll();
    }

    public synchronized int get() {
        while (isEmpty()) {
            try {
                wait();
            } catch (Exception ex) {
                ex.printStackTrace();
            }
        }
        int result = arr[front];

        front = (front + 1) % capacity;
        count--;
        notifyAll();
        return result;
    }

Answer 2

Only one consumer consumes buffer. As far as I read it can be problem with my buffer implementation. How can I solve this?

对于多个消费者同时处理缓冲区，您需要释放对_buf的锁定。在您的消费者代码中，您正在模拟睡眠时的工作。睡眠不能在 synchronized 块内：

   synchronized (_buf) {
        ...
                // sleeping in a synchronized block is an anti-pattern
                Thread.sleep(1000);
        ...
    }

在您的生产者代码中，您也在 synchronized 块内休眠。你应该关上锁，然后睡在外面，然后重新进入 synchronized 街区。

要修复消费者，您应该更改消费者代码，以便有 2 个 synchronized 块：一个等待并获取缓冲区，一个通知其他人缓冲区可用。

类似于：

    for (; ; ) {
        int contents;
        synchronized (_buf) {
            while (_buf.isEmpty()) {
                try {
                    _buf.wait();
                } catch (InterruptedException ex) {
                    ex.printStackTrace();
                }
            }
            bufferContents = buf.get();
        }
        // now that we have the contents, we can work on them OUTSIDE of the lock
        System.out.println(this.name + ": consuming value " + contents);
        try {
            Thread.sleep(1000);
        } catch (InterruptedException exception) {
            exception.printStackTrace();
        }
        synchronized (_buf) {
           _buf.notifyAll();
        }
    }

如果我们查看 Buffer，任何查看其状态的内容都必须是 synchronized。这意味着 size()、isEmpty() 和 isFull() 也需要是 synchronized.

当我运行你的代码修改后我看到：

PRODUCER producing value 1329586226
Inserting 1329586226
Removing 1329586226
Buffer is empty,Consumer thread is waiting for producer thread to put something in buffer
CONSUMER 6: consuming value 1329586226
Buffer is empty,Consumer thread is waiting for producer thread to put something in buffer
Buffer is empty,Consumer thread is waiting for producer thread to put something in buffer
Buffer is empty,Consumer thread is waiting for producer thread to put something in buffer
Buffer is empty,Consumer thread is waiting for producer thread to put something in buffer
PRODUCER producing value 176292473
Inserting 176292473
Removing 176292473
CONSUMER 1: consuming value 176292473
Buffer is empty,Consumer thread is waiting for producer thread to put something in buffer
Buffer is empty,Consumer thread is waiting for producer thread to put something in buffer
Buffer is empty,Consumer thread is waiting for producer thread to put something in buffer
Buffer is empty,Consumer thread is waiting for producer thread to put something in buffer
PRODUCER producing value -1900116049
Inserting -1900116049
Removing -1900116049
CONSUMER 5: consuming value -1900116049
Buffer is empty,Consumer thread is waiting for producer thread to put something in buffer
Buffer is empty,Consumer thread is waiting for producer thread to put something in buffer
Buffer is empty,Consumer thread is waiting for producer thread to put something in buffer
PRODUCER producing value 1933484634
Inserting 1933484634
Removing 1933484634
Buffer is empty,Consumer thread is waiting for producer thread to put something in buffer
CONSUMER 2: consuming value 1933484634
Buffer is empty,Consumer thread is waiting for producer thread to put something in buffer
PRODUCER producing value 1398832650
Inserting 1398832650
Removing 1398832650

其他一些评论：

不要捕获 Exception，捕获 InterruptedException 并确保在捕获后执行 Thread.currentThread().interrupt() 以重新中断线程。
在线程程序中使用 System.out.println(...) 时要小心，因为它是一个同步调用，可以改变线程的时间。
Buffer 方法应该 return boolean 而不是 Boolean.
我认为这是一个练习。如果您真的这样做，我建议您使用 BlockingQueue 来负责在线程之间共享数据以及为您锁定和发信号。

生产者消费者问题一个生产者多个消费者java

Producer Consumer problem one producer many consumers java

java

multithreading

producer-consumer