✔ 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/docs/api/java/util/concurrent/ThreadPoolExecutor.html

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:

public void startThread() {
        new Thread(this).start();
}
public void run() {
  ....
}

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

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.

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

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:

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?

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:

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.