根据需要调用多少页来并行化do while循环？

Question

我必须通过传递header和body来发出HTTP POST请求。在正文中，我需要在发布数据之前提供pageNumber，所以我最初以“1”开头。之后，我将发布数据，并返回JSON响应，如下所示。

{
    "response": {
        "pageNumber": 1,
        "entries": 200,
        "numberOfPages": 3
    },
    "list": [
        {
            // some stuff here
        }
    ],
    "total": 1000
}

现在，根据pageNumber1的响应，我将决定还需要进行多少次调用。现在在上面的响应中，numberofpages是3，所以我需要对同一个URL总共进行三次调用。由于我们已经进行了1个调用，我将再进行2个调用，在主体中使用pageNumber“2”和“3”。

下面是我的工作代码。我只需要通过更改正文调用相同的URL直到numberofpages次。对于每个呼叫，都应该使用相应的PageNumber进行，因此如果NumberOfPages是3，那么我将总共进行3个呼叫。从每页收集数据后，我正在填充两张地图。

public class AppParser {
  private static final ObjectMapper objectMapper = new ObjectMapper();
  private static final String lastParentIdJsonPath = "......";    
  private final Map<String, String> processToTaskIdHolder = new HashMap<>();
  private final Multimap<String, Category> itemsByCategory = LinkedListMultimap.create();
  private final int entries;
  private final String siteId;

  public AppParser(int entries, String id) {
    this.entries = entries;
    this.id = id;
    collect();
  }

  // this is only called from above constructor
  private void collect() {
    String endpoint = "url_endpoint";
    int number = 1;
    int expectedNumber;
    do {
      HttpEntity<String> requestEntity = new HttpEntity<String>(getBody(number), getHeader());
      ResponseEntity<String> responseEntity =
          HttpClient.getInstance().getClient()
              .exchange(URI.create(endpoint), HttpMethod.POST, requestEntity, String.class);
      String jsonInput = responseEntity.getBody();
      Stuff response = objectMapper.readValue(jsonInput, Stuff.class);
      expectedNumber = (int) response.getPaginationResponse().getNumberOfPages();
      if (expectedNumber <= 0) {
        break;
      }
      List<Postings> postings = response.getPostings();
      for (Postings posting : postings) {
        if (posting.getClientIds().isEmpty()) {
          continue;
        }
        List<String> lastParent = JsonPath.read(jsonInput, lastParentIdJsonPath);
        String clientId = posting.getClientIds().get(0).getId();
        Category category = getCategory(posting);
        // populate two maps now
        itemsByCategory.put(clientId, category);
        processToTaskIdHolder.put(clientId, lastParent.get(0));
      }
      number++;
    } while (number <= expectedNumber);
  }

  private String getBody(final int number) {
    Input input = new Input(entries, number, 0);
    Body body = new Body("Stuff", input);
    return gson.toJson(body);
  }

  // getters for those two above maps
}

现在我的上面的代码是按顺序为每个页面一个一个地收集数据，所以如果我有很高的numberofpages，那么收集所有这些页码的所有数据将需要一些时间。假设numberofpages是500，那么我的代码将为每个PageNumber逐个依次运行。有没有什么方法可以并行化我上面的代码，这样我们就可以同时收集数据，比如说5页？这有可能做到吗？我想我需要确保我的代码是线程安全的。

注意:HttpClient是线程安全的单例类。

Answer 1

我试图用多线程修改您的代码，但这并不容易，因为您没有提供包含所有导入的完整类源代码。此外，您的代码也不像可能的那样干净。您的任务是异步请求的常见情况。我将您的收集代码包装到java.util.concurrent.callable中。它通过ExecutorService为我提供异步使用任务，并在需要时将结果作为parseresult对象获取。在下面的代码中，我提出了一个请求来填充expectedNumber变量，循环中应用程序创建任务并将它们提交到executorservice中，并使用它们运行的专用线程池。代码：

private static final ObjectMapper objectMapper = new ObjectMapper();
private static final String URL_ENDPOINT = "url_endpoint";
private final Map<String, String> processToTaskIdHolder = new HashMap<>();
private final Multimap<String, Category> itemsByCategory = LinkedListMultimap.create();
private static final String lastParentIdJsonPath = "......";

class ParseResult {
    private String clientId;
    private Category category;
    private String lastParent;
    private int expectedNumber;
}

class ParseTask implements Callable<ParseResult> {

    private int pageNumber;

    public ParseTask(int pageNumber) {
        this.pageNumber = pageNumber;
    }

    @Override
    public ParseResult call() throws Exception {
        HttpEntity<String> requestEntity = new HttpEntity<String>(getBody(pageNumber), getHeader());
        ResponseEntity<String> responseEntity =
                HttpClient.getInstance().getClient()
                        .exchange(URI.create(URL_ENDPOINT), HttpMethod.POST, requestEntity, String.class);
        String jsonInput = responseEntity.getBody();
        Stuff response = objectMapper.readValue(jsonInput, Stuff.class);
        int expectedNumber = (int) response.getPaginationResponse().getNumberOfPages();
        if (expectedNumber <= 0) {
            return null; // or throw exception
        }
        List<Postings> postings = response.getPostings();
        for (Postings posting : postings) {
            if (posting.getClientIds().isEmpty()) {
                continue;
            }
            List<String> lastParent = JsonPath.read(jsonInput, lastParentIdJsonPath);
            String clientId = posting.getClientIds().get(0).getId();
            Category category = getCategory(posting);

            //collecting the result
            ParseResult parseResult = new ParseResult();
            parseResult.clientId = clientId;
            parseResult.category = category;
            parseResult.expectedNumber = expectedNumber;
            parseResult.lastParent = lastParent.get(0);
            writeResult(parseResult); // writing the result
            return parseResult;
        }
    }
}

public AppParser(int entries, String id) {
    // .....
    collect();
}

// this is only called from above constructor
private void collect() {
    int number = 1;
    int expectedNumber = 0;
    ParseTask parseTask = new ParseTask(number);
    try {
        ParseResult firstResult = parseTask.call();
        expectedNumber = firstResult.expectedNumber; // fill the pages amount
    } catch (Exception e) {
        e.printStackTrace();
    }

    ExecutorService executorService = Executors.newCachedThreadPool();
    while (number <= expectedNumber) {
        executorService.submit(new ParseTask(number));
    }
}

private String getBody(final int number) {
    Input input = new Input(entries, number, 0);
    Body body = new Body("Stuff", input);
    return gson.toJson(body);
}

private void writeResult(ParseResult result) {
    // populate two maps now
    itemsByCategory.put(result.clientId, result.category);
    processToTaskIdHolder.put(result.clientId, result.lastParent);
}

我们可以花很多时间来升级您的代码，但这是一个多线程的原始版本。我不确定它是否有效，因为正如我之前所说，你没有提供完整的版本。也许它需要一些语法修复。