java - 从调用者类停止异步 Spring 方法

Question

我有一个调用 Rest 网络服务以从服务器接收文件的类。在传输字节时，我创建了一个异步任务，它检查与服务器的连接是否正常，以便在出现错误时允许停止连接。这个异步任务有一个我必须停止的循环:

@Component
public class ConnectionTest {

    @Async
    //Check connection with the server, if for three attemp it failes, throw exception
    public void checkServerConnection(String serverIp) throws Exception{
        int count=0;
        for(;;Thread.sleep(7000)){
            try{
                System.out.println("TEST");
                URL url = new URL(serverIp);
            HttpURLConnection con = (HttpURLConnection) url
                    .openConnection();
            con.connect();
            if (con.getResponseCode() == 200){
                System.out.println("Connection established!!");
            }
                if (count>0) count=0;
            }catch(Exception e){
                count++;
                if (count==3)   
                    throw new Exception("Connection error");
            }
        }
    }
}

但是我怎样才能从调用者那里停止这个方法呢？

@Autowired 
    private ConnectionTest connectionTest;

    @Override
    public Response getFile(String username, String password, String serverIp, String toStorePath, String filePath){


        ResponseEntity<byte[]> responseEntity = null;
        try{
            //it is used to check if connection of the client with the server goes down
            connectionTest.checkServerConnection();
            RestClient restClient = new RestClient(username, password);     

            //          SimpleClientHttpRequestFactory requestFactory = new SimpleClientHttpRequestFactory();
            //          requestFactory.setBufferRequestBody(false);     
            //          restClient.setRequestFactory(requestFactory);
            //          RestTemplate restClient = new RestTemplate();
            responseEntity  = restClient.getForEntity(serverIp + "client/file/?filePath={filePath}", byte[].class, filePath);   

            //TODO kill async task and return false

更新:正如@Thomas 建议我在 ConnectionTest 中使用了一个 boolean 变量，我用 while (!stop) 更改了循环> 在网络服务调用之后，我设置了 ConnectionTest.setStop(true)。注意在循环之前设置 stop=false(而不是作为实例字段)，否则只有第一个请求具有此值并进入 while。

更新 2这是我最后的代码，它似乎可以工作，也许我应该用 wait-notify 更改 while 循环:

public Response getFile(String username, String password, String serverIp, String toStorePath, String filePath){
        try{
            //it is used to check if connection of the client with the server goes down
            Future<Boolean> isConnect = connectionTest.checkServerConnection(serverIp);
            Future<ResponseEntity<byte[]>> downloadResult = downloadAsync.makeRequest(username, password, serverIp, filePath);

            while(!isConnect.isDone() && !downloadResult.isDone()){
            }
            if (isConnect.isDone()){
                downloadResult.cancel(true);
                return new Response(false, false, "Error with server connection!", null);
            }else{
                connectionTest.setStop(true);
                ResponseEntity<byte[]> responseEntity = downloadResult.get();

                if (MediaType.TEXT_PLAIN.toString().equals(responseEntity.getHeaders().getContentType().toString())){
                    ErrorResponse errorResponse= ErrorResponseBuilder.buildErrorResponse(new FileException("Error with file transfert!"));
                    return new Response(false, false, new String(Base64.decodeBase64(responseEntity.getBody()),Charset.forName("UTF-8")), errorResponse);
                }else{
                    Path p = Paths.get(filePath);
                    String fileName = p.getFileName().toString();
                    FileOutputStream fos = new FileOutputStream(toStorePath+"\\"+ fileName);
                    fos.write(responseEntity.getBody());
                    fos.close();
                    return new Response(true, true, "Your file has been downloaded!", null);
                }
            }
        }catch(Exception e){
            ErrorResponse errorResponse= ErrorResponseBuilder.buildErrorResponse(e);
            return new Response(false, false, "Error on the client side!" , errorResponse);
        }
    }

异步连接检查:

@Component
public class ConnectionTest {

    private boolean stop;

    @Async
    //Check connection with the server, if for three attemp it failes, throw exception
    /**
     * 
     * @param serverIp
     * @throws IOException
     */
    public Future<Boolean> checkServerConnection(String serverIp)  throws IOException {
        int count=0;
        stop = false;
        while (!stop){
            try{
                Thread.sleep(7000);
                System.out.println("TEST");
                //java.net.InetAddress.getByName(SERVER_ADDRESSS);
                URL url = new URL(serverIp);
                HttpURLConnection con = (HttpURLConnection) url
                        .openConnection();
                con.connect();
                if (count>0) count=0;
            }catch(Exception e){
                count++;
                System.out.println(count);
                if (count==3)   
                    return new AsyncResult<Boolean>(stop);
            }
        }
        return new AsyncResult<Boolean>(stop);
    }

    /**
     * @return the stop
     */
    public boolean isStop() {
        return stop;
    }

    /**
     * @param stop the stop to set
     */
    public void setStop(boolean stop) {
        this.stop = stop;
    }
}

异步下载:

@Component
public class DownloadAsync {

    @Async
    public Future<ResponseEntity<byte[]>> makeRequest(String username, String password, String serverIp, String filePath){
        RestClient restClient = new RestClient(username, password);
        ResponseEntity<byte[]> response= restClient.getForEntity(serverIp + "client/file/?filePath={filePath}", byte[].class, filePath);    
        return new AsyncResult<ResponseEntity<byte[]>>(response);
    }
}

Answer 1

当您处理 @Async 方法时，一个好的做法是从它返回一个 Future 对象，因为您需要客户端和任务代码之间的连接点。

让我们让您的任务方法返回一个Future:

public Future<Integer> checkServerConnection(String serverIp) {
    // other code here
    return new AsyncResult<>(count);
}

您需要添加几个导入:

import java.util.concurrent.Future;
import org.springframework.scheduling.annotation.AsyncResult;

最后，在客户端代码中获取Future:

Future<Integer> checkTask = connectionTest.checkServerConnection();

现在，您可以使用 checkTask 做一些有用的事情。例如:

// Check if the task was completed including by an exception being thrown.
checkTask.isDone();

// Get the task result.
Integer count = checkTask.get(); // Note: this is a blocking method.

// If the task was finished by throwing an exception,
// get() method will also throw an exception.
// You can get the cause exception like this:
if (checkTask.isDone()) {
    try {
        checkTask.get();
    } catch(Exception e) {
        Exception cause = e.getCause(); // this will be your new Exception("Connection error")
    }
}

// Not recommended, but you can also cancel the task:
checkTask.cancel(mayInterruptIfRunning);