Как отменить HTTP-запрос AsyncRestTemplate, если они занимают слишком много времени?

Question

С самого начала я всегда путался с тем, как бороться с InterruptedException и как правильно отменить HTTP-запрос, если они занимают слишком много времени. У меня есть библиотека, в которой я предоставил два метода, синхронизацию и асинхронность для нашего клиента. Они могут вызывать тот метод, который, по их мнению, подходит для их целей.

• executeSync() - ждет, пока у меня есть результат, возвращает результат.
• executeAsync() - немедленно возвращает Будущее, которое может быть обработано после выполнения других действий, если это необходимо.

Они передадут объект DataKey, который имеет идентификатор пользователя и значение таймаута в нем. Мы выясним, какую машину вызывать на основе идентификатора пользователя, а затем создать URL-адрес с этой машиной, и мы сделаем http-обращение к URL-адресу, используя AsyncRestTemplate, а затем отправьте ответ на них по вопросу о том, успешна она или нет.

Я использую exchange метод AsyncRestTemplate который возвращает обратно ListenableFuture, и я хотел бы иметь асинхронную ноны архитектуры блокировки с клиентскими подключениями на основе НИО так, что запрос использует не блокирует IO так, поэтому я пошел с AsyncRestTemplate. Подходит ли этот подход к определению моей проблемы? Эта библиотека будет использоваться в производстве при очень большой нагрузке.

Ниже мой интерфейс:

public interface Client { 
    // for synchronous 
    public DataResponse executeSync(DataKey key); 

    // for asynchronous 
    public ListenableFuture<DataResponse> executeAsync(DataKey key); 
} 

И ниже моя реализация интерфейса:

public class DataClient implements Client { 

    // using spring 4 AsyncRestTemplate 
    private final AsyncRestTemplate restTemplate = new AsyncRestTemplate(); 

    // for synchronous 
    @Override 
    public DataResponse executeSync(DataKey keys) { 
     Future<DataResponse> responseFuture = executeAsync(keys); 
     DataResponse response = null; 

     try { 
      response = responseFuture.get(keys.getTimeout(), TimeUnit.MILLISECONDS); 
     } catch (InterruptedException ex) { 
      // do we need to catch InterruptedException here and interrupt the thread? 
      Thread.currentThread().interrupt(); 
      // also do I need throw this RuntimeException at all? 
      throw new RuntimeException("Interrupted", ex); 
     } catch (TimeoutException ex) { 
      DataLogging.logEvents(ex, DataErrorEnum.CLIENT_TIMEOUT, keys); 
      response = new DataResponse(null, DataErrorEnum.CLIENT_TIMEOUT, DataStatusEnum.ERROR); 
      responseFuture.cancel(true); // terminating the tasks that got timed out so that they don't take up the resources? 
     } catch (Exception ex) { 
      DataLogging.logEvents(ex, DataErrorEnum.ERROR_CLIENT, keys); 
      response = new DataResponse(null, DataErrorEnum.ERROR_CLIENT, DataStatusEnum.ERROR); 
     } 

     return response; 
    } 

    // for asynchronous  
    @Override 
    public ListenableFuture<DataResponse> executeAsync(final DataKey keys) { 

     final SettableFuture<DataResponse> responseFuture = SettableFuture.create(); 
     final org.springframework.util.concurrent.ListenableFuture orig = 
      restTemplate.exchange(createURL(keys), HttpMethod.GET, keys.getEntity(), String.class); 

     orig.addCallback(
       new ListenableFutureCallback<ResponseEntity<String>>() { 
        @Override 
        public void onSuccess(ResponseEntity<String> result) { 
         responseFuture.set(new DataResponse(result.getBody(), DataErrorEnum.OK, 
           DataStatusEnum.SUCCESS)); 
        } 

        @Override 
        public void onFailure(Throwable ex) { 
         DataLogging.logErrors(ex, DataErrorEnum.ERROR_SERVER, keys); 
         responseFuture.set(new DataResponse(null, DataErrorEnum.ERROR_SERVER, 
           DataStatusEnum.ERROR)); 
        } 
       }); 

     // propagate cancellation back to the original request 
     responseFuture.addListener(new Runnable() { 
      @Override public void run() { 
      if (responseFuture.isCancelled()) { 
       orig.cancel(false); // I am keeping this false for now 
      } 
      } 
     }, MoreExecutors.directExecutor()); 
     return responseFuture; 
    } 
} 

И клиент будет вызывающему как это от своего кода -

// if they are calling executeSync() method 
DataResponse response = DataClientFactory.getInstance().executeSync(dataKey); 

// and if they want to call executeAsync() method 
Future<DataResponse> response = DataClientFactory.getInstance().executeAsync(dataKey); 

Теперь задан вопрос -

1. Можем ли мы прервать вызов AsyncRestTemplate, если HTTP-запрос занимает слишком много времени? Я на самом деле звоню cancel по моему future в моем приведенном выше коде в методе executeSync, но я не уверен, как его проверить, чтобы убедиться, что он делает то, что должен? Я хочу отменить отмену в исходное будущее, чтобы отменить соответствующий HTTP-запрос (который я, вероятно, хочу сделать для сохранения ресурсов), поэтому я добавил слушателя в свой метод executeAsync. Я считаю, мы не можем прерывать звонки RestTemplate, но не уверены в том, можем ли мы это сделать или нет. Если предположим, что мы можем прервать AsyncRestTemplate звонки, то я делаю все правильно, чтобы прервать http-звонки? Или есть лучший/более чистый способ сделать это? Или мне даже нужно беспокоиться об отмене запроса Http с помощью AsyncRestTemplate с моим текущим дизайном?

   // propagate cancellation back to the original request 
       responseFuture.addListener(new Runnable() { 
        @Override public void run() { 
        if (responseFuture.isCancelled()) { 
         orig.cancel(false); // I am keeping this false for now 
        } 
        } 
       }, MoreExecutors.directExecutor()); 
   

   С текущей настройки, я могу видеть, что это бросает CancellationException часть времени (не каждый раз) - Означает ли это, что мой запрос HTTP был отменен тогда?

2. Также я делаю правильную вещь в блоке catch InterruptedException в executeSync методе? Если нет, то каков правильный способ справиться с этим. И мне нужно иметь дело с InterruptedException вообще в моем случае?
3. Верно ли, что по умолчанию AsyncRestTamplete использует блокирующие вызовы и запросы на поток? Если да, то есть ли способ подключения клиентов на основе NIO в моей текущей настройке?

Любые объяснения/предложения по коду будут очень полезны.

Answer 1

Прежде всего, почему вы используете SettableFuture? Почему нельзя просто вернуть ListenableFuture, возвращенный AsyncRestTemplate?

1. Can we interrupt AsyncRestTemplate call if http request is taking too long? 

Конечно, вы делаете! Вам нужно только позвонить по телефону Future.cancel. Этот метод прервет выполнение внутреннего RestTemplate, который фактически использует AsyncRestTemplate.

2. Also am I doing the right thing in catch block of InterruptedException in executeSync method? 

Как сказал Фил и Данило, вам не нужно прерывать текущий поток в блоке catch InterruptedException. Просто делайте то, что вам нужно делать, когда выполнение запроса должно быть отменено.

Фактически, я рекомендую создать метод, который обрабатывает это поведение, например handleInterruption, и использовать этот метод как для TimeoutException, так и для InterruptedException.

3. Is it true that by default AsyncRestTamplete uses blocking calls and request per thread? 

Да. Конструктор по умолчанию AsyncRestTamplete внутренне использует SimpleClientHttpRequestFactory и SimpleAsyncTaskExecutor.

Это TaskExecutor всегда начинается угрозой для каждой задачи, и никогда не используйте повторно тему, так что это очень неэффективно:

* TaskExecutor implementation that fires up a new Thread for each task, 
* executing it asynchronously. 
* 
* Supports limiting concurrent threads through the "concurrencyLimit" 
* bean property. By default, the number of concurrent threads is unlimited. 
* 
* NOTE: This implementation does not reuse threads! Consider a 
* thread-pooling TaskExecutor implementation instead, in particular for 
* executing a large number of short-lived tasks. 
* 

Я рекомендую вам использовать другую конфигурацию AsyncRestTemplate.

Вы должны использовать конструктор AsyncRestTemplate, который использует другой TaskExecutor:

public AsyncRestTemplate(AsyncListenableTaskExecutor taskExecutor) 

Например:

AsyncRestTemplate template = new AsyncRestTemplate(new ConcurrentTaskExecutor(Executors.newCachedThreadPool())); 

Это ExecutorService (Executors.newCachedThreadPool()) создает новые темы, по мере необходимости, но будут повторно использовать ранее построенные потоки, когда они будут доступны.

Или даже лучше, вы можете использовать другой RequestFactory. Например, вы можете использовать HttpComponentsAsyncClientHttpRequestFactory, что внутренне использует NIO, просто звоню правильный конструктор AsyncRestTemplate:

new AsyncRestTemplate(new HttpComponentsAsyncClientHttpRequestFactory()) 

Не забывайте внутреннее поведение AsyncRestTemplate будет зависеть от того, как вы создаете объект.