FiFo für Threads

[Taramsis]

Hi,

ich habe ein Server der Anfragen an MATLAB weitergibt implementiert. Für jede Anfrage wir ein Thread gestartet. Da MATLAB immer nur eine Anfrage bearbeiten kann, werden z.Z. die Threads über eine Boolean Variable gesteuert. D.h. eine globale Variable wird auf besetzt gestellt, wenn MATLAB gerade arbeitet. Danach greifen die Threads mit eine zufälligen Verzögerung auf die MATLAB zu. Also alles recht inkonsistent.

Nun wollte ich einfach eine FIFO-Warteschlange erstellen, die sich mit den Threads füllt und diese dann nach einander abarbeitet.
Kann man überhaupt Threads in einer FIFO ablegen oder braucht man einen Threadpool?
Und wie würde das denn aussehen?

[twagi]

Hi,

schau Dir mal die Executor an. Der SingleThreadExecutor ist glaube ich genau das was Du suchst.

http://download.oracle.com/javase/6/...lExecutor.html

1
2
ExecutorService theExecutor = Executors.newSingleThreadExecutor();
theExecutor.submit(new Runnable());

Gruß twagi

[Taramsis]

Nur um das richtig zu verstehen!
Ich führe mein Thread wie folgt aus:

1
2
3
4
5
6
public void startThread() {
        new Thread(this).start();
}
public void run() {
  ....
}

Nun soll ich sozusagen das "startThread()", in einem Executor ausführen?
also so:

1
2
3
4
5
6
7
public void startThread() {
        ExecutorService theExecutor = Executors.newSingleThreadExecutor();
        theExecutor.submit(new Thread(this).start(););        
}
public void run() {
  ....
}

und dann mit etwas wie shutdown schließen? Ne dumme Frage: wo ist dann die Warteschlange?

[twagi]

Nein der Executor startet die Threads für Dich. Du übergibst diesem per submit() die Runnables, die ausgeführt werden sollen. Der Executor führt diese dann nacheinander aus, da es ein SingleThreadExecutor ist.

1
2
ExecutorService theExecutor = Executors.newSingleThreadExecutor();
theExecutor.submit(this);

A pros pos:
Damit Du jetzt mehrere Runnables damit verwalten kannst solltest Du das Starten aus der Klasse auslagern.

[Taramsis]

Also eher so was

1
theExecutor.submit(new Thread(this));

um mein run zu starten!

Und alles andrere wird vom newSingleThreadExecutor() erledigt?

[twagi]

Ja,

aber wenn this das Interface Runnable implementiert, dann brauchst Du keinen neuen Thread zu erzeugen.

[Taramsis]

Mein Server erzeugt ein Thread, wenn eine Anfrage an dem Server geschickt wird in einer Klasse Network:

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
public void listen() {
        try {
            serverSkt = new ServerSocket(port);
            alive=true;
            while(alive){
                Socket skt = serverSkt.accept();
                new NetworkThread(skt,this).startThread();
            }           
        }
 
        catch (IOException ex) {
            this.getWin().getConsoleTextArea().append("Server connection could not be opened ");
            this.getWin().getStatusMessageLabel().setText("connection error");
        }
    }

Ich wollte dadurch dafür sorgen das die entsprechenden Clients bedient werden!
Starte ich jetzt von zwei Clients eine Anfrage bekomme ich einen Timout!

Muss ich in meiner Klasse NetworkThread im run() etwas anpassen?

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
public void startThread() {
        theExecutor.submit(new Thread(this));
}
 
public void run() {
        String outputLine = "";
        String inputLine = "";
        consoleOut = "";
 
 
//        while(net.getOccupied()){
//            Random flagRandom = new Random();
//            int randomTime = flagRandom.nextInt(6-2)+2;
//            this.wait(randomTime);
//        }
        
        
//        net.setOccupied(true);
        try {
            BufferedReader in = new BufferedReader(new InputStreamReader(skt.getInputStream()));
            PrintWriter out = new PrintWriter(new OutputStreamWriter(skt.getOutputStream()));
 
            String split[] = Pattern.compile("; ").split(in.readLine());
            String modelType = split[0];
            String model = split[1];
            int paramNumber = Integer.parseInt(split[2]);
            String[] param = new String[paramNumber];
 
            for (int i = 0; i < paramNumber; i++) {
                param[i] = split[i + 3];                
            }
 
            int bufferSize = Integer.parseInt(split[paramNumber + 3]);
            char[] buffer = new char[bufferSize];
 
            int pos = 0;
            int read =0;
            while ( (pos < bufferSize) && ((read = in.read(buffer, pos, (bufferSize-pos)))!= -1) ) {
                pos+=read;
            }
            inputLine = new String(buffer, 0, pos);
                       
            consoleOut += "Message received: " + "\n" + inputLine + "\n\n";
 
            if (modelType.equals("Simulink")){
                outputLine = launchSimulink(model, param, inputLine);
            }
            else {
                outputLine = launchMATLAB(model, param, inputLine);
                
            }
            out.print(outputLine.length() + "\n" + outputLine);
            out.flush();
 
            net.getWin().getConsoleTextArea().append(consoleOut + "Message sent: " + "\n" + outputLine + "\n\n");
 
             in.close();
            out.close();
        }
 
        catch (IOException ex) {
            net.getWin().getConsoleTextArea().append("Network error ");
            Logger.getLogger(Network.class.getName()).log(Level.SEVERE, null, ex);
        }
 
//        net.setOccupied(false);
       
    }

[twagi]

Probier es mal, in dem Du das Starten der Threads aus dem NetworkThread herausziehst.
Also in etwas so:

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
public void listen() {
        try {
            serverSkt = new ServerSocket(port);
            ExecutorService theExecutor = Executors.newSingleThreadScheduledExecutor();
            alive=true;
            while(alive){
                Socket skt = serverSkt.accept();
                theExecutor.submit(new NetworkThread(skt,this));
            }           
        }
 
        catch (IOException ex) {
            this.getWin().getConsoleTextArea().append("Server connection could not be opened ");
            this.getWin().getStatusMessageLabel().setText("connection error");
        }
    }

[Taramsis]

Jetzt klappt es! Danke!
Aber zwei Verständnisfragen:

1. Kann ExecutorService nicht in der selben Instanz ausgeführt werden?
2. Handelt der ExecutorService immer nach FIFO?

[twagi]

Hi,

bei Deinem ersten Beispiel hattest Du nicht die selbe Instanz, sondern hast immer wieder ein neues Objekt erzeugt, mit einem eigenen ThreadPool. Ich schätze, daher kamen die Probleme.

Ein Executor nimmt für die zu verarbeitenden Task ein Queue. Die funktioniert üblicherweise nach FIFO, muss aber nicht.

[Taramsis]

Ein Executor nimmt für die zu verarbeitenden Task ein Queue. Die funktioniert üblicherweise nach FIFO, muss aber nicht.

Sorry aber die API sagt nicht über das Warteschlangenverhalten!
D.h. jetzt funktioniert es nach FIFO kann aber in andere Arten umgewandelt werden?

[twagi]

Der SingleThreadExecutor verwendet als Collection eine Queue.

http://download.oracle.com/javase/6/...Executor%28%29

Laut API arbeitet eine Queue (BlockingQueue) in der Regel nach FIFO.

http://download.oracle.com/javase/6/...til/Queue.html

Ich sehe gerade, der Executor verwendet intern eine LinkedBlockingQueue. Diese arbeitet auch nach FIFO.
http://download.oracle.com/javase/6/...kingQueue.html

Sollte klappen...

[Taramsis]

Ich hatte mir jetzt eine Erweiterung für meinen Server überlegt:

Und zwar so eine Art Masterserver mit mehreren Slaveservern. Der Masterserver würde die Anfragen in einer Warteschlange speichern und an die Slaveserver verteilen.

Könnte ich auch hier den "newSingleThreadExecutor()" verwenden?

D.h. es wären zwei SlaveServer verfügbar müßte die Warteschlange ja nicht immer nur ein Thread zulassen , sondern zwei! Ich glaube nicht!