Wie erkenne ich mit Spring threading und TaskExecutor, wann ein Thread fertig ist?

Question

Okay, möglich eine naive Frage hier. Ich habe einen Dienst, der sich bei mehreren Netzwerkgeräten anmelden, einen Befehl ausführen und die Ergebnisse sammeln muss. Aus Gründen der Geschwindigkeit muss ich die Informationen nicht nacheinander auf jedem Gerät sammeln, sondern muss gleichzeitig auf alle zugreifen und die Ergebnisse konsumieren, wenn sie fertig sind.

Mit dem Spring-Framework und Jsch Ich bin ziemlich leicht in der Lage, jedes Gerät korrekt abzufragen. Wenn ich in Verwirrung gerät, versuche ich, die Beans neu zu verdrahten, um den TaskExecutor zu verwenden. Was ich nicht herausfinden kann, ist wie man weiß, wann der Thread fertig ist.

Was ich habe, so weit ist dies:

public class RemoteCommand { 

    private String user; 
    private String host; 
    private String password; 
    private String command; 
    private List<String> commandResults; 
    private TaskExecutor taskExecutor; 

    public RemoteCommand(String user, String host, String password, TaskExecutor taskExecutor) { 

     setUser(user); 
     setHost(host); 
     setPassword(password); 
     setTaskExecutor(taskExecutor); 
    } 

    /** 
    * @param user the user to set 
    */ 
    public void setUser(String user) { 
     this.user = user; 
    } 

    /** 
    * @return the user 
    */ 
    public String getUser() { 
     return user; 
    } 

    /** 
    * @param host the host to set 
    */ 
    public void setHost(String host) { 
     this.host = host; 
    } 

    /** 
    * @return the host 
    */ 
    public String getHost() { 
     return host; 
    } 

    /** 
    * @param password the password to set 
    */ 
    public void setPassword(String password) { 
     this.password = password; 
    } 

    /** 
    * @return the password 
    */ 
    public String getPassword() { 
     return password; 
    } 

    /** 
    * @param command the command to set 
    */ 
    private void setCommand(String command) { 
     this.command = command; 
    } 

    /** 
    * @return the command 
    */ 
    private String getCommand() { 
     return command; 
    } 

    /** 
    * @param commandResults the commandResults to set 
    */ 
    private void setCommandResults(List<String> commandResults) { 
     this.commandResults = commandResults; 
    } 

    /** 
    * @return the commandResults 
    */ 
    public List<String> getCommandResults(String command) { 
     taskExecutor.execute(new CommandTask(command)); 

     return commandResults; 
    } 

    /** 
    * @param taskExecutor the taskExecutor to set 
    */ 
    public void setTaskExecutor(TaskExecutor taskExecutor) { 
     this.taskExecutor = taskExecutor; 
    } 

    /** 
    * @return the taskExecutor 
    */ 
    public TaskExecutor getTaskExecutor() { 
     return taskExecutor; 
    } 

    private class CommandTask implements Runnable { 

     public CommandTask(String command) { 
      setCommand(command); 
      System.out.println("test: " + getCommand()); 
     } 

     /** 
     * 
     * @param command 
     */ 
     public void run() { 

      List<String> results = new LinkedList<String>(); 
      String command = getCommand(); 

      try { 
       System.out.println("running"); 
       JSch jsch = new JSch(); 

       String user = getUser(); 
       String host = getHost(); 

       java.util.Properties config = new java.util.Properties(); 
       config.put("StrictHostKeyChecking", "no"); 

       host = host.substring(host.indexOf('@') + 1); 
       Session session = jsch.getSession(user, host, 22); 

       session.setPassword(getPassword()); 
       session.setConfig(config); 
       session.connect(); 

       Channel channel = session.openChannel("exec"); 
       ((ChannelExec) channel).setCommand(command); 

       channel.setInputStream(null); 

       ((ChannelExec) channel).setErrStream(System.err); 

       InputStream in = channel.getInputStream(); 

       channel.connect(); 
       byte[] tmp = new byte[1024]; 
       while (true) { 
        while (in.available() > 0) { 
         int i = in.read(tmp, 0, 1024); 
         if (i < 0) 
          break; 
         results.add(new String(tmp, 0, i)); 
         System.out.print(new String(tmp, 0, i)); 
        } 
        if (channel.isClosed()) { 
         //System.out.println("exit-status: " 
         //  + channel.getExitStatus()); 
         break; 
        } 
        try { 
         Thread.sleep(1000); 
        } catch (Exception ee) { 
         ee.printStackTrace(); 
        } 
       } 
       channel.disconnect(); 
       session.disconnect(); 
      } catch (Exception e) { 
       System.out.println(e); 
      } 
      setCommandResults(results); 
      System.out.println("finished running"); 
     } 
    } 
} 

In meinem junit Test habe ich:

@Test 
    public void testRemoteExecution() { 

     remoteCommand = (RemoteCommand) applicationContext.getBean("remoteCommand"); 
     remoteCommand.getCommandResults("scripts/something.pl xxx.xxx.xxx.xxx"); 

      //List<String> results = remoteCommand.getCommandResults("scripts/something.pl xxx.xxx.xxx.xxx"); 
     //for (String line : results) { 
     // System.out.println(line.trim()); 
     //} 
    } 

Meine applicationContext.xml Datei:

<bean id="taskExecutor" class="org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor"> 
     <property name="corePoolSize" value="5" /> 
     <property name="maxPoolSize" value="10" /> 
     <property name="queueCapacity" value="25" /> 
    </bean>   

<!-- ******************** --> 
<!--  Utilities  --> 
<!-- ******************** --> 

    <bean name="remoteCommand" class="com.xxx.ncc.sonet.utilities.RemoteCommand" scope="prototype"> 
     <description>Remote Command</description> 
     <constructor-arg><value>${remote.user}</value></constructor-arg> 
     <constructor-arg><value>${remote.host}</value></constructor-arg> 
     <constructor-arg><value>${remote.password}</value></constructor-arg> 
     <constructor-arg ref="taskExecutor" /> 
    </bean> 

ich so weit als erste println in der run() -Methode. Dann endet der Test sauber ohne Fehler. Ich komme nie zum zweiten Druck am Ende dieser Routine. Ich habe mir diesen Thread here angesehen, der sehr nützlich war, aber nicht in einer Spring-spezifischen Weise implementiert wurde. Ich bin mir sicher, dass ich etwas Einfaches verpasse, oder hier komplett von den Schienen gerannt bin. Jede Hilfe wird geschätzt.

Answer 1

public List<String> getCommandResults(String command) { 
    FutureTask task = new FutureTask(new CommandTask(command)) 
    taskExecutor.execute(task); 

    return task.get(); //or task.get(); return commandResults; - but it not a good practice 
} 

Answer 2

Die Schnittstelle TaskExecutor ist eine Fire-and-Forget-Schnittstelle, die verwendet werden kann, wenn Sie nach Abschluss der Task nicht interessiert sind. Es ist die einfachste asynchrone Abstraktion, die Spring bietet.

Es gibt jedoch eine erweiterte Schnittstelle, AsyncTaskExecutor, die zusätzliche Methoden bietet, einschließlich submit() Methoden, die eine Future zurückgeben, mit denen Sie auf das Ergebnis warten können.

Feder bietet die ThreadPoolTaskExecutor Klasse, die beide TaskExecutor und AsyncTaskExecutor implementieren.

In Ihrem speziellen Fall würde ich die Runnable als Callable, und gibt die commandResults vom Callable.call() Verfahren neu implementieren. Die getCommandResults Verfahren können dann neu implementiert werden, wie:

public List<String> getCommandResults(String command) { 
    Future<List<String>> futureResults = taskExecutor.submit(new CommandTask(command)); 
    return futureResults.get(); 
} 

Diese Methode asynchron die Aufgabe einreichen wird, und dann warten, bis es abgeschlossen wird, bevor die Ergebnisse zurückgegeben von der Callable.call() Methode zurück. Dadurch können Sie auch das Feld commandResults loswerden.