一旦线程中断，如何中断RestTemplate调用？

Question

我需要制作一个具有同步和异步功能的库。

• executeSynchronous() -等到得到结果，然后返回结果。
• executeAsynchronous() -立即返回Future，如果需要，可以在完成其他操作后进行处理。

我图书馆的核心逻辑

客户将使用我们的库，他们将通过传递DataKey构建器对象来调用它。然后，我们将使用该DataKey对象构造一个URL，并通过执行该对象来对该URL进行HTTP客户端调用，并在将响应作为JSON字符串返回给我们之后，通过创建DataResponse对象将该JSON字符串发送回给我们的客户。有些客户会打电话executeSynchronous()，有些可能会打电话，executeAsynchronous()所以这就是为什么我需要在库中分别提供两个方法。

接口：

public interface Client {

    // for synchronous
    public DataResponse executeSynchronous(DataKey key);

    // for asynchronous
    public Future<DataResponse> executeAsynchronous(DataKey key);
}

然后我有了DataClient实现以上Client接口的代码：

public class DataClient implements Client {

    private RestTemplate restTemplate = new RestTemplate();
    private ExecutorService executor = Executors.newFixedThreadPool(10);

    // for synchronous call
    @Override
    public DataResponse executeSynchronous(DataKey key) {
        DataResponse dataResponse = null;
        Future<DataResponse> future = null;

        try {
            future = executeAsynchronous(key);
            dataResponse = future.get(key.getTimeout(), TimeUnit.MILLISECONDS);
        } catch (TimeoutException ex) {
            PotoLogging.logErrors(ex, DataErrorEnum.TIMEOUT_ON_CLIENT, key);
            dataResponse = new DataResponse(null, DataErrorEnum.TIMEOUT_ON_CLIENT, DataStatusEnum.ERROR);
            // does this looks right?
            future.cancel(true); // terminating tasks that have timed out
        } catch (Exception ex) {
            PotoLogging.logErrors(ex, DataErrorEnum.CLIENT_ERROR, key);
            dataResponse = new DataResponse(null, DataErrorEnum.CLIENT_ERROR, DataStatusEnum.ERROR);
        }

        return dataResponse;
    }

    //for asynchronous call
    @Override
    public Future<DataResponse> executeAsynchronous(DataKey key) {
        Future<DataResponse> future = null;

        try {
            Task task = new Task(key, restTemplate);
            future = executor.submit(task); 
        } catch (Exception ex) {
            PotoLogging.logErrors(ex, DataErrorEnum.CLIENT_ERROR, key);
        }

        return future;
    }
}

简单类将执行实际任务：

public class Task implements Callable<DataResponse> {

    private DataKey key;
    private RestTemplate restTemplate;

    public Task(DataKey key, RestTemplate restTemplate) {
        this.key = key;
        this.restTemplate = restTemplate;
    }

    @Override
    public DataResponse call() {
        DataResponse dataResponse = null;
        String response = null;

        try {
            String url = createURL();
            response = restTemplate.getForObject(url, String.class);

            // it is a successful response
            dataResponse = new DataResponse(response, DataErrorEnum.NONE, DataStatusEnum.SUCCESS);
        } catch (RestClientException ex) {
            PotoLogging.logErrors(ex, DataErrorEnum.SERVER_DOWN, key);
            dataResponse = new DataResponse(null, DataErrorEnum.SERVER_DOWN, DataStatusEnum.ERROR);
        } catch (Exception ex) {
            PotoLogging.logErrors(ex, DataErrorEnum.CLIENT_ERROR, key);
            dataResponse = new DataResponse(null, DataErrorEnum.CLIENT_ERROR, DataStatusEnum.ERROR);
        }

        return dataResponse;
    }

    // create a URL by using key object
    private String createURL() {
        String url = somecode;
        return url;
    }
}

当我开始研究此解决方案时，我并没有终止超时的任务。我正在向客户端报告超时，但是任务继续在线程池中运行（很长时间可能占用了我有限的10个线程之一）。因此，我在网上进行了一些研究，发现可以通过在将来使用cancel来取消超时的任务，如下所示-

future.cancel(true);

但是，如果我按照上述解决方案所示进行操作，那么是否需要像RestTemplate线程中断一样立即关闭任何其他资源？如果是，那我该怎么办？另外，我们可以打扰RestTemplate电话吗？因为我试图在任务超时后立即调用cancel，但是我猜我的线程不会被中断。

我们是否应该总是终止已经超时的任务？如果我们不这样做，那我会有什么影响？会影响我的表现吗？

在我当前的设置中，有没有更好的解决方案来处理这种情况？

Answer 1

有时不可能中断线程，尤其是当线程在Socket上执行阻塞操作时。

因此，与其在超时时取消任务，不如在http连接上设置超时。

不幸的是，每个连接工厂和RestTemplate都设置了时间限制，因此每个请求必须使用它自己的RestTemplate。

您可以为每个任务创建新的RestTemplate，或者使用ThreadLocal或资源池重用以前创建的模板。

例如，使用Thread local的任务可能如下所示：

    public class Task implements Callable<DataResponse> {

    private DataKey key;

    private ThreadLocal<RestTemplate> restTemplateThreadLocal =
            ThreadLocal.withInitial(()->new RestTemplate(new SimpleClientHttpRequestFactory()));

    public Task(DataKey key) {
        this.key = key;
    }

    private SimpleClientHttpRequestFactory getConnectionFactory(){
        return (SimpleClientHttpRequestFactory)restTemplateThreadLocal.get().getRequestFactory();
    }

    @Override
    public DataResponse call() {
        DataResponse dataResponse = null;
        String response = null;

        try {
            String url = createURL();
            //it is up to you, how to set connection and read timeouts from provided key.getTimeout
            getConnectionFactory().setConnectTimeout(1000);
            getConnectionFactory().setReadTimeout(key.getTimeout());
            response = restTemplateThreadLocal.get().getForObject(url, String.class);

            // it is a successful response
            dataResponse = new DataResponse(response, DataErrorEnum.NONE, DataStatusEnum.SUCCESS);
        } catch (RestClientException ex) {
            PotoLogging.logErrors(ex, DataErrorEnum.SERVER_DOWN, key);
            dataResponse = new DataResponse(null, DataErrorEnum.SERVER_DOWN, DataStatusEnum.ERROR);
        } catch (Exception ex) {
            PotoLogging.logErrors(ex, DataErrorEnum.CLIENT_ERROR, key);
            dataResponse = new DataResponse(null, DataErrorEnum.CLIENT_ERROR, DataStatusEnum.ERROR);
        }

        return dataResponse;
    }

    // create a URL by using key object
    private String createURL() {
        String url = somecode;
        return url;
    }
   }

顺便说一句。Spring还提供了AsyncRestTemplate，可以使您的代码更简单。如果与Netty4ClientHttpRequestFactory一起使用，则可以获取基于NIO的客户端连接。在这种情况下，即使建立Http连接，您也应该能够中断您的任务。

以下是简短示例。它使用NIO，因此您不必担心超时后是否真的取消了请求。

        URI url = new URI("http://www.chicagotribune.com/news/ct-college-of-dupage-investigation-met-20150330-story.html");
        Netty4ClientHttpRequestFactory asyncRequestFactory = new Netty4ClientHttpRequestFactory();
        AsyncRestTemplate asyncRestTemplate = new AsyncRestTemplate(asyncRequestFactory);
        ListenableFuture<ResponseEntity<String>> entity = asyncRestTemplate.getForEntity(url, String.class);
        System.out.println("entity.get() = " + entity.get());
        asyncRequestFactory.destroy();