Wie bekomme ich Threadsdazu parallel zu arbeiten?

[mephi]

Hi, ich habe ein kleines Prog dass eine Art Request System auf einem Server darstellen soll..
Ich habe 1 Interface und 3 Klassen die von Thread erben:

Interface MessageObserver: Wird von den 3 Klassen implementiert und liefert nur eine einfache Input methode.

Channel: Gibt Nachrichten an alle anhängenden Teilnehmer weiter. Teilnehmer sind Klassen vom Typ MessageObserver
Filter: Filtert über einen einfachen Regex Nachrichten in 2 Ausgänge.. also true oder false..
Rule: Kann ein beliebiges Verhalten haben. deswegen ist diese klasse bei mir abstract.

Problem: Ich hänge zum Test folgendes aneinander:

Channel 1 -> Channel 2 -> Filter -> Channel 3 -> Rule -> Channel 3

in Rule hab ich eine Klasse geschrieben die von Rule erbt. Sie zählt eine Nachricht. Wenn sie 100 mal aufgetreten ist wird sie weiter geleitet(bisher ins nichts) wenn sie noch nicht 100 mal da war wird sie zurück in Channel 3 geschickt. Läuft also wieder zurück in die Rule.

Im Konstruktor jeder Klasse starte ich mit

new Thread(this).start();

In meiner Testklasse starte ich dann folgenden Test:

public void test(String input) {
		firstChannel.addObserver(secondChannel);
		secondChannel.addObserver(firstFilter);
		firstFilter.addTrueObserver(thirdChannel);
		thirdChannel.addObserver(firstRule);
		firstRule.addBackLoop(thirdChannel);
		firstChannel.input("ab");
		firstChannel.input("bla");
		System.out.println("end");
	}

Jetzt läuft die erste Nachricht 100 mal durch bevor die 2. Nachricht "reingepackt" wird..
Kann mir da jemand helfen?

Achja wenn ich eine run methode implementiert habe in der überprüft wird ob ein input da ist wird wieder erst die 1. nachricht 100 mal durchgearbeitet bevor die 2. dran ist

 

[SlaterB]

da ich nichts verstanden habe kann ich nichts zum Problem beitragen,
nur allgemein Tipps zur megakonfusen Darstellung geben:

> Channel 1 -> Channel 2 -> Filter -> Channel 3 -> Rule -> Channel 3

wen interessiert das?

wenn schon der Übergang von Channel1 (was immer das ist) zu Channel2 (was immer das ist) nicht funktioniert, dann schreibe doch
Channel 1 -> Channel 2
und erwähne den Rest nicht, schon wäre es einfacher

wenn Channel 1 -> Channel 2 klappt, dann erwähne Chanel 1 nicht, sonder nur
Channel 2 -> Filter
oder was auch immer

> firstChannel.addObserver(secondChannel);
> secondChannel.addObserver(firstFilter);
> firstFilter.addTrueObserver(thirdChannel);
> thirdChannel.addObserver(firstRule);

dito


> 100 mal aufgetreten
> 1. nachricht 100 mal durchgearbeitet

was diese Zahl 100 bedeutet ist nicht ersichtlich, aber da du sie zweimal erwähnst scheint sie von dir und nicht zufällig zu sein:
ersetze 100 durch 2 -> einfacher..

 

[mephi]

da ich nichts verstanden habe kann ich nichts zum Problem beitragen,
nur allgemein Tipps zur megakonfusen Darstellung geben:

> Channel 1 -> Channel 2 -> Filter -> Channel 3 -> Rule -> Channel 3

wen interessiert das?

wenn schon der Übergang von Channel1 (was immer das ist) zu Channel2 (was immer das ist) nicht funktioniert, dann schreibe doch
Channel 1 -> Channel 2
und erwähne den Rest nicht, schon wäre es einfacher

wenn Channel 1 -> Channel 2 klappt, dann erwähne Chanel 1 nicht, sonder nur
Channel 2 -> Filter
oder was auch immer

> firstChannel.addObserver(secondChannel);
> secondChannel.addObserver(firstFilter);
> firstFilter.addTrueObserver(thirdChannel);
> thirdChannel.addObserver(firstRule);

dito


> 100 mal aufgetreten
> 1. nachricht 100 mal durchgearbeitet

was diese Zahl 100 bedeutet ist nicht ersichtlich, aber da du sie zweimal erwähnst scheint sie von dir und nicht zufällig zu sein:
ersetze 100 durch 2 -> einfacher..

das ist einfach nur ein test. die nachricht soll möglichst lange in einer art schleife laufen damit für mich ersichtlich wird ob die 2. nachricht auch wirklich erst nach diesen x-durchgängen abgearbeitet wird oder ob es einfach zufall ist.

und ich fand es schon wichtig die komplette reihehinzuschreiben um zu zeigen dass der erste channel eigentlich frei sein solllte für die 2. nachricht. und nicht irgendwie belegt ist..

wenns hilft kann ich ja den code posten

 

[mephi]

ok ich brech mal mein problem ganz stark herunter...
hab nun alles umgebaut.

ich hab in jeder klasse nun eine run methode..
ich wollte dort einfach überprüfen ob eine variable "boolean send" auf true ist wenn das der fall ist wird die an einen anderen thread übergeben und der aktuelle thread soll auf wait gesetzt werden.. dann wirfs mir aber exceptions... sieht folgendermaßen aus

public void run() {
		while(true) {
			if(send) {
				notifyObs(input);
				try {
					wait();
				} catch (InterruptedException e) {
					// TODO Auto-generated catch block
					e.printStackTrace();
				}
			}
		}
	}

dann kommt eine IllegalMonitorStateException..
wie lös ich das anders?

 

[SlaterB]

wait ist eine Synchronisations-Opeation, nicht zu verwechseln mit Thread.sleep() oder so,

wait kann man nur in einem synchronisations-Block aufrufen, wenn man exklusiven Zugriff auf den Monitor hat:

synchronized(this) {
wait();
}

aber irgendjemand muss den Thread dann wieder wecken..

 

[mephi]

edit:
ich geb das vorerst auf..

 

[mephi]

jetzt hab ichs doch nochmal versucht... ich komm da einfach nicht klar.... vielleicht sehr ihr ja schon richtig grobe fehler in meinem umgang mit threads..

letzter output war folgender:

Channel 2: Test
Channel 2: warte
Channel 1: Test
Channel 1: warte
Channel 1: Bekomme Input: ab
end

import java.util.ArrayList;
import java.util.LinkedList;
import java.util.Observable;
import java.util.Observer;
import java.util.Queue;


public class Channel extends Observable implements MessageObserver, Runnable, Observer{
	
	private String name = null;
	
//	Queue<String> messageQueue = new LinkedList<String>();
	
	ArrayList<MessageObserver> observerList = new ArrayList<MessageObserver>();
	
	public Channel(){
		new Thread(this).start();
	}
	
	public Channel(String name){
		setName(name);
		new Thread(this).start();
	}
	
	public void setName(String name) {
		this.name = name;
	}
	public String getName() {
		return name;
	}
	
	public void addMObs(MessageObserver s) {
		observerList.add(s);
		addObserver((Observer)s);
	}

	public void notifyObs(String input) {
		for (int i=0; i<observerList.size();i++ ) {
			observerList.get(i).input(input);
		}
		synchronized (this) {
			notifyObservers();
		}
	}
	
	synchronized public void input(String s) {
		System.out.println(getName()+": Bekomme Input: "+s);
		messageQueue.add(s);
	}
	public void run() {
		while(true) {
			System.out.println(getName()+": Test");
			while(!(messageQueue.isEmpty())) {
				String message = messageQueue.poll();
				System.out.println(getName()+": Verarbeite Nachricht: "+message);
				notifyObs(message);
			}
			try {
				synchronized (this) {
					System.out.println(getName()+": warte");
					wait();
				}
			} catch (InterruptedException e) {
				// TODO Auto-generated catch block
				e.printStackTrace();
			}
		}	
	}
	public void update(Observable arg0, Object arg1) {
		// TODO Auto-generated method stub
		
	}
	
	
	
}

import java.util.LinkedList;
import java.util.Queue;



public interface MessageObserver{
	
	Queue<String> messageQueue = new LinkedList<String>();
	public void input(String s);
	public void setName(String name);

}

import java.util.Observable;


public class TestClass extends Observable implements Runnable {
	
	Channel firstChannel = new Channel("Channel 1");
	Channel secondChannel = new Channel("Channel 2");
//	Channel thirdChannel = new Channel("Channel 3");
//	Filter firstFilter = new Filter("a*b", "Filter 1");
//	CounterRule firstRule = new CounterRule();
	
	public void test(String input) {
		this.addObserver(firstChannel);		
		firstChannel.addMObs(secondChannel);
//		secondChannel.addMObs(firstFilter);
//		firstFilter.addTrueMObs(thirdChannel);
//		thirdChannel.addObserver(firstRule);
//		firstRule.addBackLoop(thirdChannel);
		firstChannel.input(input);
		synchronized (this){
			this.notifyObservers();
		}
//		firstChannel.input("bla");
		System.out.println("end");
	}
	
	public void run() {
		test("ab");
	}
	
	public TestClass() {
		new Thread(this).start();
	}
	
	public static void main(String args[]) {
		TestClass testClass = new TestClass();

	}
}

 

[SlaterB]

Channel 2: Test
Channel 2: warte
Channel 1: Test
Channel 1: warte
Channel 1: Bekomme Input: ab
end

was ist denn der erwartete Output,

oder ganz einfach immer noch: worum gehts hier?

 

[mephi]

ich habs doch erklärt.. aber scheint wohl zu kompliziert zu sein sich da direkt "einzudenken"

aber wenn man den code anschaut sollte doch klar sein was passieren soll.. nachdem die testklasse den input übergeben und notifyObservers() aufgerufen hat sollte der "Channel 1" aufwachen und bemerken dass da was in seiner MessageQueue liegt. aber er wacht wohl nicht auf....

sonst müsste nach dem "Bekomme Input: ab" noch weitere consolen ausgaben kommen

das ganze soll ein regel-filter-framework sein. ka ob dir das nun weiterhilft ^^

 

[SlaterB]

mit deinem ewig langen Code macht es eben keinen Spass,
hab mir nun die Mühe gemacht, das zu kapieren, allgemein ist das aber nicht optimal,

schaue dir mein Beispiel an, in dem das Problem ähnlich auftritt:

public class Channel
    implements Runnable
{

    public Channel()
    {
        new Thread(this).start();
    }

    synchronized public void input(String s)
    {
        System.out.println("Channel bekommt Input: " + s);
        // synchronized (this)
        // {
        // notify();
        // }
    }


    public void run()
    {
        System.out.println("Channel: Start Run");
        while (true)
        {
            try
            {
                synchronized (this)
                {
                    System.out.println("Channel: warte");
                    wait();
                    System.out.println("Channel: aufgewacht");
                }
            }
            catch (InterruptedException e)
            {
                e.printStackTrace();
            }
        }
    }


    public static void main(String[] args)
        throws Exception

    {
        Channel c = new Channel();
        Thread.sleep(3000);
        c.input("Hello Channel");
        synchronized (Channel.class)
        {
            Channel.class.notify();
        }
    }

viel weniger Zeilen und diese wenigen auch noch so gut wie leer,
leicht zu überblicken,

was interessiert der zweite Channel wenn du nicht mal den ersten aufwecken kannst?,
wozu MessageObserver usw?
ich bin mal besonders streng damit du vielleicht das System erkennst

also: Problem: du rufst notify an einem GANZ ANDEREN Objekt auf, warum sollte das Channel1 aufwecken?
wenn dann würde es ja beide Channel sowie alle anderen wartenen Threads im Programm aufwecken,
oder wie willst du die unterscheiden?
du musst das notify natürlich an dem Objekt aufrufen, an dem der Thread wartet,
in meinem Beispiel z.B. am c-Objekt in der main (während selbst Channel.class ein GANZ ANDERES Objekt ist),
oder du machst das notify in der input-Operation

 

[mephi]

ooooook, danke erstmal. das notify klappt nun.. dachte nur als ich die api gelesen habe dass notify irgendeinen thread weckt.

so nun hab ich n anderes problem(man da prog ich ca 1 jahr java und komm mir vor wie ein anfänger...)

in der schleife

while(!(messageQueue.isEmpty())) {
				String message = messageQueue.poll();
				System.out.println(getName()+": Verarbeite Nachricht: "+message);
				notifyObs(message);
			}

• 
  prüf ich in der messageQueue ob eine nachricht da ist. die Queue wird in meinem Interface definiert.
  kann es sein dass da nun alle klassen die das interface implementieren eine einzige queue benutzen?
  denn die 2 channel haben dann total sinnlos immer wieder die eine nachricht verarbeitet bis channel2 die ins "leere" geschickt hat. das scheint mir nämlich so als ob channel1 die nachricht in die input methode von channel 2 schickt und diese methode packt die nachricht in die queue.. und dann durch zufall greift wieder channel 1 auf die queue zu und schickt die nachricht wieder an channel 2.. bis channel2 schneller war... aber vll bin ich nur total neben der kappe
  
  definiere ich die Queue im Channel und NICHT im interface, wachen die threads nicht mehr auf.. KA was das damit zutun hat
  
  definiere ich die queue im interface UND im channel klappt alles wunderbar.. ich versteh nur bahnhof..
  

 

[SlaterB]

> kann es sein dass da nun alle klassen die das interface implementieren eine einzige queue benutzen?

ja, in etwa ein statische Variable

> definiere ich die Queue im Channel und NICHT im interface, wachen die threads nicht mehr auf.. KA was das damit zutun hat

versuche es selber herauszufinden indem du zumindest ein Beispiel konstruierst, in dem das Problem auftritt,
verwende idealerweise

synchronized (this)
{
System.out.println("Channel: warte");
wait();
System.out.println("Channel: aufgewacht");
}

vielleicht klappt ja das aufwachen ganz normal, nur DANACH apssiert irgendwas komisches

 

[mephi]

das hab ich benutzt. sie wachen nicht auf.
kapiere nicht ganz was das mit der queue zutun hat. das notify sollte doch absolut garnicht damit im zusammenhang stehn

 

[SlaterB]

tja, bei mir ist alles normal,
schlimm genug dass ich mir den Test selber zusammenkopieren musste..

poste ein Programm das nicht geht, sonst gehts nicht weiter