Java Problem beim Ablauf von mehreren Threads

[darton]

Das mit dem ">10" ist natürlich falsch. Hab mich da verschrieben.
Hab mich nur gewundert, dass beim ExecutionService nicht schon der Fehler auftritt, sondern erst bei der Verarbeitung mit dem CompletionService.

Das Beispiel oben war natürlich nicht auf mein Problem bezogen. Mein Programm, welches die Dateien auslesen soll, bricht nämlich ab, nachdem der erste Thread die erste Zeile einer Datei ausgelesen hat. Das ist natürlich ziemlich blöd. Leider kann ich nirgends rausfinden, wodurch der Thread abgebrochen ist. Hab schon ein paar Ausgaben eingefügt, aber ich find damit nichts heraus. Kann man da irgendwas machen?

 

[Rossie]

Das Beispiel oben war natürlich nicht auf mein Problem bezogen. Mein Programm, welches die Dateien auslesen soll, bricht nämlich ab, nachdem der erste Thread die erste Zeile einer Datei ausgelesen hat. Das ist natürlich ziemlich blöd. Leider kann ich nirgends rausfinden, wodurch der Thread abgebrochen ist. Hab schon ein paar Ausgaben eingefügt, aber ich find damit nichts heraus. Kann man da irgendwas machen?

Notfalls geht das mit System.out#println, es empfiehlt sich aber ein Debugger. Breakpoint am Anfang des Tasks setzen und ab dafür.

Wenn keine Exception geworfen wird, kann es wohl nur ein Programmierfehler sein. Wenn man ein Programm möglichst modular aufbaut, lassen sich die einzelnen Teile gut einzeln testen. Den DateiTask in eine eigene Klasse. Dann lässt sich damit leicht ein UnitTest erstellen, und ganz ohne Multi-Threading Gerüst ausführen.

 

[darton]

Puh...hab nun endlich den Fehler gefunden. War ein blöder Programmierfehler.
Mein Code für den Consumer sieht mittlerweile so aus:

	        int remainingFutures = 0;
	    
		for (File file : files) {
			remainingFutures++;
			ecs.submit(new Producer(file));				
		}
		    
                Future<Queue<Datensatz>> future = null;
		Queue<Datensatz> datensaetze = null;

		while (remainingFutures > 0) {			
		try {
	            future = ecs.take();
	            remainingFutures--;
	            datensaetze = future.get();
	            for (Datensatz ds: datensaetze) {
	            	//Verarbeiten (z.B. in GUI schreiben)
	            }	            
	        } catch (ExecutionException e1) {
	            future.cancel(true);
	        } catch (InterruptedException e) {
				// TODO Auto-generated catch block
				e.printStackTrace();
	        }
                }		
		es.shutdown();

Wie läuft das denn jetzt eigentlich mit dem CompletionService ab? Der fängt ja schon dann an zu arbeiten, wenn die Threads noch am Einlesen sind. Merkt der denn, wenn eine neue Queue kommt und merkt der auch, wenn gerade keine Queue vorhanden ist, dass der warten muss, bis eine kommt?
Wenn ich das richtig sehe, erledigt Java für mich alle synchronized-Arbeiten, weil der Completion-Service immer alles nacheinander, wie es reingekommen ist, abarbeitet, oder?

 

[Rossie]

		for (File file : files) {
		    remainingFutures++;
                    ecs.submit(new Producer(file));				
		}
		    
                ...

		while (remainingFutures > 0) {			
		    try {
	                future = ecs.take();
	                ...	            
	            } catch (ExecutionException e1) {
	                future.cancel(true);
	            }
                }

Was hier als Producer bezeichnet wird, ist eigentlich ein Consumer. Producer ist die Routine, die die Dateien bereitstellt (was ich allerdings auch parallelisieren würde). Das future.cancel(true); ist sinnfrei. Die Exception allerdings sollte nicht ignoriert werden.

Wie läuft das denn jetzt eigentlich mit dem CompletionService ab? Der fängt ja schon dann an zu arbeiten, wenn die Threads noch am Einlesen sind. Merkt der denn, wenn eine neue Queue kommt und merkt der auch, wenn gerade keine Queue vorhanden ist, dass der warten muss, bis eine kommt?

Ja, ist doch in der API schön beschrieben.

Wenn ich das richtig sehe, erledigt Java für mich alle synchronized-Arbeiten, weil der Completion-Service immer alles nacheinander, wie es reingekommen ist, abarbeitet, oder?

Der CompletionService stellt die Ergebnisse der Tasks lediglich in einer Queue bereit. Was Du damit machst, bleibt Dir überlassen. Hier werden die Ergebnisse in der Tat suksessive mittels #take abgeholt.

Und ja, die Synchronisation übernimmt das Framework. Du kannst Dich auf Deine Anwendungslogik konzentrieren. Zum Lernen wäre es natürlich interessant, dass man das alles einmal selbst programmiert.