Inter-Thread-Kommunikation

[Wolfram]

2 Threads A und B. A kommt von Main(), B extends Thread und wurde ge-start()-et.

Die sollen miteinander reden, und während der andere was tut, jeweils blockieren. A soll einen Wert an B schicken, B damit eine Berechnung anstellen, und das Ergebnis an A zurück schicken. Danach erst soll tatsächlich parallel gearbeitet werden, aber trotzdem noch miteinder kommuniziert werden, was ich im Beispiel nicht dargestellt habe.

Also nochmal kurz:

B soll nach dem Start warten, bis etwas kommt, und dann erst losrechnen.
A soll auf das Ergebnis warten, bevor weiter gemacht wird.

Die unten gepastete Lösung tut, was ich erwarte, hat aber Warzen und Pickel bekommen, und ich denke, dass ich daher grob was falsch gemacht haben muss. Die Warzen und Pickel wären:

* Ein notify() muss in ein synchronized auf dem Objekt eingeklammert werden, das den Thread darstellt, der die Nachricht erhält. Ein wait() entsprechend in ein synchronized auf dem eigenen Objekt. Das sieht richtig aus, jedoch ist es nicht möglich, auf irgendwas anderem zu synchronized-en, man bekommt sonst eine IllegalMonitorStateException. Man muss auf dem Objekt synchronisieren, auf dem ge-wait()-et wird. (1) Ich habe keinerlei Doku gefunden, die das hergibt. Ich erwarte eigentlich, auf new Object() synchronisieren zu können. (2) Wenn das zwingend so wäre, könnte wait() das schließlich aus selbst tun.

* while(!t.initialized()); ist, um zu verhindern, dass Thread A das notify() los schickt, bevor der andere überhaupt die Chance hatte, zu wait()-en. Das ist eklig und tut nur in 99.9999% der Fälle, was man denkt: Zwischen init=true und wait() kann der thread ja schließlich unterbrochen werden.

Wer also weiß den einzig wahren Weg, das zu implementieren?

Grüße,
Wolfram

package testmultithread;

class thethread extends Thread {
    
    String msgIn=null;
    String msgOut=null;

    Object other=null;
    boolean init=false;
            
    boolean msgTerminate=false;
    
    // Communication
    public void msgSend(String m)  { msgIn=m; }    
    public String getMsg()  { return msgOut; }
    public void terminate()  { msgTerminate = true; }

    public boolean initialized()  { return init; }

    // Constructor: need to know the caller (to notify() it)
    thethread(Object other) { 
        this.other=other;
    }
    
    @Override
    public void run() {
        while (!msgTerminate) {
            
            System.out.println("thethread: wait");
            init=true;
            
            try {
                synchronized(this) {
                    wait();
                }
            } catch (InterruptedException e) {}

            if (msgIn!=null) {
                System.out.println(msgIn);
                msgIn=null;
                msgOut="msgOut";
                System.out.println("thethread: notify");
                synchronized(other) {
                    other.notify();
                }
            }
        }
    }
}


class OtherMain {
    
    public void go() {
        thethread t = new thethread(this);

        t.start();

        while (!t.initialized());

        t.msgSend("Yo");
        System.out.println("Main: notify");

        synchronized(t) {
            t.notify();
        }
        
        synchronized(this) {
            try {
                System.out.println("Main: wait");
                wait();
            } catch (InterruptedException e) {}
        }
        
        System.out.println(t.getMsg());
        
        t.terminate();
        synchronized(t) {
            t.notify();
        }
    }
}

public class Main {
    
    public static void main(String[] args) {
        new OtherMain().go();
    }

}

[/code]

 

[Pappenheimer++]

Für deine Zwecke wäre u.U. java.util.concurrent.LinkedBlockingQueue angebracht. Da kannst du einfach eine dequeende Methode aufrufen, die solange blockiert, bis etwas zum Entnehmen vorhanden ist. Da kannst du ja den ersten Aufruf als Flag nehmen, um anzuzeigen, wann der Thread "starten darf" und ansonsten eben deine "normale" Kommunikation darüber regeln.

 

[SlaterB]

> Zwischen init=true und wait() kann der thread ja schließlich unterbrochen werden.

wenn es aber in der Form
synchronized(this) {
init=true;
wait();
}
steht, dann stört das Unterbrechen nicht

--------

> while(!t.initialized()); [..] ist eklig

falls du das generell meinst, dann stimme ich dir zu,
lasse doch die Threads in der richtigen Reihenfolge starten,
oder warte an einem dritten Objekt oder setze wenigesten Thread.sleep(10) in die Schleife

 

[Wolfram]

Für deine Zwecke wäre u.U. java.util.concurrent.LinkedBlockingQueue angebracht. Da kannst du einfach eine dequeende Methode aufrufen, die solange blockiert, bis etwas zum Entnehmen vorhanden ist. Da kannst du ja den ersten Aufruf als Flag nehmen, um anzuzeigen, wann der Thread "starten darf" und ansonsten eben deine "normale" Kommunikation darüber regeln.

Yup, das geht *viel* besser. Bevor mir das Gehirn 'Mars Attacks!'-mäßig platzt, wenn ich versuche, die Anforderung auf kritische Abschnitte abzubilden, meine ich. Danke!

package javaapplication1;

import java.util.concurrent.LinkedBlockingQueue;


// Definition für die Kommandoliste.
class ITCCommand {

    static final int TERMINATE = 0;
    static final int MSG = 1;

    public int cmdnr;
    public String param;

    ITCCommand(int cmdnr, String param) {
        this.cmdnr = cmdnr;
        this.param = param;
    }
}

class thethread extends Thread {

    OtherMain other;
    
    LinkedBlockingQueue<ITCCommand> commands =
            new LinkedBlockingQueue<ITCCommand>();

    thethread(OtherMain other) {
        this.other = other;
    }
    
    // Kommando an diesen Thread schicken.
    public void command(ITCCommand cmd) {
        try {
            commands.put(cmd);
        } catch (InterruptedException e) {}
    }

    @Override
    public void run() {

        // Schleife zm Abarbeiten von Kommandos
        out:
        while (true) {
            ITCCommand cmd=null;

            try {
                cmd = commands.take();  // auf Kommandos warten
            } catch (InterruptedException e) {}

            switch (cmd.cmdnr) {
                case ITCCommand.TERMINATE:
                    break out;
                case ITCCommand.MSG:  // Ausgeben und Kommando zurück schicken
                    System.out.println(cmd.param);
                    cmd = new ITCCommand(ITCCommand.MSG, "Man");
                    other.command(cmd);
                    break;
            }
        }
    }
}

class OtherMain {

    LinkedBlockingQueue<ITCCommand> commands =
            new LinkedBlockingQueue<ITCCommand>();
    
    // Kommando an diesen Thread senden.
    public void command(ITCCommand cmd) {
        try {
            commands.put(cmd);
        } catch (InterruptedException e) {}
    }
    
    public void go() {
        ITCCommand cmd = null;
        
        // neuer thread, gleich ein Kommando senden
        thethread t = new thethread(this);
        t.start();
        t.command(new ITCCommand(ITCCommand.MSG, "Yo"));
        
        // auf die Antwort warten und ausgeben
        try {
            cmd = commands.take();   // blockiert
        } catch (InterruptedException e) {}
        System.out.println(cmd.param);

        // anderen Thread auch beenden
        t.command(new ITCCommand(ITCCommand.TERMINATE, ""));
    }
}

public class Main {

    public static void main(String[] args) {
        new OtherMain().go();
    }
}