Producer Consumer problem one producer many consumers java

user_10512012 0
Viewed 301

I'm struggling with Producer Consumer problem. The problem exists when I run one producer and more than one consumers. Only one consumer consumes buffer. As far as I read it can be problem with my buffer implementation. How can I solve this? If it's not a buffer what I do wrong?


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();
    }
}

And here is the console output:

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
...

Thanks!!!

producer-consumer multithreading java
2 Answers

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

For multiple consumers to process the buffer at the same time, you need to release your lock on _buf. In your consumer code you are simulating the working with a sleep. That sleep cannot be inside of the synchronized block:

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

In your producer code you are also sleeping inside of the synchronized block. You should close the lock and then sleep outside and then reenter the synchronized block.

To fix the consumers, you should change your consumer code so that there are 2 synchronized blocks: one to wait for and get the buffer and one to notify other folks that the buffer is available.

Something like:

    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();
        }
    }

If we look at Buffer, anything that looks at it's state needs to be synchronized. That means that size(), isEmpty(), and isFull() need also to be synchronized.

When I run your code with those modifications I see:

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

Couple other comments:

  • Don't catch Exception, catch InterruptedException and make sure you do a Thread.currentThread().interrupt() to re-interrupt the thread after catching.
  • Be careful of using System.out.println(...) in threaded programs because it is a synchronized call which can change the timing of your threads.
  • The Buffer is methods should return boolean and not Boolean.
  • I assume this is an exercise. If you were doing this for real, I'd recommend using a BlockingQueue which takes care of sharing the data between the threads as well as the locking and signaling for you.
user_1798500

The reason why your Consumers can't process the buffer in parallel is synchronized (_buf) section. So while the lock is acquired by single Consumer none of other Consumers can process. As a solution I suggest get rid of locking on Consumer and Producer side in respect of locking inside buffer operations (which is present in your code too). Code can look like this:

Producer part:

    @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();
            }
        }

    }

Consumer part:

    @Override
    public void run() {

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

Buffer part:

     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;
    }
user_41540480
Related