ConcurrentLinkedQueue nacheinander abarbeiten?

[tehK]

Hallo zusammen,

Ich bin erst Anfänger auf dem Gebiet Threading / Multithreading und hätte da mal eine konkrete Frage, bei der ich momentan evtl. auch einfach auf dem Schlauch stehe.

Folgendes Problem:
Eine ConcurrentLinkedQueue erhält der Reihe nach einfache String-Objekte, die sowas, wie Commands darstellen. Wann / in welcher Geschwindigkeit die da reinkommen kann ich nicht beeinflussen. Diese Commands sollen jetzt nacheinander abgearbeitet werden (FIFO). Für jedes Command soll ein neuer Thread aufgemacht werden, in dem das Command abgearbeitet wird. Während so ein Thread läuft, soll die ConcurrentLinkedQueue zwar weitere Strings empfangen (bis zu einem Maximum), aber es soll kein weiteres Command abgearbeitet werden.

Ich gebe mal einen groben Auszug aus der Vorgabe:

public ConcurrentLinkedQueue<String> incomingMessageQueue = new ConcurrentLinkedQueue<String>();

public void enqueueStringCommands(final String tmp) {

	if(incomingMessageQueue.size() <= INCOMING_MESSAGE_QUEUE_MAX) {
		incomingMessageQueue.add(tmp);			
	}			
		
	//... insert code here
}

Wie gesagt... Darauf, wie häufig die Funktion enqueueStringCommands aufgerufen wird, habe ich keinen Einfluss. Nach dem Hinzufügen in die Liste müsste ich jetzt den Code einfügen, der für den jeweils nächsten String einen Thread ablaufen lässt und nach dessen Ablaufen das nächste Kommando aus der Queue holt.

Wäre super, wenn mir da jemand auf die Sprünge helfen könnte

MfG,
-- tehK

 

[XHelp]

Wo ist denn jetzt die Frage?
So wie es aussieht hast du nur ein Thread, der alle Kommandos abarbeiten soll. Den kannst du ja auch starten. Bei jedem hinzufügen kannst du überprüfen ob er noch läuft, wenn nicht, dann noch mal anstoßen.

 

[Kr0e]

Was du brauchst ist nicht eine ConcurrentLinkedQueue, sondern eine LinkedBlockingQueue.
ConcurrentLinkedQueue wurde dafür gemacht, um genau dein Vorgehen zu verhindern (Locks und Wartezustände).
Eine ConcurrentLinkedQueue hat außerdem kein Maximum und immer mit size() die Größe erfragen ist sehr unperformant bei dieser Impl. Davon abgesehen, treten hier Syncfehler auf... Das was du wirklich brauchst, ist das hier:
(Und immer einen neuen Thread starten ist nicht nötig und unperformant)

private static final LinkedBlockingQueue<String> queue = new LinkedBlockingQueue<String>(INCOMING_MESSAGE_QUEUE_MAX);

public static void addToQueue(String command) throws InterruptedException {
    //Wenn queue voll ist, wird hier gewartet bis wieder Platz ist für einen String!
    queue.put(command);
}

private static final Thread worker = new Thread() {
    @Override
    public void run() {
         while(!Thread.interrupted()) {
             try {
                 //Wartet solange, bis etwas in der Queue ist!
                 String command = queue.take();

                 //Iwas mit dem Command machen....
                 process(command);
             }
             catch(Exception e) {
             }
         }
    }
};

static {
    worker.start();
}

Voila, fertig!

 

[tehK]

Genau den Wink hab ich gebraucht. Danke Dir, Kr0e!