Java 使用 Spring 线程和 TaskExecutor，我如何知道线程何时完成？

Using Spring threading and TaskExecutor, how do I know when a thread is finished?

提问by Bill

Alright, possible a naive question here. I have a service that needs to log into multiple network devices, run a command on each and collect the results. For speed, rather than collect the information on each device in sequence, I need to access them all concurrently and consume the results after they are done.

好吧，这里可能是一个幼稚的问题。我有一个服务需要登录多个网络设备，在每个设备上运行一个命令并收集结果。为了速度，我不需要按顺序收集每个设备上的信息，而是需要同时访问它们并在完成后使用结果。

Using the Spring framework and Jsch I'm quite easily able to query each device correctly. Where I am running into some confusion is in trying to rewire the beans to use TaskExecutor to accomplish this. What I can't figure out how to do is how to know when the thread is finished.

使用 Spring 框架和 Jsch，我可以很容易地正确查询每个设备。我遇到一些困惑的地方是尝试重新连接 bean 以使用 TaskExecutor 来完成此操作。我不知道该怎么做是如何知道线程何时完成。

What I have so far is this:

到目前为止，我所拥有的是：

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

Within my junit test I have:

在我的 junit 测试中，我有：

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

My applicationContext.xml 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> 

I get as far as the first println in the run() method. Then the test exits cleanly with no errors. I never get to the second println at the bottom of that routine. I've looked at this thread here, which was very useful, but not implemented in a Spring specific fashion. I'm sure I'm missing something simple, or have completely run off the rails here. Any help is appreciated.

我得到了 run() 方法中的第一个 println 。然后测试干净地退出，没有错误。我永远不会到达该例程底部的第二个 println。我在这里查看了这个线程，它非常有用，但没有以特定于 Spring 的方式实现。我确定我错过了一些简单的东西，或者在这里完全脱离了轨道。任何帮助表示赞赏。

采纳答案by Igor Artamonov

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
}

回答by skaffman

The TaskExecutorinterface is a fire-and-forget interface, for use when you don't care when the task finishes. It's the simplest async abstraction that Spring offers.

该TaskExecutor界面是一个即发即弃的界面，供您在不关心任务何时完成时使用。这是 Spring 提供的最简单的异步抽象。

There is , however, an enhanced interface, AsyncTaskExecutor, which provides additional methods, including submit()methods that return a Future, which let you wait on the result.

但是，有一个增强的接口 ，AsyncTaskExecutor它提供了额外的方法，包括submit()返回 a 的方法，Future让您等待结果。

Spring provides the ThreadPoolTaskExecutorclass, which implement both TaskExecutorand AsyncTaskExecutor.

Spring 提供了ThreadPoolTaskExecutor实现TaskExecutor和的类AsyncTaskExecutor。

In your specific case, I would re-implement the Runnableas a Callable, and return the commandResultsfrom the Callable.call()method. The getCommandResultsmethod can then be reimplemented as:

在您的特定情况下，我会重新实现Runnableas a Callable，并commandResults从Callable.call()方法返回。getCommandResults然后可以重新实现该方法：

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

This method will submit the task asynchronously, and then wait for it to complete before returning the results returned from the Callable.call()method. This also lets you get rid of the commandResultsfield.

该方法会异步提交任务，然后等待任务完成，再返回Callable.call()方法返回的结果。这也可以让您摆脱该commandResults领域。