5
\$\begingroup\$

I've developed the following multi-threaded code and would like a review. It is supposed to be implementing the Producer-consumer problem as defined here.

I have multiple consumers and one producer. I would like to know whether there are any problems with my use of concurrency and also if you have any suggestions on how to better implement the solution. I don't have a lot of experience with concurrent programming and am trying to improve.

package consumerproducer;

import java.util.ArrayDeque;
import java.util.ArrayList;
import java.util.Deque;
import java.util.List;


public class ConsumerProducer1 {

    private static int QUEUE_SIZE = 10;

    private Deque<Integer> queue = new ArrayDeque<Integer>(QUEUE_SIZE);


    public synchronized boolean addWorkItem(Integer i) {
        if (queue.size() < QUEUE_SIZE) {
            queue.addLast(i);
            return true;
        } else {
            return false;
        }
    }

    public synchronized Integer getWork() {
        if (queue.size() > 0) {
            return queue.removeFirst();
        } else {
            return null;
        }
    }

    public static void main(String[] args) {
        new ConsumerProducer1().go();
    }

    public void go() {
        ConsumerThread ct = new ConsumerThread();
        ct.start();
        ConsumerThread ct2 = new ConsumerThread();
        ct2.start();
        ProducerThread pt = new ProducerThread();
        pt.start();
    }

    class ConsumerThread extends Thread {
        public void run() {
            while(true) {
                Integer work = getWork();
                if (work == null) {
                    Thread.yield();
                } else {
                    System.out.println("Thread: " + this.getId() + " received work: " + work);
                }
            }
        }
    }

    class ProducerThread extends Thread {
        private List<Integer> retryList = new ArrayList<Integer>();
        public void run() {
            while(true) {
                Integer currWork;
                if (retryList.size() == 0) {
                    currWork = (int) (Math.random() * 100);
                    System.out.println("Thread: " + this.getId() + " generated new work: " + currWork);
                } else {
                    currWork = retryList.remove(0);
                    System.out.println("Thread: " + this.getId() + " retrying old work: " + currWork);
                }
                if (!addWorkItem(currWork)) {
                    System.out.println("Thread: " + this.getId() + " could not add work (because buffer is probably full): " + currWork);
                    retryList.add(currWork);
                    Thread.yield();
                } else {
                    System.out.println("Thread: " + this.getId() + " added work to queue: " + currWork);
                }
            }
        }
    }
}
\$\endgroup\$
1
\$\begingroup\$

Suggest adding final to the QUEUE_SIZE constant and the collection instance also.

synchronized usage for protecting the ArrayDeque collection from concurrent access, but ArrayDeque is not thread safe. Perhaps two threads, one calls getWork() and another the addWorkItem(Integer i). Suggest using a thread safe collection instead of ArrayDeque.

Multiple threads may not always have access to System.out. Instead can use a concurrent logger or other mechanism to capture output, or have each thread write to a thread safe collection as an instance in ConsumerProducer1. Another option is having a Callable that returns desired output as a String, with the main routine the only thread accessing System.out and echoing that result.

\$\endgroup\$
1
\$\begingroup\$

Concurrency is hard enough; don't make it harder by adding threads.

In most cases where you want to divide your work among multiple threads, the right approach to take is to submit Runnables to an ExecutorService, rather than to manage your own threads.

You also seem to be using Thread.yield() as a way of idling until work is available. The usual mechanism is to wait() on the lock until you can make progress. Using that approach, getWork might look like this:

public synchronized Integer getWork() {
    // While our progress is blocked...
    while(queue.isEmpty()) {
        // release the lock; idle until we are woken up
        // and re-acquire the lock
        wait();
        // now we have the lock again, but we still need
        // to check to see if we can make progress
    }

    // Now we have the lock AND a queue that isn't empty

    // Take a moment to notify the other threads that are waiting that
    // the state of the queue has changed, and they may need to re-evaluate
    // whether they can make progress.  To be honest, I'm not sure if you
    // need it when you have one non-blocking producer.
    notifyAll();

    // grab the data we need before releasing the lock.
    return queue.removeFirst();
}

The producer logic definitely needs to notify the idle threads when it has added items to the queue, so that they can wake up and pull them off

public synchronized boolean addWorkItem(Integer i) {
    if (queue.size() < QUEUE_SIZE) {
        queue.addLast(i);

        // let any idling consumers know that the queue is no
        // longer empty
        notifyAll();

        return true;
    } else {
        return false;
    }
}

BTW: I love that you implemented the producer this way. If the producers are outrunning the consumers, fail fast is often a better approach than blocking (for more thoughts on this, search for "backpressure")

            if (!addWorkItem(currWork)) {
                System.out.println("Thread: " + this.getId() + " could not add work (because buffer is probably full): " + currWork);
                retryList.add(currWork);

That said, if you respond to backpressure by writing the workload to an unbounded list for later replay, what's the point in using a bounded queue?

For clarity, I would probably write this as

            if (!addWorkItem(currWork)) {
                onOverloaded(currWork);

Note that Effective Java, Item #69 would tell you to eschew wait() and notify() in favor of the facilities available in java.util.concurrent. In this case, a LinkedBlockingQueue might be preferred to rolling your own.

If you want to learn a whole lot about producer/consumer, read up on LMAX Disruptor.

\$\endgroup\$

Your Answer

By clicking “Post Your Answer”, you agree to our terms of service, privacy policy and cookie policy

Not the answer you're looking for? Browse other questions tagged or ask your own question.