Utilizzo di Spring Threading e TaskExecutor, come faccio a sapere quando un thread è finito?

Question

Va bene, è possibile una domanda ingenua qui. Ho un servizio che deve accedere a più dispositivi di rete, eseguire un comando su ciascuno e raccogliere i risultati. Per velocizzare, invece di raccogliere le informazioni su ciascun dispositivo in sequenza, ho bisogno di accedervi tutti contemporaneamente e consumare i risultati dopo che sono stati completati.

Utilizzando il framework Spring e Jsch sono abbastanza facilmente in grado di interrogare correttamente ciascun dispositivo. Dove mi sto imbattendo in una certa confusione è nel tentativo di ricablare i bean per utilizzare TaskExecutor per realizzare questo. Quello che non riesco a capire come fare è sapere quando il thread è finito.

Quello che ho finora è questo:

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"); 
     } 
    } 
} 

Nel mio test JUnit ho:

@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()); 
     //} 
    } 

Il mio file applicationContext.xml:

<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> 

ho andare il più lontano come prima println nel metodo run(). Quindi il test esce in modo pulito senza errori. Non arrivo mai alla seconda stampa in fondo a quella routine. Ho esaminato questa discussione here, che è stata molto utile, ma non implementata in modo specifico per Spring. Sono sicuro che mi manca qualcosa di semplice, o sono completamente scappato dai binari qui. Qualsiasi aiuto è apprezzato.

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

L'interfaccia TaskExecutor è un'interfaccia ignifuga, da utilizzare quando non ti interessa quando l'attività termina. È la più semplice astrazione asincrona che la primavera offre.

Esiste, tuttavia, un'interfaccia avanzata, AsyncTaskExecutor, che fornisce metodi aggiuntivi, inclusi i metodi submit() che restituiscono un Future, che consente di attendere il risultato.

Spring fornisce la classe ThreadPoolTaskExecutor, che attuano sia TaskExecutor e AsyncTaskExecutor.

Nel tuo caso specifico, sarebbe re-implementare la Runnable come Callable, e restituire il commandResults dal metodo Callable.call(). Il metodo getCommandResults può quindi essere reimplementata come:

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

Questo metodo presenterà il compito in modo asincrono, e quindi attendere che sia completato prima di restituire i risultati restituiti dal metodo Callable.call(). Questo ti consente anche di sbarazzarti del campo commandResults.