I have created a class that attempts to process some messages in multiple threads, where each message belongs to a particular group. Each message is being added to the ConcurrentHashMap which has a key as the group number and is filling up while these threads are "processing". I have noticed that they sometimes run in parallel and sometimes not. And to make matters worse when there are more than 2 process threads running always they deadlock entirely.
[EDIT]
Iteration of the ConcurrentHashMap
seemed at the time to be a good way of going through all the elements as the numbered message groups (keys) are not known and it could change over time. The task specified that all messages be grouped together for processing but when when there is only one message in a group it should still process. So I thought this was a way to sort the elements as they arrive without knowing at the beginning which groups exist.
[\EDIT]
public class GroupPriorityProcess implements Runnable {
private static final Object lock = new Object();
private static final Object counterLock = new Object();
private static int threadCounter = 0;
private final int currentThreadNumber;
private static Iterator<Integer> groupIterator;
private ConcurrentHashMap<Integer, LinkedBlockingQueue<Message>> groupMsgQueues;
public GroupPriorityProcess(ConcurrentHashMap<Integer, LinkedBlockingQueue<Message>> groupedMsgQueues) {
groupMsgQueues = groupedMsgQueues;
synchronized(lock){
if (groupIterator == null)
groupIterator = groupedMsgQueues.keySet().iterator();
}
synchronized (counterLock) {
currentThreadNumber = (threadCounter++);
}
}
// Main while loop for threads to process messages
public void run() {
while (true) {
LinkedBlockingQueue<Message> queue = chooseGroup();
synchronized (queue) {
process(queue);
}
}
}
// Loops till finds a message group available for processing.
private LinkedBlockingQueue<Message> chooseGroup() {
synchronized (lock) {
while (!groupIterator.hasNext()) {
groupIterator = groupMsgQueues.keySet().iterator();
}
LinkedBlockingQueue<Message> queue = groupMsgQueues.get(groupIterator.next());
return queue;
}
}
// takes messages from the a particular message group queue to completes the
// send process
private void process(LinkedBlockingQueue<Message> queue) {
try {
while (!queue.isEmpty()) {
Message msg = queue.take();
msg.appendMessage("Thread: " + currentThreadNumber);
msg.completed();
}
} catch (InterruptedException e) {
e.printStackTrace();
}
}
}
[EDIT]
The messages are added here in another class GatewayImp
.
public void send(Message msg) {
int groupID = msg.getGroupID();
if (groupedMsgQueues.containsKey(groupID)) {
LinkedBlockingQueue<Message> queue = groupedMsgQueues.get(groupID);
queue.add(msg);
} else {
LinkedBlockingQueue<Message> queue = new LinkedBlockingQueue<Message>();
try {
queue.put(msg);
} catch (InterruptedException e) {
e.printStackTrace();
}
groupedMsgQueues.put(groupID, queue);
}
}
}
I have a few times when I use 'check then act' which I thought needs to be in a synchronized block to make atomic but I would like to know there is some better way. Please any help with this is greatly appreciated as I am only just starting to learn about concurrency and I'm finding it hard to get my head around locking in particular.
ConcurrentHashMap
. Having to iterate through the whole KeySet each time you use the Map seems suboptimal as you're not taking advantage of the best features ofConcurrentHashMap
. If you useAtomicInteger
you won't need do your own locking on your counter variable (which is better for concurrency).groupMsgQueues
is empty. This thread holds a global lock that will block other threads from callingchooseGroup()
or from constructing additionalGroupPriorityProcess
instances. How that might cause your whole process to lock up depends on how messages are produced and how queues are added togroupMsgQueues
. Please provide more information.AtomicInteger
for keeping track of threads but other than readability in this instance are there any other advantages?AtomicInteger
requires no external locking. For this reason, it avoids lock contention in a highly concurrent environment and can offer better performance. There is also no chance of liveness failures when using classes that require no locking.CopyOnWriteArrayList
. No synchronization is necessary, even during iteration.