Thread Warteschlange

[Taramsis]

Hi,

ich habe einen Server der für jede Anfrage einen Thread erstellt und diese dann nacheinander abarbeitet.

Z.Z. habe ich folgendes:

public void listen() {
        try {
            serverSkt = new ServerSocket(port);
            ExecutorService theExecutor = Executors.newSingleThreadExecutor();
            alive=true;
            while(alive){
                Socket skt = serverSkt.accept();
                theExecutor.submit(new NetworkThread(skt,this));

            }           
        }
        catch (IOException ex) {
        }
}

Nun wollte ich aus meinem Server eine Art Masterserver machen, der die Anfragen an Slaveserver weiterleitet.
Ich werde aus "Executors.newSingleThreadExecutor();" nicht wirklich schlau, man kann hier ja nicht auf die Elemente zugreifen, oder? Ergo ich muss einen anderen Mechanismus benutzen!? Was bietet sich den an?

 

[SlaterB]

NetworkThread ist ein schlechter Name wenn die einzelnen Verbindungen keine Threads sind sondern nur Runnable,
bzw. sie müssen keine Threads sein wenn du sie nur als Runnable verwendest,
NetworkTask wäre passenderer Name für aktuelle Verwendung,

die Objekte selber könntest du auch in einer Liste speichern, Senden dürfte dann gut umsetzbar sein, das steht dem Executor nicht unbedingt entgegen,
allerdings ist das Empfangen nun gewiss eher eine Aufgabe des sich jeweils darum kümmernden Tasks,
wenn die im Exceutor nicht gleichzeitig laufen, dann kannst du schwerlich von allen Verbindungen zu beliebigen Zeiten Nachrichten empfangen und nebenläufig verarbeiten,

dann ist NetworkThread vielleicht doch besser, typischerweise ist eine solche wichtige Verbindung einen eigenen Thread wert,
das machen glaube ich auch hohe Frameworks wie RMI, J2EE usw.

 

[Taramsis]

Also soetwas?:

private ArrayList<Object> threadList = new ArrayList<Object>();
private Iterator iterrator = threadList.iterator();

public void listen() {
        try {
            serverSkt = new ServerSocket(port);
            ExecutorService theExecutor = Executors.newSingleThreadExecutor();
            alive=true;
            while(alive){
                Socket skt = serverSkt.accept();
                threadList.add(theExecutor.submit(new NetworkThread(skt,this)));

                while(serverFree){
                    sende(iterrator.next());
                }

            }           
        }
        catch (IOException ex) {
        }
}

Was auch immer sende und serverFree seien sollen!

Wäre es über RMI ode JavaEE elegenater/schneller/besser?

 

[SlaterB]

> Was auch immer sende und serverFree seien sollen!
solange du nicht weißt was eine Liste oder ein Iterator ist, solltest du nicht mit komplizierten Sockets + Threads arbeiten,
aber theoretisch alles Schritte in die richtige Richtung, wobei ich das Ziel nur ahnen kann, du kennst es hoffentlich

> Wäre es über RMI ode JavaEE elegenater/schneller/besser?
die Frage drückt leider auch kaum Verständnis aus, inhaltlich aber durchaus angebracht,
auch hier würde ich aber beim einfachen bleiben:
erst lernen was eine Liste + ein Iterator ist,
dann 2 Monate später Sockets kennenlernen,
und dann noch später vielleicht RMI + anderes

 

[Taramsis]

... das sollte nur eine art pseudo Code sein - ist mir schon klar, dass ich nicht den Iterator schicken darf ...

Aber Grundsätzlich brauche ich Threads, da ich ja die Anfragen von verschiedenen Clients mit dem Server bearbeiten will.
->Damit ich weiß wer welche Antwort, mit dem entsprechenden Socket, erhält. (richtig?)

Ich dachte meine SlaveServer können ja ein Flag an dem Master schicken, wenn sie frei sind und nur dann wird das nächste Element aus der Liste bzw. Warteschlange verschickt.
->Ist dann der Exucutor nicht überflüssig?

 

[SlaterB]

> Ist dann der Exucutor nicht überflüssig?
deine Vorgänge sind nicht leicht vorzustellen, 1-2 Sätze helfen nicht unbedingt,
im Extremfall ist es aber für mich denkbar, ja, dass das ganze funktionieren kann ohne Executor, und auch nur mit einem Thread statt mehreren,
nämlich dann wenn alle Verbindungen standardmäßig ruhen und nur nach genauen Ablauf benutzt werden

- Server hält 10 Verbindungen, niemand horcht ob dort Nachrichten kommen
- Server schickt an Socket A etwas, keine Antwort nötig, niemand hört
- Server schickt an Socket B etwas, direkte Antwort erforderlich, Server horcht dort, macht solange nix anderes
- Server erhält Antwort von Socket B, macht danach ...

so könnte es ablaufen

 

[Taramsis]

- Server hält 10 Verbindungen, niemand horcht ob dort Nachrichten kommen
- Server schickt an Socket A etwas, keine Antwort nötig, niemand hört
- Server schickt an Socket B etwas, direkte Antwort erforderlich, Server horcht dort, macht solange nix anderes
- Server erhält Antwort von Socket B, macht danach ...
so könnte es ablaufen

Das wär nicht das was ich will!

Vielleicht kann ein Beispiel mehr Aufschluß geben, was ich mir gedacht habe:

Geg.: 1 Master Server, 2 SlaveServer und 10 Anfragen (die zeitgleich eintreffen)

Im Grunde soll der MasterServer Anfragen von 10 Clients sammeln und dann an die SlaveServer verteilen.

1. MasterServer verwalten die Verfügbarkeit von SlaveServer 1 u. 2
2. erstelle einen Thread für jede Anfrage und speichere es in die Queue
3. schicke Thread 1 (Anfgrage 1 (von Client 1)) an Slave1 (also haben wir einen Socket zu C1 und einen zu S1) -> warte auf Antwort, S1 belegt
4. schicke Thread 2 (Anfgrage 2 (von Client 2)) an Slave2 (also haben wir einen Socket zu C2 und einen zu S2) -> warte auf Antwort, S2 belegt
5. Antwort von Client 2 erhalten (da es z.B. sehr kurz war), S2 frei -> schicke Antwort an C2
6. weiter mit 3. für nächsten Thead in Queue
. . .

Meine Uhrsprüngliche Frage war darauf gerichet, ob man folgendes machen kann (bezüglich Schritt 2):

threadList.add(new NetworkThread(skt,this));

Threads einfach in die Liste speichern - kein "theExecutor.submit" for dem "new"

???:L

 

[SlaterB]

Threads in der Liste gespeichert sind einfach nur in der Liste gespeichert, auf die Ausführung hat das keinen Einfluss,
du musst immer noch überlegen, wer ganz konkret wann die run-Methode ausführt,
entweder den Thread selber starten oder einen Executor bitten irgendwann mal run auszuführen oder anderes

bei deinem Beispiel-Ablauf sehe ich nicht ganz den Zusammenhang etwa zwischen 3. + 4. + 6.
wenn alle Threads für Clients gleichzeitig laufen, wer hält dann 6. davon ab, gleichzeitig zu 3. + 4. stattzufinden?
wenn mehrere Threads gleichzeitig was von Slave1 wollen, könnte es brenzlich werden, evtl. mit Synchronisation zu lösen,

bzw. du schreibst ja "MasterServer verwalten die Verfügbarkeit von SlaveServer 1 u. 2",
wenn da wirklich der Zugriff gemanaged wird, ist das vielleicht die Lösung zu diesem Teilpunkt

wie auch immer, ja, (auch) das klingt erstmal nach einem realistischen Grundkonzept

 

[Taramsis]

bei deinem Beispiel-Ablauf sehe ich nicht ganz den Zusammenhang etwa zwischen 3. + 4. + 6.

Ich habs noch nicht so mit Abläufen ...

4. muss weg und 3. müßte sein:
schicke nächsten Thread aus der Queue an freien SlaveServer!

wenn alle Threads für Clients gleichzeitig laufen, wer hält dann 6. davon ab, gleichzeitig zu 3. + 4. stattzufinden?
wenn mehrere Threads gleichzeitig was von Slave1 wollen, könnte es brenzlich werden, evtl. mit Synchronisation zu lösen,

bzw. du schreibst ja "MasterServer verwalten die Verfügbarkeit von SlaveServer 1 u. 2",
wenn da wirklich der Zugriff gemanaged wird, ist das vielleicht die Lösung zu diesem Teilpunkt

Ich dachte die Threads befinden sich in der Queue (static Queue q = new LinkedList(); ) und einVerwaltungs-Thread stellt nun die Verbindungen zwischen Anfrage-Threads und SlaveServer her.

 

[SlaterB]

gerne, das entspricht ja mehr deinem Anfangspost, wie du es vielleicht die ganze Zeit wolltest,
was war jetzt noch die Frage?

ein Exceutor oder wer auch immer bearbeitet alle Clients, wenn diese dran kommen fragen sie die Slaves, was immer das ist und es geht voran,
ursprünglich wolltest du irgendwas zum Zugriff auf die NetworkThread-Objekte? sind das nicht nur die Clients? brauchen die sich untereinander?

wie auch immer, parallel und unabhängig vom Executor können diese Objekte in eine Liste gesteckt oder sonstwie strukturiert werden, ja

 

[Taramsis]

Ich habe mir folgendes überlegt:

Ich brauche eigentlich nur eine Warteschlange in denen die Threads pausiert und warten bis sie herausgenommen und aktiviert werden.

Also etwas in der Form:

Klasse Network:

public static Queue q = new LinkedList();
 
 public void listen() {
       try {
            serverSkt = new ServerSocket(port);
            alive=true;
            new NetworkSlaveServerStatus(this).startThread();
            
            while(alive){
                Socket skt = serverSkt.accept();
                NetworkThread netThread = new NetworkThread(skt,this);  //hier werden die Anfragen erstellt
                q.add(netThread); 
                
                synchronized(netThread){
                    while(getSlave1Status()|getSlave2Status() == true){
                        q.remove();
                        netThread.proceed(); // geht natürlich nicht soll nur die Idee darstellen
                    }
                }
            }           
        }
 
        catch (IOException ex) {
        }
    }

Klasse NetworkThread:

private boolean threadStatus = true;
 
public void run() {
 
//....
 
synchronized (this){
            while(threadStatus) {
                try {
                    wait();
                } catch (InterruptedException ex) {
                    Logger.getLogger(NetworkThread.class.getName()).log(Level.SEVERE, null, ex);
                }
 
            }
        }
}
 
//....
 
public void pause() {
      threadStatus = true;
    }
 
    public void proceed() {
      threadStatus = false;
      notify();
    }

Wie kann man jetzt einen bestimmten Thread aus in und aus einer Warteschlange einfügen/entnehmen?

 

[SlaterB]

dein Code kann doch schon mehr oder weniger funktionieren, bisschen mehr ausprobieren und dann konkretes nachfragen wäre schön

in der Schleife mit // geht natürlich nicht soll nur die Idee darstellen bzw. in allen Schleifen, die nichts machen und nur einen boolean abfragen immer an Pause durch etwa Thread.sleep(100) denken,
Check 10x pro Sekunde reicht, nicht Millionen Mal pro Sekunde mit 100% CPU-Belastung

der Code mit Bearbeitung der Queue gehört sicherlich allgemein an andere Stelle als bei Socket-accept,
ansonsten durchaus denkbar, herausnehmen bzw. einfach nur per Index wählen, mit wait/ notify + boolean-Flag aktivieren/ pausieren usw.,
klingt für mich nicht falsch

organisatorisch etwas einfacher und sparsamer wäre wahrscheinlich dann doch, auf viele Threads zu verzichten, nur 2-3,
einer zur Ausführung, bearbeitet eine bestimmte Verbindung, wenn diese aus der Queue heraus zugeteilt wird,
dann vielleicht weniger wait/notify/ Flag nötig,
aber wichtiger als Optimierung ist überhaupt erstmal irgendeine Variante zu bauen,

 

[Taramsis]

dann konkretes nachfragen wäre schön

einfach nur per Index wählen, mit wait/ notify + boolean-Flag aktivieren/ pausieren usw.,

Meine konkrete Frage sollte sich auf Zeile 16 u. 17 in der Klasse Netwok beziehen. Wie man das richtig umsetzt?
Meinst du so etwas? Wie würde ich nun den Flag ändern?

synchronized(netThread){
   while(getSlave1Status()|getSlave2Status() == true){
      q.iterator().next().notify();
      q.remove();                        
   }
}

in der Schleife mit // geht natürlich nicht soll nur die Idee darstellen bzw. in allen Schleifen, die nichts machen und nur einen boolean abfragen immer an Pause durch etwa Thread.sleep(100) denken,
Check 10x pro Sekunde reicht, nicht Millionen Mal pro Sekunde mit 100% CPU-Belastung

Wenn ich die besagte Schleife in einen eigenen Thread ausführe, dann sollte dieser Thread auf sleep gesetzt werden? Das verstehe ich nicht ganz! Oder beziest du dich auf das wait() in der Klasse NetworkThread?

public void run(){
//...
   synchronized(netThread){
      while(getSlave1Status()|getSlave2Status() == true){
         q.iterator().next().notify();
         q.remove();   
         Thread.sleep(100);
      }
  }
//...
}

 

[SlaterB]

> Wenn ich die besagte Schleife in einen eigenen Thread ausführe, dann sollte dieser Thread auf sleep gesetzt werden? Das verstehe ich nicht ganz! Oder beziest du dich auf das wait() in der Klasse NetworkThread?

das erstere ist gemeint und die Folge habe ich bereits benannt: der Thread läuft nicht millionenmal pro Sekunde durch die Schleife mit 100% CPU-Belastung obwohl quasi nichts passiert, nur gewartet wird auf Zustandsänderung,
wie das nicht zu versehen ist verstehe ich selber nicht ganz, weiter kann ich es ohne konkrete Frage nicht beschreiben

> Meinst du so etwas? Wie würde ich nun den Flag ändern?

durch Aufruf der Methode?
ich bezweifle dass die beiden Befehle
> q.iterator().next().notify();
> q.remove();
für sich ausreichen, möchte aber für dich auch nichts programmieren

überlege dir alles ganz in Ruhe, beschreibe in deutscher Sprache, in Papierskizzen oder was auch immer dir hilft, was du eigentlich vorhast und was dann konkret umzusetzen ist,
evtl. erstmal auch ohne Schleife und einfach nur drei bestimmte Netzwerkverbindungen a, b, c der Reihe nach sauber bearbeiten

"als erstes wende ich mich a zu und sorge mit folgenden ganz konkreten Aktionen dafür, dass a [von wem auch immer] bearbeitet wird:
Schritt 1: a wird an .. übergeben
Schritt 2: a bekommt folgendes Flag
Schritt 3: a oder wer immer a ausführt muss informiert werden
nebenher: irgendjemand wartet bis irgendwas fertig ist"
usw. usf., ich habe überhaupt kein Interesse, da alles selber durchzugehen, schon viel zu detailliert, das ist alles deine Aufgabe

 

[Taramsis]

durch Aufruf der Methode?

Ich glaube ich sitze auf der Leitung!
Im ArbeiterThread sollen die einzelnen Threads bearbeitet werden.
Wenn ich mehrere Threads in meiner Queue habe und dann ein SlaveServer ist frei, macht der ArbeiterThread folgendes:
1. Rufe nächsten Thread auf
2. entferne es aus der Queue
3. ändere Flag der Queue

zu 3 habe ich in jedem AnfrageThread ein proceed() und pause(): (siehe unten in Der Klasse NetworkThread)
Wie kann man auf ein bestimmtes proceed() aus dem ArbeiterThreadzugreifen?

... , möchte aber für dich auch nichts programmieren

Will ich auch nicht ... Mir ist es nur nicht klar wie ich den Flag bei einem Thread ändern soll, ohne alle zu ändern?;( Oder seh ich etwas komplett falsch!

 

[SlaterB]

wenn du bei einem bestimmten NetworkThread-Objekt proceed() aufrufst, dann wird auch nur dort das Flag geändert,
nur dieser Thread, falls wartend, aufgeweckt, wieso sollten alle betroffen sein?

wenn es um den ArbeiterThread geht, was immer der aktuell für eine Aufgabe hat, da gibt es ja nur einen sofern ich überhaupt grob ahnen kann, worum es geht,
da ist die Situation eine ganz andere, ob der benachrichtigt werden muss ist eine Frage für sich,
aber wenn dann ist es nur der eine, auch da sehe ich keinen Bezug zu 'alle'

was ist die Frage?

----

weiterer Interpretationsversuch:
> Wie kann man auf ein bestimmtes proceed() aus dem ArbeiterThreadzugreifen?

soll die Methode proceed() in NetworkThread auf den zentralen ArbeiterThread zugreifen?
das könnte man als deutliche Frage formulieren, und der ArbeiterThread könnte als Parameter übergeben werden?

 

[Taramsis]

was ist die Frage?

Ich glaub ich hab mich wieder mal nicht präzise ausgedrückt

pseudoCode ArbeiterThread:

synchronized(this){
   while(einer der SlaveServer frei){
      nehme nächsten Thread aus queue
      ändere Flag von diesem Thread
  }
}

Was ich mit "alle" meinte ist:
Wenn ich in ArbeiterThread den nächsten Thread aus der Queue nehme muss ich ja nur proceed() für diesen Thread ausführen!
Woher weiß der Interpreter, dass er gerade den Flag von dem Thread ändern soll, dass ich aus der Queue genommen habe?
Ich denke aus meinem Verständnis (etwas beschränkt ich weiß), dass wenn ich instanz.proceed() ausführe er nicht weiß, dass es das Thread aus:

q.iterator().next().notify(); 
q.remove();

ist!

Verstehst du was ich meine?

 

[SlaterB]

nein, verstehe ich nicht,
ein Objekt wird geändert, andere nicht,
in welcher Liste das Objekt ist, ob es dort vorher, parallel oder später entfernt wird, all das hat keinen Einfluss,

ein Objekt wird geändert oder auch nicht, damit ist alles gesagt

> Woher weiß der Interpreter, dass er gerade den Flag von dem Thread ändern soll, dass ich aus der Queue genommen habe?

du sagst durch deinen Code welches Objekt zu ändern ist, und das wird dann auch geändert,
was könnte da unklar sein? welche Varianten hälst du für denkbar?

 

[Taramsis]

Vielleicht etwa konkreter:

in Network ist folgendes:
Queue: Thread1, ...., ThreadN

in ArbeiterThread soll folgendes gemacht werden:
nehme nächstes Element aus der Queue -> Thread 1
ändere Flag im nächsten Element -> Thread 1

Wenn ich jetzt proceed() ausführe würde doch nicht nur das eine Thread angeprochen sondern alle, die sich in der Queue befinden?

Konkreter die Klasse NetworkWorkerThread erbt von Network und NetworkThread:

public class NetworkWorkerThread implements Runnable{
    
    private Network net;
    private NetworkRequestThread netThread;
    
    public NetworkWorkerThread(NetworkRequestThread netThread, Network net) {
        this.netThread = netThread;
        this.net = net;
    }
    
    public void startThread() {
        new Thread(this).start();
    }
    
    public void run() {
        synchronized(this){
            while(net.getSlave1Status()|net.getSlave2Status() == true){
                net.queueNotifyNext();
                net.queueRemoveNext();
                netThread.proceed();  //<---*              
            }
        }
    }
}

dabei sieht Network wie folgt aus:

public class Network {

    private ServerSocket serverSkt;
    private int port = 4444;
    private boolean alive = false;
    private Window win;
    public static Queue q = new LinkedList();
    public static boolean slaveServer1Status = false;
    public static boolean slaveServer2Status = false;
    

    public Network(Window view) {
        this.win = view;
    }

    public void listen() {
        try {
            serverSkt = new ServerSocket(port);
            alive=true;
            NetworkRequestThread netThread = null;
            new NetworkSlaveServerStatus(this).startThread();
            new NetworkWorkerThread(netThread, this).startThread();
            
            while(alive){
                Socket skt = serverSkt.accept();
                netThread = new NetworkRequestThread(skt,this);
                q.add(netThread); 
            }           
        }

        catch (IOException ex) {
         }
    }
    
            
    public Window getWin() {
        return win;
    }

    public void setWin(Window win) {
        this.win = win;
    }
    
    public void queueNotifyNext() {
        q.iterator().next().notify();
    }

    public void queueRemoveNext() {
        q.remove();  
    }
    
    public boolean getSlave1Status() {
        return slaveServer1Status;
    }

    public void setSlave1Status(boolean slaveServer1Status) {
        Network.slaveServer1Status = slaveServer1Status;
    }
    
    public boolean getSlave2Status() {
        return slaveServer2Status;
    }

    public void setSlave2Status(boolean slaveServer2Status) {
        Network.slaveServer2Status = slaveServer2Status;
    }

}

*Zeile 20: Kann man das so machen oder nicht????:L

 

[SlaterB]

> Wenn ich jetzt proceed() ausführe würde doch nicht nur das eine Thread angeprochen sondern alle, die sich in der Queue befinden?

wie zuvor sehe ich von dir keinerlei Begründung für diese aberwitzige Behauptung,
wenn du in einer Liste 5 Sockets hast und bei einem close() aufrufst, werden dann auch alle geclosed?

-----

> net.queueNotifyNext();
> net.queueRemoveNext();
> netThread.proceed();
hier scheint die netThread-Variable ein komplett andere zu sein als irgendein Element in der Queue,
oder soll da ein Zusammenhang bestehen?

grundsätzlich finde ich deinen Befehl q.iterator().next().notify(); immer merkwürdig,
hole oder ein Objekt heraus, speichere es in der Variablen netThread oder wo auch immer,
und rufe dann an dieser Variablen notify(), proceed() auf und was immer nötig ist

 

[Taramsis]

Reicht es aus dann das Objekt so zu casten?:

public class NetworkWorkerThread implements Runnable{
    
    private Network net;
    private NetworkRequestThread netThread;
    
    public NetworkWorkerThread(NetworkRequestThread netThread, Network net) {
        this.netThread = netThread;
        this.net = net;
    }

    
    
    public void startThread() {
        new Thread(this).start();
    }
    
    public void run() {
        synchronized(this){
            while(net.getSlave1Status()|net.getSlave2Status() == true){
                NetworkRequestThread netThreadtemp; 
                netThreadtemp = (NetworkRequestThread) net.queueRemoveNext();
                netThreadtemp.proceed();                
            }
        }
    }
}

Network:

public Object queueRemoveNext() {
       return q.remove();  
    }

 

[SlaterB]

mit Generics ab Java 1.5 ließe sich der Cast einsparen, so dürften diese Code-Zeilen aber auch funktionieren, ja,