我正在使用将对象DataRequest作为输入参数并基于该对象的库,我将构造一个URL,然后使用apache http客户端调用我们的应用服务器,然后将响应返回给使用我们图书馆的顾客。一些客户将调用executeSync方法以获取相同的功能,而某些客户将调用我们的executeAsync方法以获取数据。


executeSync()-等待直到得到结果,然后返回结果。
executeAsync()-立即返回Future,如果需要,可以在完成其他操作之后进行处理。


下面是我的DataClient类,它具有以上两种方法:

public class DataClient implements Client {
  private final ForkJoinPool forkJoinPool = new ForkJoinPool(16);
  private CloseableHttpClient httpClientBuilder;

  // initializing httpclient only once
  public DataClient() {
    try {
      RequestConfig requestConfig =
          RequestConfig.custom().setConnectionRequestTimeout(500).setConnectTimeout(500)
              .setSocketTimeout(500).setStaleConnectionCheckEnabled(false).build();
      SocketConfig socketConfig =
          SocketConfig.custom().setSoKeepAlive(true).setTcpNoDelay(true).build();

      PoolingHttpClientConnectionManager poolingHttpClientConnectionManager =
          new PoolingHttpClientConnectionManager();
      poolingHttpClientConnectionManager.setMaxTotal(300);
      poolingHttpClientConnectionManager.setDefaultMaxPerRoute(200);

      httpClientBuilder =
          HttpClientBuilder.create().setConnectionManager(poolingHttpClientConnectionManager)
              .setDefaultRequestConfig(requestConfig).setDefaultSocketConfig(socketConfig).build();
    } catch (Exception ex) {
      // log error
    }
  }

  @Override
  public List<DataResponse> executeSync(DataRequest key) {
    List<DataResponse> responsList = null;
    Future<List<DataResponse>> responseFuture = null;

    try {
      responseFuture = executeAsync(key);
      responsList = responseFuture.get(key.getTimeout(), key.getTimeoutUnit());
    } catch (TimeoutException | ExecutionException | InterruptedException ex) {
      responsList =
          Collections.singletonList(new DataResponse(DataErrorEnum.CLIENT_TIMEOUT,
              DataStatusEnum.ERROR));
      responseFuture.cancel(true);
      // logging exception here
    }
    return responsList;
  }

  @Override
  public Future<List<DataResponse>> executeAsync(DataRequest key) {
    DataFetcherTask task = new DataFetcherTask(key, this.httpClientBuilder);
    return this.forkJoinPool.submit(task);
  }
}


下面是我的DataFetcherTask类,它也有一个静态类DataRequestTask,该类通过创建URL来调用我们的应用服务器:

public class DataFetcherTask extends RecursiveTask<List<DataResponse>> {
  private final DataRequest key;
  private final CloseableHttpClient httpClientBuilder;

  public DataFetcherTask(DataRequest key, CloseableHttpClient httpClientBuilder) {
    this.key = key;
    this.httpClientBuilder = httpClientBuilder;
  }

  @Override
  protected List<DataResponse> compute() {
    // Create subtasks for the key and invoke them
    List<DataRequestTask> requestTasks = requestTasks(generateKeys());
    invokeAll(requestTasks);

    // All tasks are finished if invokeAll() returns.
    List<DataResponse> responseList = new ArrayList<>(requestTasks.size());
    for (DataRequestTask task : requestTasks) {
      try {
        responseList.add(task.get());
      } catch (InterruptedException | ExecutionException e) {
        Thread.currentThread().interrupt();
        return Collections.emptyList();
      }
    }
    return responseList;
  }

  private List<DataRequestTask> requestTasks(List<DataRequest> keys) {
    List<DataRequestTask> tasks = new ArrayList<>(keys.size());
    for (DataRequest key : keys) {
      tasks.add(new DataRequestTask(key));
    }
    return tasks;
  }

  // In this method I am making a HTTP call to another service
  // and then I will make List<DataRequest> accordingly.
  private List<DataRequest> generateKeys() {
    List<DataRequest> keys = new ArrayList<>();
    // use key object which is passed in contructor to make HTTP call to another service
    // and then make List of DataRequest object and return keys.
    return keys;
  }

  /** Inner class for the subtasks. */
  private static class DataRequestTask extends RecursiveTask<DataResponse> {
    private final DataRequest request;

    public DataRequestTask(DataRequest request) {
      this.request = request;
    }

    @Override
    protected DataResponse compute() {
      return performDataRequest(this.request);
    }

    private DataResponse performDataRequest(DataRequest key) {
      MappingHolder mappings = DataMapping.getMappings(key.getType());
      List<String> hostnames = mappings.getAllHostnames(key);

      for (String hostname : hostnames) {
        String url = generateUrl(hostname);
        HttpGet httpGet = new HttpGet(url);
        httpGet.setConfig(generateRequestConfig());
        httpGet.addHeader(key.getHeader());

        try (CloseableHttpResponse response = httpClientBuilder.execute(httpGet)) {
          HttpEntity entity = response.getEntity();
          String responseBody =
              TestUtils.isEmpty(entity) ? null : IOUtils.toString(entity.getContent(),
                  StandardCharsets.UTF_8);

          return new DataResponse(responseBody, DataErrorEnum.OK, DataStatusEnum.OK);
        } catch (IOException ex) {
          // log error
        }
      }
      return new DataResponse(DataErrorEnum.SERVERS_DOWN, DataStatusEnum.ERROR);
    }
  }
}


对于每个DataRequest对象,都有一个DataResponse对象。现在,一旦有人通过传递DataRequest对象调用我们的库,我们内部将创建List<DataRequest>对象,然后我们并行调用每个DataRequest对象,然后将List<DataResponse>返回到列表中每个DataResponse对象将响应的位置。相应的DataRequest对象。

以下是流程:


客户将通过传递DataClient对象来调用DataRequest类。他们可以根据自己的要求调用executeSync()executeAsync()方法。
现在在DataFetcherTask类(这是RecursiveTask子类型之一的ForkJoinTask's)中,给定key对象是单个DataRequest,我将生成List<DataRequest>,然后并行地为每个子任务调用列表中的每个DataRequest对象。这些子任务在与父任务相同的ForkJoinPool中执行。
现在在DataRequestTask类中,我通过创建一个URL执行每个DataRequest对象,并返回其DataResponse对象。




问题陈述:

由于在非常高吞吐量的环境中调用此库,因此它必须非常快。对于同步调用,可以在单独的线程中执行吗?在这种情况下,这将导致线程的额外成本和资源以及线程的上下文切换成本,因此我有点困惑。另外我在这里使用ForkJoinPool可以节省额外的线程池,但这是正确的选择吗?

有没有更好,更有效的方法来完成同样可以提高性能的事情?我正在使用Java 7,并且还可以访问Guava库,因此,如果它可以简化任何事情,那么我也愿意这样做。

看起来当它在非常重的负载下运行时,我们看到了一些争论。在非常重的负载下运行时,此代码有什么办法可以进入线程争用状态?

最佳答案

我认为在您的情况下,最好使用异步http调用,请参见链接:HttpAsyncClient。而且您不需要使用线程池。

在executeAsync方法中,创建一个空的CompletableFuture ()并将其传递给客户端调用,在回调调用中,通过对其调用complete(或在异常引发时,则调用completeException)来设置completableFuture的结果。
ExecuteSync方法的实现看起来不错。

编辑:

对于Java 7,只需替换completableFuture即可保证在番石榴中实现,如ListenableFuture或任何类似的东西

07-25 23:21