I recently faced this following question in a Java coding round, needless to say I did not get a callback. So I want to know where I am going wrong in my solution and also what can be done to improve it.
The Problem statement was as follows:
Create a java program, with two components, a supplier and consumer.
Supplier’s responsibility is to provide ‘random integer values’ in ‘random time intervals’ to consumer.
Consumer is responsible to receive data provided by supplier in the same order and add it to a binary tree. When all the data is passed and processed, consumer should print the data of binary tree in visually tree format.
Supplier’s ‘random time interval’ should be dynamically calculated which can be between 1 and 5 second(s). A total of 10 randomly generated integer elements can be passed on to the consumer.
Maximum number of thread instances which can be initialized in the solution is three.
You can ignore the part about adding it to a binary tree and printing it. I am more interested in the multithreaded implementation of producer/consumer.
The Solution that I submitted was as follows:
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.atomic.AtomicInteger;
public class InternalQueue {
private static final int MAX_QUEUE_SIZE = 10;
private final BlockingQueue<Integer> queue = new ArrayBlockingQueue<>(MAX_QUEUE_SIZE);
private final AtomicInteger itemsProduced = new AtomicInteger(0);
private final AtomicInteger itemsConsumed = new AtomicInteger(0);
public boolean put(int n) {
if(itemsProduced.incrementAndGet() > MAX_QUEUE_SIZE)
return false;
System.out.println(Thread.currentThread().getName() + " trying to put " + n);
return queue.offer(n);
}
public Optional<Integer> get() throws InterruptedException {
if(itemsConsumed.incrementAndGet() > MAX_QUEUE_SIZE)
return Optional.empty();
Integer n = queue.take();
System.out.println(Thread.currentThread().getName() + " consumed " + n);
return Optional.of(n);
}
}
I went with a queue data structure where all the elements would be held until a consumer was ready to consume it. (Somewhat how MQ systems operate)
Then the producer and consumers:
import java.util.concurrent.CountDownLatch;
public class Consumer implements Runnable{
private InternalQueue queue;
private BTree bTree;
private CountDownLatch latch;
public Consumer(InternalQueue queue, BTree bTree, CountDownLatch latch) {
this.queue = queue;
this.bTree = bTree;
this.latch = latch;
}
@Override
public void run() {
while (true) {
Optional<Integer> num;
try {
num = queue.get();
} catch (InterruptedException e) {
e.printStackTrace();
latch.countDown();
break;
}
if(num.isEmpty()) {
latch.countDown();
break;
}
int n = num.get();
System.out.println(Thread.currentThread().getName() + " pushing to BTREE " + n);
bTree.insert(n);
}
}
}
public class Producer implements Runnable{
private InternalQueue queue;
private CountDownLatch latch;
public Producer(InternalQueue queue, CountDownLatch latch) {
this.queue = queue;
this.latch = latch;
}
@Override
public void run(){
while (true) {
int num = ThreadLocalRandom.current().nextInt();
boolean added = queue.put(num);
if(!added) {
latch.countDown();
break;
}
int delay = ThreadLocalRandom.current().nextInt(1, 5+1);
try {
Thread.sleep(delay * 1000L);
} catch (InterruptedException e) {
e.printStackTrace();
latch.countDown();
break;
}
}
}
}
Please ignore the B-Tree part.
And lastly the void main(String args[]) of the program:
public class Runner {
private static final int THREAD_POOL_SIZE = 3;
public static void main(String[] args) throws InterruptedException {
ExecutorService service = Executors.newFixedThreadPool(THREAD_POOL_SIZE);
InternalQueue queue = new InternalQueue();
BTree bTree = new BTree();
int num_threads = 3;
CountDownLatch latch = new CountDownLatch(num_threads);
Producer producer1 = new Producer(queue,latch);
Producer producer3 = new Producer(queue,latch);
Consumer consumer1 = new Consumer(queue, bTree,latch);
Consumer consumer2 = new Consumer(queue, bTree,latch);
service.execute(producer1);
service.execute(consumer2);
// service.execute(producer3);
service.execute(consumer1);
latch.await();
bTree.printTree();
service.shutdown();
}
}
So a couple of questions:
- Is this solution thread-safe and correct ?
I used
AtomicIntegerto guard against when the producer has produced over the limit and anotherAtomicIntegerso guard against queue underflow. Is it ok ? Or is there something better or wrong ? - Was it correct to use a
CountDownLatch? Or was it overkill ? BlockingQueueneeded or not ? I used it because I was not sure if any concurrency problems would be there if I hadn't used a thread-safe Collection from the library.
Any feedback or suggestions ?
Thanks.
