Java,J2ee,Hybris E-commerce suit,Apache Ofbiz ,database, E-commerce domain Experts: Thread

Showing posts with label Thread. Show all posts

Producer-Consumer Problem in Java | Java Thread interview question for Experinced developer

The producer-consumer problem is one of the most frequently encountered problems when we attempt multi threaded programming. While not as challenging as some of the other problems in multi-threaded programming, an incorrect implementation of this problem can create a mess of your application. Produced items will be left unconsumed, starting items will be skipped, consumption depends on whether the production began earlier or later than consumption attempts etc. To add to this you might notice the anomalies long after it has actually happened and most importantly like almost all multi-threaded programs, this one is hard to debug and reproduce too.

So in this post I thought I would attempt to solve this problem in Java with the help of Java' awesome java.util.concurrent package and its classes.

First of all, let us see the characteristics of the producer-consumer problem:

Producer(s) produce items.
Consumer(s) consume the items produced by the producer(s).
Producer(s) finish production and let the consumers know that they are done.

Note that in this producer-consumer problem the producer is running on a different thread than the ones on consumer. This setup makes sense in two cases:

The steps to do the consumption of the item produced in independent and not dependent on other items.
The time to process the items is larger that the time to produce them.

The term "larger" in the second point is used a bit loosely. Consider the case where producer reads a line from a file and the "consumption and processing" is just to log the line in a special format back to a file then the use of a producer consumer problem solution can be considered a case of over-engineering a solution. However if for each of those lines the "consumption and processing" step is to make a HTTP GET/POST request to a web-server and then dump the result somewhere then we should opt for a producer-consumer solution. In this case I am assuming that all the data to do a GET/POST is available in the line (item) itself and we are not dependent on previous/next lines.

So let us first take a look at the characteristics of the producer-consumer problem solution that I have posted below:

There can be multiple producer.
There will be multiple consumers.
Once the production of new items is done the producer(s) will let the consumers know so that the consumer will exit after the last item is consumed and processed.

It is interesting to note that to solve this problem at a generic level we can address only the consumer side and not the producer side. This is because the production of items might be done at any time and there is very little that we can do in a generic way to control the production of items. We can, however control the consumer's behaviour while accepting items from producer(s). Having laid out the rules let us take a look at the consumer contract:

package com.maximus.producerconsumer;
public interface Consumer
{
public boolean consume(Item j);
public void finishConsumption();
}

Here the consumer can be shared between multiple producers of similar items; by similar items I mean producer that produces objects of type "Item". The definition if Item is as follows:

package com.maximus.consumer;
public interface Item
{
public void process();
}

Now we take a look at an implementation of the Consumer interface:

package com.maximus.consumer;
import java.util.LinkedList;
import java.util.List;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.LinkedBlockingQueue;
public class ConsumerImpl implements Consumer
{
private BlockingQueue< Item > itemQueue =
new LinkedBlockingQueue< Item >();
private ExecutorService executorService =
Executors.newCachedThreadPool();
private List< ItemProcessor > jobList =
new LinkedList< ItemProcessor >();
private volatile boolean shutdownCalled = false;
public ConsumerImpl(int poolSize)
{
for(int i = 0; i < poolSize; i++)
{
ItemProcessor jobThread =
new ItemProcessor(itemQueue);
jobList.add(jobThread);
executorService.submit(jobThread);
}
}
public boolean consume(Item j)
{
if(!shutdownCalled)
{
try
{
itemQueue.put(j);
}
catch(InterruptedException ie)
{
Thread.currentThread().interrupt();
return false;
}
return true;
}
else
{
return false;
}
}
public void finishConsumption()
{
shutdownCalled = true;
for(ItemProcessor j : jobList)
{
j.cancelExecution();
}
executorService.shutdown();
}
}

Now the only point of interest is the ItemProcessor that the consumer internally uses to process the incoming items. ItemProcessor is coded as follows:

package com.maximus.consumer;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.TimeUnit;
public class ItemProcessor implements Runnable
{
private BlockingQueue< Item> jobQueue;
private volatile boolean keepProcessing;
public ItemProcessor(BlockingQueue< Item > queue)
{
jobQueue = queue;
keepProcessing = true;
}
public void run()
{
while(keepProcessing || !jobQueue.isEmpty())
{
try
{
Item j = jobQueue.poll(10, TimeUnit.SECONDS);
if(j != null)
{
j.process();
}
}
catch(InterruptedException ie)
{
Thread.currentThread().interrupt();
return;
}
}
}
public void cancelExecution()
{
this.keepProcessing = false;
}
}

The only challenge above is the condition in the while loop. The while loop is so written to support the continuation of the consumption of items even after the producer(s) have finished production and has notified the consumer that production is finished. The above while loop ensures that consumption of all the items is done before the threads exit.This will be the case when producers run faster that consumers.

The above consumer is thread-safe and can be shared multiple producers such that each producer may concurrently call consumer.consume() without bothering about synchronization and other multi-threading caveats. Producers just need to submit an implementation of the Item interface whose process() method will contain the logic of how the consumption will be done.

As a bonus for reading the post I put forward a test program that demonstrates how to use the above classes:

package com.maximus.consumer;
import java.io.BufferedReader;
import java.io.File;
import java.io.FileInputStream;
import java.io.InputStreamReader;
public class Test
{
public static void main(String[] args) throws Exception
{
Consumer consumer = new ConsumerImpl(10);
BufferedReader br =
new BufferedReader(
new InputStreamReader(
new FileInputStream(
new File(args[0]))));
String line = "";
while((line = br.readLine()) != null)
{
System.out.println(
"Producer producing: " + line);
consumer.consume(new PrintJob(line));
}
consumer.finishConsumption();
}
}
class PrintJob implements Item
{
private String line;
public PrintJob(String s)
{
line = s;
}
public void process()
{
System.out.println(
Thread.currentThread().getName() +
" consuming :" + line);
}
}

The above consumer can be tweaked in a host of different ways to make it more flexible. We can define what the consumer will do when production is done. It may be tweaked to allow batch processing but I leave that to the user. Feel free to use it and twist it in whatever way you want.

Thread.State in Java? BLOCKED vs WAITING | What is Thread.State in Java? What's it used for?

Thread.State - This is a static nested class (Read more about nested classes in the article - Nested Classes & Inner Classes in Java >>) of the Thread class. This is one of the additions of Java 5 and this class actually inherits the abstract class Enum which is the common base class of all Java language enumeration types i.e., Thread.State is actually is actually an enumeration type.

Thread.State enumeration contains the possible states of a Java thread in the underlying JVM. These states are different from the Operating System thread states. The possible values of the Thread.State are:-

    NEW - this state represents a new thread which is not yet started.
    RUNNABLE - this state represents a thread which is executing in the underlying JVM. Here executing in JVM doesn't mean that the thread is always executing in the OS as well - it may wait for a resource from the Operating system like the processor while being in this state.
    BLOCKED - this state represents a thread which has been blocked and is waiting for a moniotor to enter/re-enter a synchronized block/method. A thread gets into this state after calling Object.wait method.
    WAITING - this state represnts a thread in the waiting state and this wait is over only when some other thread performs some appropriate action. A thread can get into this state either by calling - Object.wait (without timeout), Thread.join (without timeout), or LockSupport.park methods.
    TIMED_WAITING - this state represents a thread which is required to wait at max for a specified time limit. A thread can get into this state by calling either of these methods: Thread.sleep, Object.wait (with timeout specified), Thread.join (with timeout specified), LockSupport.parkNanos, LockSupport.parkUntil
    TERMINATED - this state reprents a thread which has completed its execution either by returning from the run() method after completing the execution OR by throwing an exception which propagated from the run() method and hence caused the termination of the thread.

Difference between BLOCKED state and WAITING / TIMED_WAITING states?

When a thread calls Object.wait method, it releases all the acquired monitors and is put into WAITING (or TIMED_WAITING if we call the timeout versions of the wait method) state. Now when the thread is notified either by notify() or by notifyAll() call on the same object then the waiting state of the thread ends and the thread starts attempting to regain all the monitors which it had acquired at the time of wait call. At one time there may be several threads trying to regain (or maybe gain for the first time) their monitors. If more than one threads attempt to acquire the monitor of a particular object then only one thread (selected by the JVM scheduler) is granted the monitor and all other threads are put into BLOCKED state. Got the difference?

Difference between WAITING and TIMED_WAITING states?

The difference is quite obvious between the two. A thread in a TIMED_WAITING state will wait at max for the specified timeout period whereas a thread in the WAITING state keeps waiting for an indefinite period of time. For example, if a thread has called Object.wait method to put itself into WAITING state then it'll keep waiting until the thread is interrupted either by notify() method (OR by notifyAll() method) call on the same object by another thread. Similarly, if a thread has put itself into WAITING state by calling Thread.join method then it'll keep waiting until the specified thread terminates.

We can easily figure out that a thread in a WAITING state will always be dependent on an action performed by some other thread whereas a thread in TIMED_WAITING is not completely dependent on an action performed by some other thread as in this case the wait ends automatically after the completion of the timeout period.