Warteschlange warten lassen bis Ereignis

[Taramsis]

Hi Leute,

ich habe einen Client-Server. Der Client liefert String Daten in Java und der Server führt diese in MATLAB aus.
Mein Problem ist, wenn mehrere gleichzeitig hierrauf Zugriff haben. Ich habe hierzu eine simple Warteschlang und möchte nun, dass solange keine Antwort vom Server kommt die inputs in die Warteschlange gepusht werden und dort dann seriell abgearbeitet werden.
Vielleicht sehe ich vor lauter Bäume den Wald nicht mehr, aber mir fällt keine gescheite Methode ein dies umzusetzen.
Wie kann ich denn aus ein und dem selben Thread wait und notify ausführen??

 

[Taramsis]

achso so siehts aus:

public void run() {
        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 type = split[0];
            int bufferSize = Integer.parseInt(split[1]);
            char[] buffer = new char[bufferSize];
            int charLength = in.read(buffer, 0, bufferSize);


            String inputLine = new String(buffer, 0, charLength);
            fifoQueue.inQueue(inputLine);
            
            if (out.equals("")){
            
            }
            while (!fifoQueue.queueStatus()) {
                String temp = fifoQueue.outQueue();
                view.getConsoleTextArea().append("Message received: " + "\n" + temp + "\n\n");

                String outputLine = handler.assign(type, temp);
               
                out.print(outputLine.length() + "\n" + outputLine);
                out.flush();
                view.getConsoleTextArea().append("Message sent: " + "\n" + outputLine + "\n\n");
            }
            out.close();
            in.close();
            skt.close();
           
            
        }       

        catch (IOException ex) {
            view.getConsoleTextArea().append("Network error");
        }
    }

 

[kay73]

Der ganze Ansatz ist nicht ok, wozu dient diese Queue genau? Laeuft es darauf hinaus, einfach alles, was aus dem InputStream kommt, in den OutputStream zu schreiben? Wenn Du zwischendurch noch Zeilen loggen willst, kannst Du das mit einer LinkedBlockingQueue und ein oder zwei Threads erreichen. Idealerweise solltest Du ein Listener-Interface definieren, wenn messages gelesen werden.

import java.io.BufferedReader;
import java.io.IOException;
import java.io.InputStream;
import java.io.InputStreamReader;
import java.io.OutputStream;
import java.io.PipedInputStream;
import java.io.PipedOutputStream;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.Callable;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.LinkedBlockingQueue;


public class ReaderWriter {

	public static void process (final InputStream is, final OutputStream os) throws IOException{
		final BlockingQueue<String> q = new LinkedBlockingQueue<String>();
		
		final ExecutorService exs = Executors.newFixedThreadPool(2);
		
		exs.submit(new Callable<Object>() {
			
			@Override
			public Object call() throws IOException, InterruptedException {
				final BufferedReader br = new BufferedReader(new InputStreamReader(is));
				for(;;) {
					for(final String line : br.readLine().split(";")) {
						q.put(line);
					}							
				}						
			}
		});
		
		exs.submit(new Callable<Object>() {

			@Override
			public Object call() throws Exception {
				for(;;) {
					final String line = q.take();
					System.out.println("Received "+line);							
					os.write(line.getBytes());
				}
			}					
		});
	}
	
	/**
	 * @param args
	 * @throws IOException 
	 */
	public static void main(String[] args) throws IOException {
		final PipedInputStream pis = new PipedInputStream();
		final PipedOutputStream pos = new PipedOutputStream();
		pos.connect(pis);

		process(pis, new OutputStream() {@Override public void write(int b) throws IOException { }});
		
		pos.write("Hello; World\nHow;are\n\n;you\n".getBytes());
		pos.flush();
				
		for(;;);
	}

}

 

[Taramsis]

Danke für die Rückmeldung!

Ich möchte alle Input Strings die mein Server erhählt in eine FIFO Warteschlange einspeisen. Und diese sollen dann nacheinander abgearbeitet werden. Manchmal braucht MATLAB länger, deswegen muss solange gewartet werden bis der Output zurückgegeben wurde! MATLAB arbeitet hierbei seriell!

 

[Michael...]

Was soll das:

[JAVA=17] if (out.equals("")){[/code]

Warum überträgst Du die "bufferSize" und liest nicht direkt alles per readLine() ein?
Ist es bewusst so gemacht, dass ein Client nur eine Anfrage senden kann?

Zu der ursprünglichen Frage:

1. Anfrage in die Queue stecken
2. Queue in einem separaten Thread abarbeiten
3. Ist die Anfrage abgearbeiten, dem Client Rückmeldung geben, kann nicht im Thread passieren, in dem die Abfrage eingelesen wurde.

 

[Taramsis]

Was soll das:

Warum überträgst Du die "bufferSize" und liest nicht direkt alles per readLine() ein?

Sorry war ein Fragment das ich löschen wollte!

Ist es bewusst so gemacht, dass ein Client nur eine Anfrage senden kann?

Ja da MATLAB nur seriell arbeitet!

 

[Michael...]

Ja da MATLAB nur seriell arbeitet!

Was hat das damit zu tun? Nur weil der Server nur eine Anfrage nach der anderen beantworten kann, könnte ein Client doch mehrere Anfrage in die Queue stellen.

 

[Taramsis]

Dann hab ich dich falsch verstanden! Ja es sollen mehrere Anfragen in die Queue!

 

[Taramsis]

@Kay: ich hab noch nie ne leere for-Schleife gesehen! Ist das nur ein trick um den return zu vermeiden? Wie genau funktioniert das?

 

[kay73]

@Kay: Ist das nur ein trick um den return zu vermeiden? Wie genau funktioniert das?

Ist semantisch dasselbe wie

while(true) {}

Schlechte Angewohnheit mancher C++-geschaedigter Zeitgenossen... Was Dein Vorhaben betrifft: Gibt es da noch Klaerungsbedarf? Falls ja beschreib mal, wie Du MATLAB angebunden hast und die Kommunikation funktionieren sollte.

Was auch moeglich warere ist, das Codestueck so aufzubohren, dass man mehrere

InputStream

s "einreihen" kann, jeder dann in einem

Callable

ausgelesen und die gelesene Zeile synchronisiert in der

BlockableQueue

landet.

 

[Taramsis]

Ich habe folgendes probiert:

public void run() {
        try {
            BufferedReader in = new BufferedReader(new InputStreamReader(skt.getInputStream()));

            String split[] = Pattern.compile("; ").split(in.readLine());

            setType(split[0]);
            int bufferSize = Integer.parseInt(split[1]);
            char[] buffer = new char[bufferSize];
            int charLength = in.read(buffer, 0, bufferSize);

            String inputLine = new String(buffer, 0, charLength);
            System.out.println("vorher: " + q.size());
            try {
                q.put(inputLine);
            } catch (InterruptedException ex) {
                Logger.getLogger(Network.class.getName()).log(Level.SEVERE, null, ex);
            }
            System.out.println("nachher: " + q.size());
            in.close();
            this.process();
        }

        catch (IOException ex) {
            view.getConsoleTextArea().append("Network error");
        }
    }

    public void process() {
        System.out.println("1111");
        ExecutorService exs = Executors.newFixedThreadPool(1);

        exs.submit(new Callable<Object>() {

            
            public Object call() throws Exception {

                PrintWriter out = new PrintWriter(new OutputStreamWriter(skt.getOutputStream()));
                System.out.println("vorher1: " + q.size());
                String line = q.take();
                System.out.println("nachher1: " + q.size());
                for(;;){
                    view.getConsoleTextArea().append("Message received: " + "\n" + line + "\n\n");

                    String outputLine = handler.assign(getType(), line);

                    out.print(outputLine.length() + "\n" + outputLine);
                    out.flush();
                    view.getConsoleTextArea().append("Message sent: " + "\n" + outputLine + "\n\n");

                    out.close();
                    skt.close();
                }
            }
        });
    }

Die queue wird gefüllt, hab aber Probleme diese herauszunehmen und weiter zu leiten!

 

[kay73]

Sorry, vielleicht reden wir aneinander vorbei, aber: Der "Trick" ist die BlockingQueue, die in einem Thread befüllt und in einem anderen geleert wird. Ich habe den Verdacht, dass Du die vielleicht gar nicht brauchst, weil Du den Input aus dem InputStream sowieso in einen Socket schreibst.

 

[Taramsis]

Eigentlich reicht doch ein Boolean flag und jedesmal, wenn grade in Benutzung muss der Thread warten! Aber klappen will es nicht wirklich:

public void run() {
        try {           

            while(this.getOccupied()){
                Random flagRandom = new Random();
                int randomTime = flagRandom.nextInt(6-2)+2;
                
                System.out.println(this.toString());
                System.out.println("vorher: "+ getOccupied());

                this.wait(randomTime);
                
            }
            this.setOccupied(true);
            BufferedReader in = new BufferedReader(new InputStreamReader(skt.getInputStream()));
            PrintWriter out = new PrintWriter(new OutputStreamWriter(skt.getOutputStream()));

            String split[] = Pattern.compile("; ").split(in.readLine());

            String type = split[0];
            int bufferSize = Integer.parseInt(split[1]);
            char[] buffer = new char[bufferSize];
            int charLength = in.read(buffer, 0, bufferSize);

            String inputLine = new String(buffer, 0, charLength);

            view.getConsoleTextArea().append("Message received: " + "\n" + inputLine + "\n\n");
            String outputLine = handler.assign(type, inputLine);
            out.print(outputLine.length() + "\n" + outputLine);
            out.flush();

            view.getConsoleTextArea().append("Message sent: " + "\n" + outputLine + "\n\n");
            
            in.close();
            out.close();
            skt.close();
            
            this.setOccupied(false);
            System.out.println("nachher" + getOccupied());
            
        }

        catch (IOException ex) {
            view.getConsoleTextArea().append("Network error");
        }      
    
    }

Eine Message wird geschickt und empfangen, dann aber hängt er sich auf, da es in der while Schleife verhart. Das Problem ist hierbei, dass ich Ihm mehrere Messages schicke und er für jeden ein Thread erstellen soll, die dann auf "false" warten, um ausgeführt zu werden. Ich weiß einfach keinen Rat!!!!!!

 

[maki]

kay73 hat schon recht mit seinen Tipps, könntest ruhig auf ihn hören

Wenn du es trotzdem selber machen willst (schwieriger), musst du es richtig machen

Ist occupied als volatile deklariert?

 

[Taramsis]

Ich weiss zwar nicht was du mit volatile meinst. Es ist wie folgt definiert:

private boolean occupied = false;

und

private boolean getOccupied(){
        return occupied;
    }

    private void setOccupied(boolean occupied){
        this.occupied = occupied;
    }

 

[maki]

Ich weiss zwar nicht was du mit volatile meinst...

In diesem Falle solltest du nicht versuchen so eine Multithreaded Anwendung zu schreiben, gehört zu den Grundlagen der Nebenläufigkeit in Java.

Da empfehle ich die Artikel von Angelika Langer: AngelikaLanger.com - Effective Java - A column by Angelika Langer & Klaus Kreft - Angelika Langer Training/Consulting

Ab "since 2008" wird es für dich wichtig.

Dein Code ist u.a. falsch, weil nicht garantiert ist dass die Änderung an der Variablen [c]occupied [/c] die ein Thread durchführt von einem anderen Thread auch "gesehen" werden.

 

[Taramsis]

Danke ! Dieser Artikel ist auch sehr gut!
Aber leider funzt der Zusatz von "volatile" nicht wirklich! Ich habe nun einfach folgendes gemacht:

import java.io.BufferedReader;
import java.io.IOException;
import java.io.InputStreamReader;
import java.io.OutputStreamWriter;
import java.io.PrintWriter;

import java.net.ServerSocket;
import java.net.Socket;
import java.util.Random;


import java.util.regex.Pattern;


public class Network implements Runnable {

    private ServerSocket serverSkt;
    private Socket skt;
    
    private int port = 4444;
    private boolean alive = false;

    private volatile boolean occupied = false;

    private View view;

    private Handling handler = new Handling(this);

    
    public Network(View view) {
        this.view = view;
    }

    public void listen() {
        try {
            serverSkt = new ServerSocket(port);
            alive = true;

            while(alive) {
                skt = serverSkt.accept();
                new Thread(this).start();
            }
        }

        catch (IOException ex) {
            view.getConsoleTextArea().append("Server connection could not be opened \n" +
                                             "Please make sure port " + port + " is open \n" +
                                             "and restart JaMaLa");
            view.getStatusMessageLabel().setText("JaMaLa connection error");
        }
    }
    
    public void run() {         
        Window win = new Window();
        win.setVisible(true);
        win.toFront();
        int count=0;

        while(occupied){
            Random flagRandom = new Random();
            int randomTime = flagRandom.nextInt(6-2)+2;
            count++;
            this.wait(randomTime);
        }

        win.getjTextArea1().append(this.toString() + "\n" + "vorher: " + count+ "\n");
        occupied = 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 type = split[0];
            int bufferSize = Integer.parseInt(split[1]);
            char[] buffer = new char[bufferSize];
            int charLength = in.read(buffer, 0, bufferSize);

            String inputLine = new String(buffer, 0, charLength);

            view.getConsoleTextArea().append("Message received: " + "\n" + inputLine + "\n\n");
            String outputLine = handler.assign(type, inputLine);
            out.print(outputLine.length() + "\n" + outputLine);
            out.flush();

            view.getConsoleTextArea().append("Message sent: " + "\n" + outputLine + "\n\n");

            String[] test = outputLine.split("\n");
            int j=0;
            for (int i = 0; i < test.length; i++) {
                j++;
            }
            win.getjTextArea1().append(test[j-1]+ "\n");
            
            in.close();
            out.close();
            skt.close();     
        }

        catch (IOException ex) {
            view.getConsoleTextArea().append("Network error");
        }

        occupied =false;
        win.getjTextArea1().append(this.toString() + "\n" + "nachher: " + occupied);
    }

    
    public View getView() {
        return view;
    }
    
    public void wait(int n){
        long t0,t1;
        t0=System.currentTimeMillis();
        do{
            t1=System.currentTimeMillis();
        }
        while (t1-t0<n);
    }


}

Würde mich über jede Anregung freuen!

 

[maki]

Was genau geht denn nicht?

Währe übrigens wichtig die Infos aus Exceptions nicht einfach zuverwerfen, printStackTrace zeigt dir wo was schiefgelaufen ist.

 

[Taramsis]

Ich habe ein Fenster erstellt mit einem Textarea, indem jeder erstellter Thread angezeigt wird, aber anscheinend warten die Thread nicht bis ein anderes durch ist. Matlab benötigt manchmal schon paar Sekunden für die Berechnung, doch anscheinend werden in der Zwischenzeit weitere anfragen an Matlab geschickt. Ich habe z.b. inhalb von 30 sek 15 Anfragen an dem Server geschickt, dann müßten, da der Input immer der selbe ist, auch das selbe Ergebnis herauskommen. Doch manchmal kommt ein anderes Ergebnis, da der Matlab Input anscheinend zum Teil gestört wird.

 

[Taramsis]

Hi Leute ich habe nun folgenden Stand:

import java.io.BufferedReader;
import java.io.IOException;
import java.io.InputStreamReader;
import java.io.OutputStreamWriter;
import java.io.PrintWriter;

import java.net.ServerSocket;
import java.net.Socket;
import java.util.logging.Level;
import java.util.logging.Logger;


import java.util.regex.Pattern;


public class Network implements Runnable {

    private ServerSocket serverSkt;
    private Socket skt;
    
    private int port = 4444;
    private boolean alive = false;

    private View view;
    private Handling handler = new Handling(this);


    public Network(View view) {
        this.view = view;
    }

    public void listen() {
        try {
            serverSkt = new ServerSocket(port);
            alive = true;

            while(alive) {
                skt = serverSkt.accept();
                new Thread(this).start();
            }
        }

        catch (IOException ex) {
            view.getConsoleTextArea().append("Server connection could not be opened \n" +
                                             "Please make sure port " + port + " is open \n" +
                                             "and restart Server");
            view.getStatusMessageLabel().setText("Server connection error");
        }
    }



    public void run() {
        
        String outputLine = "";
        String inputLine = "";
        
        synchronized(this){
            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 type = split[0];
                int bufferSize = Integer.parseInt(split[1]);
                char[] buffer = new char[bufferSize];
                int charLength = in.read(buffer, 0, bufferSize);

                inputLine = new String(buffer, 0, charLength);

                
                view.getConsoleTextArea().append("Message received: " + "\n" + inputLine + "\n\n");
                outputLine = handler.assign(type, inputLine);
                out.print(outputLine.length() + "\n" + outputLine);
                out.flush();

                view.getConsoleTextArea().append("Message sent: " + "\n" + outputLine + "\n\n");

                String[] test = outputLine.split("\n");
                int j=0;
                for (int i = 0; i < test.length; i++) {
                    j++;
                }
                
                in.close();
                out.close();
                skt.close();
            }

            catch (IOException ex) {
                view.getConsoleTextArea().append("Network error");
                Logger.getLogger(Network.class.getName()).log(Level.SEVERE, null, ex);
            }
        }
   }

    public View getView() {
        return view;
    }

}

Ich habe den "occupied" boolean nun entfernt und mich für "synchronized" entschieden.
Ich gebe mir auch den Fehler aus!
Dieser besagt, dass der Socket geschlossen sei! Wenn ich nun "skt.closed()" auskommentier muss ich auch "in" und "out" auskommentieren. Erst dann läuft es ohne Fehlermeldung durch, doch dann kommen anscheinend nicht alle Threads durch, da ich auf Client-Seite einen Time-Out bekomme!
Ich habe auch festgestellt, dass bei diesen Threads kein Input mitgegeben wird!

Ich weiß einfach nicht mehr weiter;(

 

[Michael...]

Wenn man die Streams schliesst ist auch die Socketverbindung geschlossen.

Ansonsten würde ich den Thread/das Runnable in eine eigene Klasse auslagern und den Instanzen eine Referenz auf das aktuell verbindende Socket übergeben. Das mit einer Instanzvariablen skt die eine Referenz auf das zuletzt verbundene Socket hält, ist keine gute Idee und könnte mal Probleme bereiten - wenn es das nicht schon macht.

 

[Taramsis]

Hi, danke für den Tipp!

Ich häng trotzdem etwas in der Luft, da ich mich mit Threads nicht so gut auskenne! Soll in meiner "Network"-Klasse dann auch meine Methode "listen()" sein, dass den Thread dann in einer anderen Klasse startet?

Network:

import java.io.IOException;
import java.net.ServerSocket;
import java.net.Socket;


public class Network {

    private ServerSocket serverSkt;
    private Socket skt;
    private int port = 4444;
    private boolean alive = false;
    private View view;

    private Handling handler = new Handling(this);
    private NetworkInOut networkInOut = new NetworkInOut(this);

    public Network(View view) {
        this.view = view;

    }

    public void listen() {
        try {
            serverSkt = new ServerSocket(port);
            this.setSocket(serverSkt.accept());
            networkInOut.startThread();                      
            
        }

        catch (IOException ex) {
            view.getConsoleTextArea().append("Server connection could not be opened \n" +
                                             "Please make sure port " + port + " is open \n" +
                                             "and restart Server ");
            view.getStatusMessageLabel().setText("Server connection error");
        }
        
    }

    
    public View getView() {
        return view;
    }
    
    public Socket getSocket(){
        return skt;
    }

    public void setSocket(Socket skt){
        this.skt = skt;
    }

    
}

NetworkInOut:

import java.io.BufferedReader;
import java.io.IOException;
import java.io.InputStreamReader;
import java.io.OutputStreamWriter;
import java.io.PrintWriter;
import java.util.logging.Level;
import java.util.logging.Logger;
import java.util.regex.Pattern;

public class NetworkInOut implements Runnable{

    private Network networker;
    private View view;
    
    private Handling handler;




    public NetworkInOut(Network networker) {
        this.networker = networker;
        
    }


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

    }
    
    public void run() {
        Window win = new Window();
        win.setVisible(true);
        win.toFront();
        int count=0;

        String outputLine = "";
        String inputLine = "";

        win.getjTextArea1().append(this.toString() + "\n" + "vorher: " + count+ "\n");
        synchronized(this){

        try {
            BufferedReader in = new BufferedReader(new InputStreamReader(networker.getSocket().getInputStream()));
            PrintWriter out = new PrintWriter(new OutputStreamWriter(networker.getSocket().getOutputStream()));

            String split[] = Pattern.compile("; ").split(in.readLine());

            String type = split[0];
            int bufferSize = Integer.parseInt(split[1]);
            char[] buffer = new char[bufferSize];
            int charLength = in.read(buffer, 0, bufferSize);

            inputLine = new String(buffer, 0, charLength);

            win.getjTextArea1().append("input: " + "\n" + inputLine + "\n");

            view.getConsoleTextArea().append("Message received: " + "\n" + inputLine + "\n\n"); //<--Null Pointer
            outputLine = handler.assign(type, inputLine);
            out.print(outputLine.length() + "\n" + outputLine);
            out.flush();

            view.getConsoleTextArea().append("Message sent: " + "\n" + outputLine + "\n\n");

            String[] test = outputLine.split("\n");
            int j=0;
            for (int i = 0; i < test.length; i++) {
                j++;
            }
            win.getjTextArea1().append(test[j-1]+ "\n");

            in.close();
            out.close();

        }

        catch (IOException ex) {
            win.getjTextArea1().append("input: " + "\n" + "Network error" + "\n");
            view.getConsoleTextArea().append("Network error");
            Logger.getLogger(Network.class.getName()).log(Level.SEVERE, null, ex);
        }


        win.getjTextArea1().append("output: " + "\n" + outputLine + "\n");
        }

    }

}

Ich bekomme nun ständig ein Nullpointer beim Pfeil! Was mache ich falsch?

 

[Andi_CH]

Pfeil ???:L

 

[Taramsis]

Zeile 58 in der Klasse NetworkINOut!

 

[Andi_CH]

Ich nehme an es gibt eine NullpointerException

Vermutlich liefert view.getConsoleTextArea() null - ist die textarea denn schon instanziert?

so zum testen:

            if (view.getConsoleTextArea()==null) {
            	System.out.println("Text area ist null");
            }

einfügen ...

 

[Taramsis]

Genau, der Fehler war, dass ich in Network die selbe Instanz verwenden wollte! Habe get und set Methoden in Network eingefügt für die Klasse Handling und View:

Nun bekomme ich folgenden Fehler:

Exception in thread "Thread-10" java.lang.NullPointerException
at java.util.regex.Matcher.getTextLength(Matcher.java:1140)
at java.util.regex.Matcher.reset(Matcher.java:291)
at java.util.regex.Matcher.<init>(Matcher.java:211)
at java.util.regex.Pattern.matcher(Pattern.java:888)
at java.util.regex.Pattern.split(Pattern.java:997)
at java.util.regex.Pattern.split(Pattern.java:1059)
at meinPacket.NetworkInOut.run(NetworkInOut.java:57)
at java.lang.Thread.run(Thread.java:619)

in Zeile 47 in der Klasse NetworkInOut! ;(

 

[Gast2]

Dann schau halt nach was in Zeile 47 null ist, sollte doch kein großes problem sein.

 

[Andi_CH]

Solche verketteten Ausdrücke wie der fogende Fantasieausdruck sind immer schei....

getView().getTabel().getRow().getfield(). .....

Wenn es irgendwo knallt gibt es zwei Varianten - entweder mit dem Debugger reinsteppen oder die Ausdrücke zerlegen und jeden einzelnen return-Wert auf Gültigkeit überprüfen.