Optimizing multithreaded queue processing code in Java

Question

I have a code that creates 10 objects of a class, that implement runnable. Each object is kept in hashmap for later usage. Each object is running on a separate thread. Each object has a public method where items can be added into the queue. The object processes the queue with infinite loop.

I want to know if this solution is OK or is there something that is totally wrong/useless/missing(especially the use of volatile and syncronized keywords)?

MultithreadingTest.class

package multithreadingtest;

import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

/**
 * Multithreading example.
 *
 * @author lkallas
 */
public class MultithreadingTest {

    private static final int NUM_OF_THREADS = 10;
    private static String name;
    private static final Map<Integer, ThreadWorker> objectMap = new HashMap<>();    //Map or storing Threadworker objects

    public static void main(String[] args) {
        ExecutorService executor = Executors.newFixedThreadPool(NUM_OF_THREADS);
        //Creating threads
        for (int i = 0; i < NUM_OF_THREADS; i++) {
            name = "ThreadWorker" + String.valueOf(i);
            ThreadWorker thread = new ThreadWorker(name);
            objectMap.put(i, thread);   //Add objects to map            
            executor.execute(thread);
        }
        for (int i = 0; i < 10; i++) {
            ThreadWorker worker = objectMap.get(i);
            for (int j = 0; j < 10; j++) {
                worker.addToQueue("Test1");
            }
        }
    }
}

ThreadWorker.class

package multithreadingtest;

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

/**
 * Worker class that performs operations in another thread.
 *
 * @author lkallas
 */
public class ThreadWorker implements Runnable {

    private final String threadName;
    private volatile Queue workQueue;   //Does this have to volatile??

    /**
     * Class constructor.
     *
     * @param threadName Name of the thread for identifying.
     *
     */
    public ThreadWorker(String threadName) {
        this.threadName = threadName;
        this.workQueue = new LinkedList();
        System.out.println(String.format("Thread %s started!", threadName));
    }

    /**
     * Adds items to the queue.
     *
     * @param object Object to be added to the queue.
     */
    public synchronized void addToQueue(String object) {
        workQueue.add(object); //Does it have to be syncronized void
    }

    @Override
    public void run() {
        while (true) {
            if (!workQueue.isEmpty()) {
                System.out.println("Queue size: " + workQueue.size());
                String item = (String) workQueue.peek();
                //Process item
                System.out.println(threadName + " just processed " + item);
                workQueue.remove();
            }
        }
    }
}

Any help and suggestions are much appreciated!

I think this is more of a case for [Code Review](http://codereview.stackexchange.com/). — Siguza, Jun 16 '15 at 20:01
Synchronizer object could be a trivial object instead of thread itself. — huseyin tugrul buyukisik, Jun 16 '15 at 20:04
You could get rid of the while loop in the worker by changing to a BlockingQueue implementation. With the blocking queue you can use `take()` to get the head of the queue, blocking until one is available. — roby, Jun 16 '15 at 20:12
The synchonization should be done on the data structure. Use a ConcurrentLinkedQueue or wrap the list in `Collections.synchronizedList`. In this case the remove method is not synchonized so you could run into trouble if you add and remove items at the same time, which is exactly what you're doing. — Cosu, Jun 16 '15 at 20:17
Idea of ExecutorService is to have less number of threads than you have Runnables. Since here number of Runnables is equal to the number of threads, you need not ExecutorService. Just create a thread for each Runnable. — Alexei Kaigorodov, Jun 16 '15 at 20:21
Take a look here: [How to make a Java thread wait for another thread's output?](https://stackoverflow.com/questions/289434/how-to-make-a-java-thread-wait-for-another-threads-output) — tirz, Jun 16 '15 at 21:40
Thank you all for these useful suggestions! I will try all of these suggestions and will post the code that suits me the best. — lkallas, Jun 17 '15 at 07:19

score 2 · Accepted Answer · answered Jun 16 '15 at 20:17

workQueue is thread-local and does not need to be volatile (it is private and doesn't have a public setter method)
Make workQueue a BlockingQueue - this queue is thread-safe, so you don't need to synchronize addToQueue. In addition, you don't need to spin inside of run - instead you call take() on the queue, and the thread blocks until an item is available.
You appear to be doing too much work inside of MultithreadingTest - rather than adding items to individual queues, you can instead have all workers share the same BlockingQueue, then main just needs to add items to that single BlockingQueue and the workers will take care of load balancing themselves. Note that even though the BlockingQueue is shared, it still doesn't need to be volatile because the reference to the BlockingQueue doesn't change once a worker is initialized (make the field private final BlockingQueue<String> workQueue - a final field never needs to be volatile).

Thank you for this very good explanation! I will try using BlockingQueue. Cheers! — lkallas, Jun 17 '15 at 07:20

Optimizing multithreaded queue processing code in Java

1 Answers1