Alle Threads .notify()

[ThreadProb]

Tag,

hab ein kleines Problem mit Threads.

Mein HauptThread startet mehrere "Unter-Threads" und steuert sie.
Das ganze habe ich so implementiert:

public void startThreads() {
    threads = new Thread[10];

    for(int i = 0; i < threads.length; i++) {
        threads[i] = new Thread(MyThread());
        threads[i].start();
    }

    for(int i = 0; i < threads.length; i++) {
        try {
            threads[i].join();
        } catch (InterruptedException e) {
            threads[i].interrupt();
        }
    }
}

// bei pauseThreads setze ich einfach ein boolean auf true und in meinen
// unterthreads check ich dieses boolean und mach:
// synchronized(this) {
//    while(isWaiting) {
//         try {
//             wait();
//         } catch(InterruptedException e) {
//             currentThread.interrupt();
//         }
//     }

public void continueThreads() {
    isWaiting = false;

    for(int i = 0; i < threads.length; i++) {
        synchronized(threads[i] {
            threads[i].notify();
        }
     }
}

public void stopThreads() {
    for(int i = 0; i < threads.length; i++) {
        synchronized(threads[i] {
            threads[i].interrupt();
        }
     }
}

Zu meinen Problemen:

- Das Pausieren klappt, allerdings werden nicht alle Threads wieder fortgesetzt (nur ein paar).
- stopThreads() funktioniert überhaupt nicht, das interrupt() interessiert meine Threads nicht ^^

Hab ich was falsch implementiert?

 

[ThreadProb]

Ich poste mal den original-Code beider Klassen, nachdem ich sie nun noch ein wenig abgeändert habe:

public class Controller extends AController {
	/**
	 * 
	 */
	private Helper helper = null;
	/**
	 * 
	 */
	private Queue<URL> queue= null;
	/**
	 * 
	 */
	private int amount = 0; 
	/**
	 * 
	 */
	private String path = null;
	/**
	 * 
	 */
	private SimpleDateFormat sdf = new SimpleDateFormat("dd.MM.yyy_HH.mm");
	/**
	 * 
	 */
	private Thread[] threads = null;
	
	/**
	 * 
	 * @param view
	 * @param mode
	 * @param amount
	 */
	public Controller(AView view, int amount) {
		this.view = view;
		this.model = new Model();
		this.model.openLog();
		this.helper = new HelperCrawler();
		this.amount = amount;
	}
	
	/**
	 * 
	 */
	public void run() {
		queue = model.getQueue();
		
		path = (model.getKey(Constants.INPUT) + "/" + sdf.format(new Date()));
		helper.createDirectory(path);
		
		startProcess();
		shutDown();
	}
	
	/**
	 * 
	 */
	@Override
	public void startProcess() {
		isStarted = true;
		isWaiting = false;
		isStopped = false;
		
		view.startState();
		
		threads = new Thread[listOfUrls.size()];
		
		for(int i = 0; i < threads.length; i++) {
			threads[i] = new Thread(new Worker(this, queue.poll(), amount, path));
			threads[i].start();
		}
		
		for(int i = 0; i < threads.length; i++) {
			try {
				threads[i].join();
			} catch (InterruptedException e) {
				threads[i].interrupt();
			}
		}
	}
	
	/**
	 * 
	 */
	@Override
	public void pauseProcess() {
		isWaiting = true;
		
		for(int i = 0; i < threads.length; i++) {
			synchronized(threads[i]) {
				try {
					threads[i].wait();
				} catch (InterruptedException e) {
					threads[i].interrupt();
				}
			}
		}
		
		view.pauseState();
	}
	
	/**
	 * 
	 */
	@Override
	public void continueProcess() {
		isWaiting = false;
		
		for(int i = 0; i < threads.length; i++) {
			synchronized(threads[i]) {
				threads[i].notify();
			}
		}
		
		view.continueState();
	}

	/**
	 * 
	 */
	@Override
	public void stopProcess() {
		isWaiting = false;
		isStopped = true;
		
		for(int i = 0; i < threads.length; i++) {
			synchronized(threads[i]) {
				threads[i].interrupt();
			}
		}
		
		isStarted = false;
	}

	/**
	 * 
	 */
	public synchronized void registerObserver(IObserver observer, int identifier) {
		if(identifier == Constants.DOCUMENT_IDENTIFIER) {
			observersDocs.add(observer);
		} else if(identifier == Constants.MESSAGES_IDENTIFIER) {
			observersMessages.add(observer);
		} 
	}

	/**
	 * 
	 */
	public synchronized void removeObserver(IObserver observer, int identifier) {
		if(identifier == Constants.DOCUMENT_IDENTIFIER) {
			observersDocs.remove(observer);
		} else if(identifier == Constants.MESSAGES_IDENTIFIER) {
			observersMessages.remove(observer);
		} 
	}

	/**
	 * 
	 */
	public synchronized void notifyObservers(String message, int identifier, int logLevel) {
		if(identifier == Constants.DOCUMENT_IDENTIFIER) {
			for(int i = 0; i < observersDocs.size(); i++) {
				observersDocs.get(i).update(message, 0);
			}
		} else if(identifier == Constants.MESSAGES_IDENTIFIER) {
			for(int i = 0; i < observersMessages.size(); i++) {
				observersMessages.get(i).update(message, logLevel);
			}
		} 
	}
}

public class CrawlerWorker extends Thread {
	/**
	 * 
	 */
	private AController controller = null;
	/**
	 * 
	 */
	private Helper helper = null;
	/**
	 * 
	 */
	private int amount = 0; 
	/**
	 * 
	 */
	private Queue<URL> queue = null;
	/**
	 * 
	 */
	private URLConnection urlConnection = null;
	/**
	 * 
	 */
	private URL url = null;
	
	
	/**
	 * 
	 * @param controller
	 */
	public CrawlerWorker(AController controller, URL url, int amount, String path) {
		this.controller = controller;
		this.url = url;
		this.amount = amount;
		this.helper = new Helper();
		this.uriQueue = new LinkedList<URL>();
		this.path= path;
	}
	
	/**
	 * 
	 */
	public void run() {
		path = (path + "/" + url.getHost());
		helper.createDirectory(path);
		
		if(check()) {
			grab();
		} 
			
		int counter = 1;
		
		while(!controller.getIsStopped()) {
			if(counter == amount) {
				break;
			} else {
				url = queue.poll();
					
				if(check()) {
					grab();
				} 
			}	
		}
	}
	
	/**
	 * 
	 * @param url
	 * @return
	 */
	private boolean check() {
		// some logic

		return true;
	}
	
	/**
	 * 
	 * @param url
	 */
	private void grab() {
		// some logic
	}
}

So wie es jetzt ist, funktioniert auch das pausieren nicht. sobald ich pausiere
hängt sich meine grafische oberfläche auf und es passiert nichts mehr ^^

 

[Marc T.]

Vielleicht genauere Beschreibung 🙂

 

[ThreadProb]

Das interrupten funktioniert doch, hatte mich geirrt, da der Thread natürlich
nicht sofort unterbrochen wird.

Allerdings passiert nach dem ersten .wait() irgendwas, sodass nichts mehr
weiter geht. Die GUI friert ein und es passiert nichts weiter.

Ist der Aufruf von threads.wait() so nicht legitim?

 

[Attila]

Worker[] workers = new Worker[3];
for (int i=0;i<workers.length; i++){
    workers[i] = new Worker();
    Thread thread = new Thread(workers[i]);
    thread.setDaemon(true);
    thread.start();
}
//...

// pause:
for (int i=0;i<workers.length; i++){
    workers[i].stopThread();
}
//...

// continue:
for (int i=0;i<workers.length; i++){
    workers[i].contThread();
}

public class Worker implements Runnable {
    private boolean isStopped = false;
    public synchronized void stopThread(){
        isStopped = true;
    }
    public synchronized void contThread(){
        isStopped = false;
        notify();
    }
    private synchronized void isStoppedThread(){
        if ( isStopped ){
            try {
                wait();
            } catch (InterruptedException ex) { }
        }
    }
    public void run() {
        while ( true ) {
              isStoppedThread();
              System.out.println(Thread.currentThread().getName()+" "+new Date());
              try {
                    Thread.sleep(500);
              } catch (InterruptedException ex) { }
        }
    }
}

 

[bygones]

Das interrupten funktioniert doch, hatte mich geirrt, da der Thread natürlich
nicht sofort unterbrochen wird.

Allerdings passiert nach dem ersten .wait() irgendwas, sodass nichts mehr
weiter geht. Die GUI friert ein und es passiert nichts weiter.

Ist der Aufruf von threads.wait() so nicht legitim?

du legst den EDT lahm.

wenn du UI maessig unterwegs bist schau dir mal die Klasse SwingWorker an, die hilft.

 

[ThreadProb]

Kann möglich sein aber wieso? ^^
Alles was ich tue ist doch nur meinen Worker-Thread zu stoppen.
Natürlich schickt dieser Worker über das Observer-Pattern Nachrichten
an die GUI aber nach dem pausieren werden doch keine Nachrichten
mehr geschickt. Und die Nachrichten werden sowieso schon in einem
Swing-Worker in der GUI empfangen, da die GUI sehr sehr viele
Nachrichten bekommt.

 

[ThreadProb]

Also ich rufe über das observer pattern von meinem model aus die Methode
Update meiner GUI(observer) auf und darin mache ich folgendes:

public void update(String message) {
    final String temp = message;

    SwingUtilities.invokeLater(new Runnable() {
        public void run() {
            myTextArea.append(temp);
        }
    });
}

Das notifyObservers(String message) des models rufe ich in meinen workern auf.

Und wie gesagt. Wenn ich meine Worker jetzt mit wait() pausieren möchte friert die UI ein.

sorry wenn Fehler drin sind, hab das hier aufm Smartphone im Zug geschrieben 🙂

 

[ThradProb]

Hier ein Ausführbares "Mini-Programm" (Geschrieben auf dem Smartphone, verzeit bitte
kleien Fehler und die fehlenden einrückungen).

Es hat meiner Meinung nach nichts mit dem EDT zu tun, denn wenn man
den entsprechenden Code auskommentiert bleibt der Fehler vorhanden.

Egal ob Pause oder Stop, die GUI friert ein:

public interface IObserver {
 public void update(String message);
 public IObserverable getObserverable();
 public void setObserverable(IObserverable observerable);
}

public interface IObserverable {
 public void registerObserver(IObserver observer);
 public void removeObserver(IObserver observer);
 public void notifyObservers(String message);
}

public class View extends JFrame implements ActionListener, IObserver {
 private Controller controller = null;
 private IObserverable observerable = null;
 private JPanel panel = new JPanel();
 private JTextArea textArea = new JTextArea();
 private JButton start = new JButton("start");
 private JButton pause = new JButton("pause");
 private JButton cont = new JButton("continue");
 private JButton stop = new JButton("stop");

 public View() {
 this.setDefaultCloseOperation(JFrame.EXIT_ON_CLOSE);
 this.setMinimumSize(new Dimension(400, 400));
 this.setContentPane(initPane());
 this.setVisible(true);
 }
 private JPanel initPane() {
 JScrollPane scrollPane = new JScrollPane(textArea);
 scrollPane.setPreferredSize(new Dimension(300, 300));

 textArea.setEditable(false);

 initButtons();

 panel.add(scrollPane);
 panel.add(start);
 panel.add(pause);
 panel.add(cont);
 panel.add(stop);

 return Panel
 }

 private void initButtons() {
 start.addActionListener(this);
 pause.addActionListener(this);
 cont.addActionListener(this);
 stop.addActionListener(this);
 }

 public void actionPerformed(ActionEvent e) {
 Object source = e.getSource();

 if(source.equals(start)) {
 this.controller = new Controller(this);
 controller.start();
 } else if(source.equals(pause)) {
 controller.pauseProcess();
 } else if(source.equals(cont)) {
 controller.continueProcess();
 } else if(source.equals(stop)) {
 controller.stopProcess();
 } 
}

 public void update(String message) {
 final String temp = message;
 SwingUtilities.invokeLater(new Runnable() {
 public void run() {
 textArea.append(temp + "\n");
 }
 }
 }

 public IObserverable getObserverable() {}
 public void setObserverable(IObserverable observerable) {}
}

public class Model implements IObserverable {
 protected ArrayList<IObserver> observers = new ArrayList<IObserver>();

 public synchronized void registerObserver(IObserver observer) {
 observers.add(observer);
 }

 public synchronized void removeObserver(IObserver observer) {
 observers.remove(observer);
 }

 public synchronized void notifyObservers(String message) {
 for(int i = 0; i < observers.size(); i++) {
 observers.get(i).update(message);
 }
 }
}

public class Controller extends Thread {
 private Thread[] threads = null;
 private View view = null;
 public Model model = null;

 public Controller(View view) {
 this.view = view;
 this.model = new Model();

 model.registerObserver(view);
 }

 public void run() {
 model.notifyObservers("Initalize", 0);
 startProcess();
 }

 public void startProcess() {
 model.notifyObservers("Initalize", 0);

 threads = new Thread[10];

 for(int i = 0; i < threads.length; i++) {
 threads[i] = new Thread(new Worker(this, i));
 threads[i].start();
 }

 for(int i = 0; i < threads.length; i++) {
 try {
 threads[i].join();
 } catch(InterruptedException e) {
 threads[i].interrupt();
 }
 }

 }

 public void pauseProcess() {
 model.notifyObservers("Initalize", 0);
 for(int i = 0; i < threads.length; i++) {
 synchronized(threads[i]) {
 try {
 threads[i].wait();
 } catch(InterruptedException e) {
 threads[i].interrupt();
 } 
 }
 }

 public void contProcess() {
 model.notifyObservers("Initalize", 0);
 for(int i = 0; i < threads.length; i++) {
 synchronized(threads[i]) {
 threads[i].notify();
 }
 }
 }

 public void stopProcess() {
 model.notifyObservers("Initalize", 0);
 for(int i = 0; i < threads.length; i++) {
 synchronized(threads[i]) {
 threads[i].interrupt();
 }
 }
 }

public class Worker extends Thread {
private Controller controller = null;
private int number = 0;

public Worker(Controller controller, int number) {
this.controller = controller;
this.number = number;
}

public void run() {
while(!isInterrupted()) {
controller.model.notifyObservers("Running: " + number, 0);
try {
sleep(1000);
} catch(InterruptedException e) {
currentThread().interrupt();
}
}
}
}

Hoffe es sind nicht zu viele Fehler drin aber rein von der Struktur und dem
Afubau müsste es ausführbar sein und das pausieren und Stoppen NICHT funtkionieren.

Ich habe keine Ahnung wieso nicht. Bitte helft mir 🙂

 

[Attila]

Ist der Aufruf von threads.wait() so nicht legitim?

Ja, es ist nicht korrekt.

 

[Michael...]

hab's noch nicht ganz überblickt, aber das Vorgehen erscheint mir falsch.
Ausserdem wird m.M. notify/wait falsch eingesetzt.
Der Aufruf von wait() an einem belieben Objekt, also auch an einem Thread Objekt, bewirkt nicht, das dieses Thread Objekt wartet, sondern dass der aufrufende Thread auf ein notify... dieses Objekts wartet. Und der aufrufende Thread so - wenn ich das richtig interpretiere - in diesem Fall nicht warten/bockieren?

 

[Attila]

Ein kleines Beispiel zu diesem Thema:

import java.awt.BorderLayout;
import java.awt.event.ActionEvent;
import java.awt.event.ActionListener;
import java.util.Date;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import javax.swing.JButton;
import javax.swing.JFrame;
import javax.swing.JLabel;
import javax.swing.JPanel;
import javax.swing.JScrollPane;
import javax.swing.JTextArea;
import javax.swing.SwingUtilities;

public class TestThread extends JFrame{
    public final int INIT     = 0;
    public final int STARTED  = 1;
    public final int PAUSED   = 2;
    public final int CONT     = 3;
    public final int PROGRESS = 4;
    private int buttonStatus = INIT;
    private int numWorkerThreads;
    private PauseProgress pauseProgress;
    private Worker[] workers;
    private Thread[] threads;
    private JButton startButton = new JButton("start");
    private JButton pauseButton = new JButton("pause");
    private JButton contButton  = new JButton("continue");
    private JButton stopButton  = new JButton("stop");
    private JTextArea txtA;
    private JLabel statusLabel;

    public TestThread(int numWorkerThreads){
        this.numWorkerThreads = numWorkerThreads;
        initComponents();

        setDefaultCloseOperation(JFrame.EXIT_ON_CLOSE);
        setLocationByPlatform(true);
        pack();
        setVisible(true);
    }
    private void setButtonStatus(int status){
        buttonStatus = status;
        startButton.setEnabled(( buttonStatus == INIT ? true : false ));
        pauseButton.setEnabled((( buttonStatus == STARTED || buttonStatus == CONT ) ? true : false ));
        contButton.setEnabled(( buttonStatus == PAUSED ? true : false ));
        stopButton.setEnabled((( buttonStatus == STARTED || buttonStatus == CONT ) ? true : false ));
    }
    private void initComponents(){
        startButton.addActionListener(new ActionListener(){
            public void actionPerformed(ActionEvent e) {
                setButtonStatus(PROGRESS);
                workers = new Worker[numWorkerThreads];
                threads = new Thread[numWorkerThreads];
                for ( int i=0; i<numWorkerThreads; i++ ){
                    workers[i] = new Worker();
                    threads[i] = new Thread(workers[i]);
                    threads[i].start();
                }
                setButtonStatus(STARTED);
                setStatusLabel("Threads run");
            }
        });
        pauseButton.addActionListener(new ActionListener(){
            public void actionPerformed(ActionEvent e) {
                setButtonStatus(PROGRESS);
                setStatusLabel("Paused in progress ...");
                pauseProgress = new PauseProgress();
                new Thread(pauseProgress).start();
            }
        });
        contButton.addActionListener(new ActionListener(){
            public void actionPerformed(ActionEvent e) {
                setButtonStatus(PROGRESS);
                for ( int i=0; i<numWorkerThreads; i++ ){
                    workers[i].contThread();
                }
                setButtonStatus(CONT);
                setStatusLabel("Continue");
            }
        });
        stopButton.addActionListener(new ActionListener(){
            public void actionPerformed(ActionEvent e) {
                setButtonStatus(PROGRESS);
                for ( int i=0; i<numWorkerThreads; i++ ){
                    threads[i].interrupt();
                }
                for ( int i=0; i<numWorkerThreads; i++ ){
                    try {
                        threads[i].join();
                    } catch (InterruptedException ex) {  }
                }
                setStatusLabel("Threads cancelled");
                setButtonStatus(INIT);
            }
        });
        JPanel buttonPanel = new JPanel();
        buttonPanel.add(startButton);
        buttonPanel.add(pauseButton);
        buttonPanel.add(contButton);
        buttonPanel.add(stopButton);
        setButtonStatus(INIT);
        add(buttonPanel,BorderLayout.PAGE_START);
        txtA = new JTextArea(10,35);
        txtA.setEditable(false);
        JScrollPane scroll = new JScrollPane(txtA);
        add(scroll,BorderLayout.CENTER);
        JPanel statusPanel = new JPanel();
        statusLabel = new JLabel("Status: ");
        statusPanel.add(statusLabel);
        add(statusPanel,BorderLayout.PAGE_END);
    }
    public void outputToTextArea(final String str){
        SwingUtilities.invokeLater(new Runnable(){
            public void run() {
                txtA.append( str + "\n" );
            }
        });
    }
    public void setStatusLabel(final String str){
        SwingUtilities.invokeLater(new Runnable(){
            public void run() {
                statusLabel.setText("Status: " + str );
            }
        });
    }
    public static void main(String[] args) {
        SwingUtilities.invokeLater(new Runnable() {
            public void run() {
                 new TestThread(3);
            }
        });
    }

    class Worker implements Runnable {
        private boolean isPaused = false;
        public synchronized void pauseThread(){
            isPaused = true;
        }
        public synchronized void contThread(){
            isPaused = false;
            notify();
        }
        private synchronized void isPausedThread(){
            if ( isPaused ){
                pauseProgress.setPausedThreadStatus(Thread.currentThread().getName());
                try {
                    wait();
                } catch (InterruptedException ex) { }
            }
       }
       public void run() {
           while ( !Thread.currentThread().isInterrupted() ) {
              isPausedThread();
              outputToTextArea( Thread.currentThread().getName()+" "+new Date() );
              try {
                  Thread.sleep(5000);
              } catch (InterruptedException ex) {
                  Thread.currentThread().interrupt();
              }
              outputToTextArea( Thread.currentThread().getName()+" "+new Date() );
           }
       }
    }
    class PauseProgress implements Runnable{
        private ConcurrentHashMap<String,Boolean> threadPauseStatus;
        public void run() {
            threadPauseStatus = new ConcurrentHashMap<String,Boolean>(numWorkerThreads);
            for ( int i=0; i<numWorkerThreads; i++ ){
               if ( threads[i].isAlive() ) {
                  threadPauseStatus.put(threads[i].getName(), false);
                  workers[i].pauseThread();
               } else {
                  threadPauseStatus.put(threads[i].getName(), true);
               }
            }
            while ( !isAllThreadsPaused() ) {
                waiting();
            }
            setButtonStatus(PAUSED);
            setStatusLabel("Threads paused");
        }
        private synchronized void waiting(){
            try {
                wait();
            } catch (InterruptedException ex) {  }
         }
        private synchronized boolean isAllThreadsPaused(){
            boolean ret = true;
            for ( Map.Entry<String, Boolean> e: threadPauseStatus.entrySet() ){
                if ( !e.getValue() ){
                    ret = false;
                    break;
                }
            }
            return ret;
        }
        public synchronized void setPausedThreadStatus(String threadName) {
            threadPauseStatus.put(threadName, true);
            notify();
        }
    }
}

Mit freundlichen Grüßen
Attila Farkas

 

[ThreadProb]

Danke dir für das tolle Beispiel 🙂

Das pausieren Problem ist damit gelöst, nur beim stoppen friert die GUI nach wie vor ein (Nicht bei meinem mini-Beispiel, da funktioniert es) kann es Vllt. Daran liegen, dass pro Sekunde Ca. 500 mal das Update aufgerufen wird?

 

[Attila]

Das hängt davon ab, wie du das realisiert hast.