05
Mar

Producer Consumer with hardcoded stopping signal

Producer Consumer with hardcoded stopping signal
==================================================

The producer-consumer problem (also known as the bounded-buffer problem) is a classic Java Example of a multi-process synchronization problem.

The problem describes two processes, the producer and the consumer, who share a common, fixed-size buffer used as a queue. The producer’s job is to generate a piece of data, put it into the buffer and start again.

At the same time, the consumer is consuming the data (i.e., removing it from the buffer) one piece at a time. The problem is to make sure that the producer won’t try to add data into the buffer if it’s full and that the consumer won’t try to remove data from an empty buffer.


Step 1: Create Consumer which waits till data read matches to hardcoded value before stop

package ProdConsWithStop;

import java.util.concurrent.BlockingQueue;

public class ConsumerPoison implements Runnable {

	private final BlockingQueue<Integer> queue;
	private final Integer POISON;

	@Override
	public void run() {

		try {
			while (true) {
				Integer take = queue.take();
				System.out.println("[Consumer] pulled : "+take);
				// if this is a poison pill, break, exit
				if (take == POISON) {
					break;
				}

			}
		} catch (InterruptedException e) {
			Thread.currentThread().interrupt();
		}

	}


	public ConsumerPoison(BlockingQueue<Integer> queue, Integer POISON) {
	        this.queue = queue;
	        this.POISON = POISON;
	    }
}

Step 2: Create Producer which produces for specific duration and adds Blocker

package ProdConsWithStop;

import java.util.concurrent.BlockingQueue;

public class ProducerPoison implements Runnable {

	private final BlockingQueue<Integer> queue;
	private final Integer POISON;

	public ProducerPoison(BlockingQueue<Integer> queue, Integer POISON) {
		this.queue = queue;
		this.POISON = POISON;
	}

	@Override
	public void run() {
		try {
			try {
				for (int i = 0; i <= 100; i+=20) {
					queue.put(i);
					System.out.println("[Producer] Added : " + i);
				}

			} finally {
				queue.put(POISON);
			}
		} catch (InterruptedException e) {
			e.printStackTrace();
		}

	}

}


Output:
[Producer] Added : 0
[Producer] Added : 20
[Producer] Added : 40
[Producer] Added : 60
[Producer] Added : 80
[Producer] Added : 100
[Consumer] pulled : 0
[Consumer] pulled : 20
[Consumer] pulled : 40
[Consumer] pulled : 60
[Consumer] pulled : 80
[Consumer] pulled : 100
[Consumer] pulled : -1

http://defectracker.com/wp-content/uploads/2020/03/ProdConsWithStop.7z