Producer-Consumer Design Pattern in Java

Consumer.java

package com.demo.pc;

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

public class Consumer implements Runnable {

	private final BlockingQueue<Integer> sharedQueue;

	public Consumer(BlockingQueue<Integer> sharedQueue) {
		this.sharedQueue = sharedQueue;
	}

	@Override
	public void run() {
		
		while(true) {
			try {
				System.out.println("Consumer: " + this.toString() + " " + sharedQueue.take());
			} catch (InterruptedException e) {
				Logger.getLogger(Consumer.class.getName()).log(Level.SEVERE, null, e);
			}
		}
	}	
}

Notes

http://javarevisited.blogspot.com/2012/02/producer-consumer-design-pattern-with.html

http://javarevisited.blogspot.com/2015/07/how-to-use-wait-notify-and-notifyall-in.html

http://javarevisited.blogspot.com/2011/05/wait-notify-and-notifyall-in-java.html

http://javarevisited.blogspot.com/2012/02/why-wait-notify-and-notifyall-is.html

Write to File
https://www.mkyong.com/java/how-to-write-to-file-in-java-bufferedwriter-example/

Formatting String
https://www.dotnetperls.com/format-java

Producer.java

package com.demo.pc;

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

public class Producer implements Runnable {

	private final Logger logger = Logger.getLogger(Producer.class.getName());
	
	private final BlockingQueue<Integer> sharedQueue;
	
	public Producer(BlockingQueue<Integer> sharedQueue) {
		super();
		this.sharedQueue = sharedQueue;
	}

	@Override
	public void run() {
		
		for(int i = 0; i < 10; i++) {
			try {
				System.out.println("Produced: " + this.toString() + " " + i);
				sharedQueue.put(i);
			} catch (InterruptedException e) {
				logger.log(Level.SEVERE, e.getMessage());
			}
		}
	}
}

ProducerConsumerBlockingQueue.java

package com.demo.pc;

import java.util.concurrent.BlockingQueue;
import java.util.concurrent.LinkedBlockingQueue;

public class ProducerConsumerBlockingQueue {

	public static void main(String[] args) {
		
		//Create shared object
		BlockingQueue<Integer> sharedQueue = new LinkedBlockingQueue<Integer>();
		
		Producer producer = new Producer(sharedQueue);
		Consumer consumer1 = new Consumer(sharedQueue);
		Consumer consumer2 = new Consumer(sharedQueue);
		
		new Thread(producer).start();
		new Thread(consumer1).start();
		new Thread(consumer2).start();
	}
	
}

ProducerConsumerInJava.java

package com.demo.pc.waitNotify;

import java.util.LinkedList;
import java.util.Queue;
import java.util.Random;

public class ProducerConsumerInJava {

	public static final int MAX_SIZE = 10;
	
	public static void main(String[] args) {
		
		Queue<Integer> buffer = new LinkedList<>();
		
		Thread producer = new Producer(buffer, "PRODUCER");
		Thread consumer = new Consumer(buffer, "CONSUMER");
		
		producer.start();
		consumer.start();
	}
}

class Producer extends Thread {

	private Queue<Integer> queue;
	
	public Producer(Queue<Integer> queue, String name) {
		super(name);
		this.queue = queue;
	}

	@Override
	public void run() {
		while (true) {
			synchronized(queue) {
				while(queue.size() == ProducerConsumerInJava.MAX_SIZE) {
					try {
						System.out .println("Queue is full, " + "Producer thread waiting for " + "consumer to take something from queue"); 
						queue.wait();
					} catch (InterruptedException e) {
						e.printStackTrace();
					}
				}
			
				Random random = new Random();
				int i = random.nextInt();
				System.out.println("Producing value: " + i);
				queue.add(i);
				queue.notifyAll();
			}
		}
	}
}

class Consumer extends Thread {
	private Queue<Integer> queue;
	
	public Consumer(Queue<Integer> queue, String name) {
		super(name);
		this.queue = queue;
	}

	@Override
	public void run() {
		while(true) {
			synchronized(queue) {
				while(queue.isEmpty()) {
					System.out.println("Queue is empty," + "Consumer thread is waiting" + " for producer thread to put something in queue");
					try {
						queue.wait();
					} catch (InterruptedException e) {
						e.printStackTrace();
					}
				}
				
				System.out.println("Consuming value: " + queue.remove());
				queue.notifyAll();
			}
		}
	}
}

ProducerConsumerWithMessage.java

package com.ul.exercise;

import java.io.BufferedWriter;
import java.io.FileWriter;
import java.io.IOException;
import java.util.Date;
import java.util.PriorityQueue;
import java.util.Random;
import java.util.UUID;

class Message {
	
	String text;
	int priority;
	
	public Message(String text, int priority) {
		this.text = text;
		this.priority = priority;
	}

	public String getText() {
		return text;
	}

	public void setText(String text) {
		this.text = text;
	}

	public int getPriority() {
		return priority;
	}

	public void setPriority(int priority) {
		this.priority = priority;
	}
}

public class ProducerConsumerWithMessage {

	public static final int MAX_SIZE = 5;
	
	public static void main(String[] args) {
		
		PriorityQueue<Message> buffer = new PriorityQueue<Message>(MAX_SIZE, (a,b) -> b.getPriority() - a.getPriority());
		
		Thread producer = new Producer(buffer, "PRODUCER");
		Thread consumer = new Consumer(buffer, "CONSUMER");
		
		producer.start();
		consumer.start();
	}
}

class Producer extends Thread {

	private PriorityQueue<Message> queue;
	
	public Producer(PriorityQueue<Message> queue, String name) {
		super(name);
		this.queue = queue;
	}

	@Override
	public void run() {
		while (true) {
			synchronized(queue) {
				while(queue.size() == ProducerConsumerWithMessage.MAX_SIZE) {
					try {
						System.out.println("PriorityQueue is full, " + "Producer thread waiting for " + "consumer to take something from queue"); 
						queue.wait();
					} catch (InterruptedException e) {
						e.printStackTrace();
					}
				}
			
				Random random = new Random();
				int priority = random.nextInt(5);
				String text = UUID.randomUUID().toString();
				Message message = new Message(text, priority);
				
				//System.out.println("Producing value: " + message.getText() + " with priority " + message.getPriority());
				System.out.format("Producing value with message %s and with priority of %d at %s %n", message.getText(), message.getPriority(), new Date());
				queue.add(message);
				queue.notifyAll();
			}
		}
	}
}

class Consumer extends Thread {
	
	//private static final String FILE_NAME = "C:\\temp\\messageoutput.txt";
	private static final String FILE_NAME = "/Users/biniamasnake/Desktop/Java.txt";
	private PriorityQueue<Message> queue;
	
	public Consumer(PriorityQueue<Message> queue, String name) {
		super(name);
		this.queue = queue;
	}

	@Override
	public void run() {
		while(true) {
			synchronized(queue) {
				while(queue.isEmpty()) {
					System.out.println("PriorityQueue is empty," + "Consumer thread is waiting" + " for producer thread to put something in queue");
					try {
						queue.wait();
					} catch (InterruptedException e) {
						e.printStackTrace();
					}
				}
				Message message = queue.remove();
				
				String formattedOutput = String.format("Consuming value with message %s and with priority of %d at %s %n", message.getText(), message.getPriority(), new Date());

				// To STDOUT
				System.out.format("Consuming value with message %s and with priority of %d at %s %n", message.getText(), message.getPriority(), new Date());
				// System.out.print(formattedOutput);
				
				// To File System
				try(BufferedWriter bw = new BufferedWriter(new FileWriter(FILE_NAME, true))) {
					
					bw.write(formattedOutput);
					
				} catch (IOException e) {
					e.printStackTrace();
				} 
				
				queue.notifyAll();
			}
		}
	}
}