Signalling like juancn does in his answer is not my strong suit, so I made an even cruder solution using one Semaphore for signalling combined with a request-counter.
There is one lock involved (subjectsLock) which synchronizes everything at one point in time. The lock is required to ensure there are no memory leaks: since there can be any number of subjects (a.k.a. object identifiers in your question), cleanup is essential. And cleanup requires knowing when something can be removed and that is difficult to determine without a lock that brings everything to one known state at a certain point in time.
The test in the main-method in the code shown below is a bit hard to read, but it serves as a starting point for a demonstration of how the code works internally. The main logic is in the methods executeRequest, addSubject and removeSubject. If those three methods do not make sense, another solution should be used.
Stress-testing will have to determine if this solution is fast enough: it depends on the number of requests (per second) and the amount of time it takes to complete an action. If there are many requests and the action is short/fast, the (synchronization) overhead from the lock could be to high.
// package so;
import java.util.ArrayList;
import java.util.LinkedHashMap;
import java.util.List;
import java.util.Map;
import java.util.Random;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Semaphore;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.locks.ReentrantLock;
import java.util.stream.IntStream;
public class RequestQueue {
public static void main(String[] args) {
// Randomized test for "executeRequest" method below.
final int threadCount = 4;
ExecutorService threadPool = Executors.newFixedThreadPool(threadCount);
try {
final int requestCount = 100;
final RequestQueue rq = new RequestQueue();
final Random random = new Random();
IntStream.range(0, requestCount).forEach(i -> threadPool.execute(new Runnable() {
@Override
public void run() {
try {
String subject = "" + (char) (((int)'A') + random.nextInt(threadCount));
rq.executeRequest(subject, new SleepAction(i, subject, 50 + random.nextInt(5)));
} catch (Exception e) {
e.printStackTrace();
}
}
}));
sleep(100); // give threads a chance to start executing.
while (true) {
sleep(200);
List<String> subjects = rq.getSubjects();
System.out.println("Subjects: " + subjects);
if (subjects.isEmpty()) {
break;
}
}
} catch (Exception e) {
e.printStackTrace();
} finally {
threadPool.shutdown();
}
}
private Map<String, QueueLock> subjects = new LinkedHashMap<>();
// a fair ReentrantLock is a little bit slower but ensures everybody gets their turn in orderly fashion.
private final ReentrantLock subjectsLock = new ReentrantLock(true);
private class QueueLock {
// a fair Semaphore ensures all requests are executed in the order they arrived.
final Semaphore turn = new Semaphore(1, true);
final AtomicInteger requests = new AtomicInteger(1);
public String toString() { return "request: " + requests.get(); }
}
/**
* Allow all requests for different subjects to execute in parallel,
* execute actions for the same subject one after another.
* Calling thread runs the action (possibly after waiting a bit when an action for a subject is already in progress).
*/
public String executeRequest(String subject, Runnable action) throws InterruptedException {
QueueLock qlock = addSubject(subject);
try {
int requestsForSubject = qlock.requests.get();
if (requestsForSubject > 1) {
System.out.println(action.toString() + " waiting for turn " + requestsForSubject);
}
qlock.turn.acquire();
if (requestsForSubject > 1) {
System.out.println(action.toString() + " taking turn " + qlock.requests.get());
}
action.run();
} catch (Exception e) {
e.printStackTrace();
} finally {
removeSubject(subject);
}
return timeSinceStart() + " " + subject;
}
private QueueLock addSubject(String s) {
QueueLock qlock = null;
subjectsLock.lock();
try {
qlock = subjects.get(s);
if (qlock == null) {
qlock = new QueueLock();
subjects.put(s, qlock);
} else {
qlock.requests.incrementAndGet();
}
} finally {
subjectsLock.unlock();
}
return qlock;
}
private boolean removeSubject(String s) {
boolean removed = false;
subjectsLock.lock();
try {
QueueLock qlock = subjects.get(s);
if (qlock.requests.decrementAndGet() == 0) {
subjects.remove(s);
removed = true;
} else {
qlock.turn.release();
}
} finally {
subjectsLock.unlock();
}
return removed;
}
public List<String> getSubjects() {
List<String> subjectsBeingProcessed = new ArrayList<>();
subjectsLock.lock();
try {
// maintains insertion order, see https://stackoverflow.com/a/18929873/3080094
subjectsBeingProcessed.addAll(subjects.keySet());
} finally {
subjectsLock.unlock();
}
return subjectsBeingProcessed;
}
public static class SleepAction implements Runnable {
final int requestNumber;
final long sleepTime;
final String subject;
public SleepAction(int requestNumber, String subject, long sleepTime) {
this.requestNumber = requestNumber;
this.sleepTime = sleepTime;
this.subject = subject;
}
@Override
public void run() {
System.out.println(toString() + " sleeping for " + sleepTime);
sleep(sleepTime);
System.out.println(toString() + " done");
}
public String toString() {return timeSinceStart() + " " + subject + " [" + Thread.currentThread().getName() + "] " + String.format("%03d",requestNumber); }
}
public static final long START_TIME = System.currentTimeMillis();
public static String timeSinceStart() {
return String.format("%05d", (System.currentTimeMillis() - START_TIME));
}
public static void sleep(long milliseconds) {
try {
Thread.sleep(milliseconds);
} catch (Exception e) {
throw new RuntimeException(e);
}
}
}
