Làm cách nào để hủy yêu cầu HTTP AsyncRestTemplate nếu chúng mất quá nhiều thời gian?

Question

Kể từ khi bắt đầu, tôi luôn luôn nhầm lẫn về cách đối phó với InterruptedException và cách hủy đúng yêu cầu http nếu họ đang dùng quá nhiều thời gian. Tôi có một thư viện mà tôi đã cung cấp hai phương pháp, đồng bộ hóa và không đồng bộ cho khách hàng của chúng tôi. Họ có thể gọi bất cứ phương pháp nào họ cảm thấy là đúng cho mục đích của họ.

• executeSync() - đợi cho đến khi tôi có kết quả, trả về kết quả.
• executeAsync() - trả về một tương lai ngay lập tức có thể được xử lý sau khi những việc khác được thực hiện, nếu cần.

Chúng sẽ vượt qua đối tượng DataKey có id người dùng và giá trị thời gian chờ trong đó. Chúng tôi sẽ tìm ra máy để gọi cơ sở trên id người dùng và sau đó tạo một URL với máy đó và chúng tôi sẽ thực hiện cuộc gọi http đến URL bằng cách sử dụng AsyncRestTemplate và sau đó gửi phản hồi về cơ sở cho dù thành công hay không.

Tôi đang sử dụng phương pháp exchangeAsyncRestTemplate mà trả lại một ListenableFuture và tôi muốn có kiến ​​trúc chặn phi async với các kết nối client NIO dựa có yêu cầu sử dụng không chặn IO vì vậy đó là lý do tại sao tôi đã đi với AsyncRestTemplate. Cách tiếp cận này có đúng với định nghĩa vấn đề của tôi không? Thư viện này sẽ được sử dụng trong sản xuất dưới tải nặng.

Dưới đây là giao diện của tôi:

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

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

Và dưới đây là thực hiện của tôi về giao diện:

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

Và khách hàng sẽ được gọi như thế này từ mã của họ -

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

Câu hỏi đặt ra là -

1. Chúng tôi có thể ngắt AsyncRestTemplate gọi nếu yêu cầu http mất quá nhiều thời gian không? Tôi thực sự đang gọi số cancel trên mã số future trong mã trên của mình theo phương thức executeSync nhưng tôi không chắc chắn làm cách nào để xác minh nó để đảm bảo rằng nó đang làm những gì cần? Tôi muốn tuyên truyền hủy bỏ trở lại tương lai ban đầu, để tôi có thể hủy bỏ yêu cầu http tương ứng (mà tôi có thể muốn làm để tiết kiệm tài nguyên) vì vậy đó là lý do tại sao tôi đã thêm một trình lắng nghe trong phương thức executeAsync của mình. Tôi tin rằng, chúng tôi không thể gián đoạn các cuộc gọi RestTemplate nhưng không chắc chắn về số AsyncRestTemplate cho dù chúng tôi có thể thực hiện điều đó hay không. Nếu chúng ta nói rằng chúng ta có thể làm gián đoạn các cuộc gọi AsyncRestTemplate, thì tôi có đang làm mọi thứ đúng để gián đoạn cuộc gọi http không? Hoặc là có cách nào tốt hơn/sạch hơn để làm điều này? Hoặc thậm chí tôi có cần phải lo lắng về việc hủy yêu cầu Http với AsyncRestTemplate với thiết kế hiện tại của tôi không?

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

   Với sự thiết lập hiện tại, tôi có thể nhìn thấy nó được ném CancellationException một số lần (không mọi) - Điều đó có nghĩa yêu cầu HTTP của tôi đã bị hủy bỏ sau đó?

2. Tôi cũng đang làm điều đúng trong khối catch của InterruptedException trong phương thức executeSync? Nếu không, thì cách thích hợp để giải quyết vấn đề đó là gì. Và tôi có cần xử lý InterruptedException trong trường hợp của tôi không?
3. Có đúng là theo mặc định AsyncRestTamplete sử dụng các cuộc gọi chặn và yêu cầu cho mỗi chuỗi? Nếu có, thì có cách nào để có các kết nối máy khách dựa trên NIO trong thiết lập hiện tại của tôi không?

Mọi đề xuất giải thích/mã sẽ được trợ giúp rất nhiều.

Answer 1

Trước hết, Tại sao bạn sử dụng SettableFuture? Tại sao không thể trả lại ListenableFuture được trả về bởi AsyncRestTemplate?

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

Tất nhiên rồi! Bạn chỉ cần gọi phương thức Future.cancel. Phương thức này sẽ làm gián đoạn việc thực hiện RestTemplate nội bộ mà AsyncRestTemplate thực sự đang sử dụng.

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

Như Phil và Danilo đã nói, bạn không cần phải làm gián đoạn luồng hiện tại trong khối catch InterruptedException. Chỉ cần làm bất cứ điều gì bạn cần làm khi thực hiện yêu cầu phải được hủy bỏ.

Thực tế, tôi khuyên bạn nên tạo phương thức xử lý hành vi này, chẳng hạn như handleInterruption và sử dụng phương thức này cho cả hai TimeoutException và InterruptedException.

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

Có. Hàm tạo mặc định là AsyncRestTamplete được sử dụng nội bộ SimpleClientHttpRequestFactory và SimpleAsyncTaskExecutor.

TaskExecutor này luôn bắt đầu một mối đe dọa cho mọi công việc, và không bao giờ tái sử dụng Chủ đề, vì vậy nó rất hiệu quả:

* 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. 
* 

tôi khuyên bạn nên sử dụng một cấu hình của AsyncRestTemplate.

Bạn nên sử dụng các nhà xây dựng của AsyncRestTemplate sử dụng khác TaskExecutor:

public AsyncRestTemplate(AsyncListenableTaskExecutor taskExecutor) 

Ví dụ:

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

ExecutorService này (Executors.newCachedThreadPool()) tạo chủ đề mới khi cần thiết, nhưng sẽ sử dụng lại các chuỗi được tạo trước đó khi chúng có sẵn.

Hoặc thậm chí tốt hơn, bạn có thể sử dụng một RequestFactory khác. Ví dụ, bạn có thể sử dụng HttpComponentsAsyncClientHttpRequestFactory, mà trong nội bộ sử dụng nio, chỉ cần gọi các nhà xây dựng đúng AsyncRestTemplate:

new AsyncRestTemplate(new HttpComponentsAsyncClientHttpRequestFactory()) 

Đừng quên các hành vi nội bộ của AsyncRestTemplate sẽ phụ thuộc vào cách bạn tạo đối tượng.