如何在线程中断后立即中断RestTemplate调用



我需要制作一个库,在其中我将具有同步和异步功能。

  • 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个线程中的一个)。所以我在网上做了一些研究,我发现我可以通过在未来使用取消来取消那些超时的任务,如下所示-

future.cancel(true);

但是,如果我像上面的解决方案中所示的那样这样做,那么一旦线程中断,我是否需要关闭任何其他资源(如RestTemplate)?如果是,我该怎么做?另外,我们可以中断RestTemplate调用吗?因为我试着在任务超时后立即调用cancel,但我想我的线程没有被中断。

我们应该总是终止超时的任务吗?如果我们不这样做,我会有什么影响?这会影响我的表现吗?

用我目前的设置,有没有更好的解决方案来处理这种情况?

RestTemplate的调用似乎无法中断或取消。即使使用了使用回调的"kludge",RestTemplate也可能在内部锁定资源,等待响应后再调用回调。

当底层套接字可访问时,可以通过从另一个线程关闭套接字来中止网络I/O。例如,可以启动计时器以在超时后关闭套接字。或者,如果您想要一个对中断敏感的无限期超时(例如,由于用户按下"取消"按钮),您可以提交一个无限期等待但通过关闭套接字来响应中断的任务。

不幸的是,RestTemplate的作者似乎并没有提供这种能力。

是的,您应该清理由于任务取消或过期而不再需要的资源。是的,它会影响性能。如果线程池中的线程数量有限,那么最终所有线程都将被困在不起作用的任务中。如果它有无限数量的线程,那么内存最终会耗尽。

有时无法中断线程,尤其是当线程对Socket执行阻塞操作时。

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

不幸的是,超时是根据Connection Factory和RestTemplate设置的,因此每个请求都必须使用自己的RestTemplate。

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

例如,使用线程本地的任务可能如下所示:

    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,因此您不必关心请求是否真的在Timeout之后被取消。

        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();

相关内容

  • 没有找到相关文章

最新更新