Выполнение после ожидания событий нескольких потоков при сохранении запущенных процессов

Question

В основном я запускаю несколько потоков, которые снова запускают длительный процесс, который следует контролировать. В примере «длинные» запущенные процессы - это простые вызовы ping. IpListener несет ответственность за запуск событий, когда процесс завершается или завершается с ошибкой.

Теперь идет интересная часть: каждый раз, когда ip находится на выходе консоли, запускается метод onIpFound. То, что я хочу сделать, - подождать, пока все потоки не найдут IP-адрес в первый раз, а затем продолжите выполнение основного метода. Однако запущенные процессы должны продолжать выполняться и в конечном итоге должны вызывать вышеупомянутые события onProcessEnd и onProcessFailure. Есть ли у вас какие-либо намеки для реализации этого поведения (звуковым образом)?

Вот часть небольшого примера приложения, чтобы дать вам лучшее понимание:

ReaderTest.java

package com.test; 
public class ReaderTest { 
    public static void main(String[] args) { 
     IpReader ipReader = new IpReader("stackoverflow.com"); 
     ipReader.setListener(new SimpleIpListener()); 
     new Thread(ipReader).start(); 

     IpReader ipReader2 = new IpReader("stackexchange.com"); 
     ipReader2.setListener(new SimpleIpListener()); 
     new Thread(ipReader2).start(); 

     // TODO: start next step as soon as both ips are known 
    } 
} 

IpReader.java

package com.test; 

import java.io.BufferedReader; 
import java.io.IOException; 
import java.io.InputStreamReader; 
import java.util.regex.Matcher; 
import java.util.regex.Pattern; 

public class IpReader implements Runnable { 

    private IpListener ipListener = null; 
    private String hostName; 

    public IpReader(String hostName) { 
     this.hostName = hostName; 
    } 

    public interface IpListener { 
     public void onIpFound(String ip); 

     public void onProcessEnd(String string); 

     public void onProcessFailure(String string); 
    } 

    public void setListener(IpListener ipListener) { 
     this.ipListener = ipListener; 
    } 

    @Override 
    public void run() { 
     // TODO Auto-generated method stub 
     ProcessBuilder pb = new ProcessBuilder().command("cmd", "/c", "ping", "-n", "10", hostName); 
     pb.redirectErrorStream(true); 

     Process process = null; 
     try { 
      try { 
       process = pb.start(); 

       String line = null; 
       // print stream 
       BufferedReader br = new BufferedReader(new InputStreamReader(process.getInputStream())); 
       while ((line = br.readLine()) != null) { 

        Pattern p = Pattern.compile("(?:[0-9]{1,3}\\.){3}[0-9]{1,3}"); 
        Matcher m = p.matcher(line); 

        if (m.find()) { 
         // IPv4 found 
         if (ipListener != null) { 
          ipListener.onIpFound(m.group(0)); 
         } 
        } 
       } 

       process.waitFor(); 
       if (ipListener != null) 
        ipListener.onProcessEnd("Process END"); 
      } finally { 
       if (process != null) 
        process.destroy(); 
      } 

     } catch (InterruptedException | IOException e) { 
      if (ipListener != null) 
       ipListener.onProcessFailure("Process failure"); 
      // TODO Auto-generated catch block 
      e.printStackTrace(); 
     } 

    } 

} 

SimpleIpListener.java

package com.test; 
import com.test.IpReader.IpListener; 

public class SimpleIpListener implements IpListener { 

    private boolean ipFound = false; 

    @Override 
    public void onIpFound(String ip) { 
     if (!ipFound) 
      System.out.println("IP " + ip + " found."); 
     ipFound = true; 
    } 

    @Override 
    public void onProcessEnd(String string) { 
     System.out.println("Process ended."); 
    } 

    @Override 
    public void onProcessFailure(String string) { 
     System.out.println("Process failure"); 
    } 

} 

Answer 1

Один из способов сделать это - ввести объект, совместно используемый потоками IpReader (или вашего слушателя), которые они могут обновить, когда они нашли ip.

Например ввести этот простой класс:

class WaitTask { 
    private int numberRead; 

    public int getNumberRead(){ 
     return numberRead; 
    } 

    public void increment(){ 
     numberRead++; 
    } 
} 

Теперь в главном методе:

public static void main(String[] args) { 
    // create an instance of that class, which keeps track of how many threads found an ip 
    final WaitTask waitTask = new WaitTask(); 
    // create an int to know how many we are expecting 
    final int totalThreadsRunning = 2; 

    IpReader ipReader = new IpReader("stackoverflow.com"); 
    ipReader.setListener(new SimpleIpListener()); 

    // pass our object to the thread 
    ipReader.setWaitTask(waitTask); 
    new Thread(ipReader).start(); 

    IpReader ipReader2 = new IpReader("stackexchange.com"); 
    ipReader2.setListener(new SimpleIpListener()); 

    // pass our object to the thread 
    ipReader2.setWaitTask(waitTask); 
    new Thread(ipReader2).start(); 

    // Now wait until the wait task object is safe to be accessed 

    synchronized(waitTask){ 

     // once we know it's safe let's check if we read as many responses as required 

     while(waitTask.getNumberRead() < totalThreadsRunning) 
     { 
      // instead of looping forever wait for a thread to notify that 
      // the wait task number read was changed 

      waitTask.wait(); 
      // the wait blocks execution and waits until waitTask was    
      // changed, after that the while loop condition gets evaluated 
      // once we read as many as required we exit the loop 
     } 
    } 

    // when we reach here all threads have finished so do anything you want 
} 

И в вашем IpThread добавить:

private WaitTask waitTask; 

public void setWaitTask(WaitTask task){ 
    this.waitTask = task; 
} 

[...] 

// IPv4 found 
    if (ipListener != null) { 
     ipListener.onIpFound(m.group(0)); 
     // wait again until the wait task is safe to be accesed 
     synchronized(waitTask){ 
      // increment it's value 
      waitTask.increment(); 
      // notify that we changed the value 
      waitTask.notify(); 
     } 

    } 
[...] 

Answer 2

Используйте "атомарные переменные" в хранить IP-адреса. Они могут быть установлены потоками пинга и проверены основным потоком. В вашей основной теме остановите «переменную условия». Когда пинговые потоки получат IP-адрес, они разбудят основной поток. https://docs.oracle.com/javase/7/docs/api/java/util/concurrent/locks/Condition.html https://docs.oracle.com/javase/8/docs/api/java/util/concurrent/atomic/AtomicReference.html

Что это будет делать это, вы создали переменную, которая пинг нить может размещать данные, а основной поток также может читать. Кроме того, основной поток останавливает переменную условия (и не потребляет процессорное время), но пробуждается при обнаружении нового ip. Затем он проверяет, заполнены ли обе ссылки ip.

Определить эти глобально:

AtomicReference ip1 = new AtomicReference(new String("")); 
AtomicReference ip2 = new AtomicReference(new String("")); 
final Lock lock = new ReentrantLock(); 
final Condition wakeup = lock.newCondition(); 

Затем в главном потоке

while(ip1.get().length == 0 && ip2.get().length == 0) 
{ 
wakeup.await(); 
} 

В определении IpReader:

private String hostName; 
private AtomicReference ipstring; 

public IpReader(String hostName,AtomicReference outputstring) { 
    this.hostName = hostName; 
    this.ipstring = outputstring; 
} 

Наконец

@Override 
public void onIpFound(String ip, AtomicReference ref) { 
    if (!ipFound) 
     System.out.println("IP " + ip + " found."); 
    ipFound = true; 
    ref.set(ip); 
    wakeup.signal(); 
}