Java：使用异步编程优化应用程序

Question

我必须修改 dropwizard 应用程序以提高其运行时间。基本上，该应用程序每天接收大约 300 万个 URL，并下载并解析它们以检测恶意内容。问题是应用程序只能处理 100 万个 URL。当我查看应用程序时，我发现它正在进行大量的顺序调用。我想要一些关于如何通过异步或其他技术改进应用程序的建议。

所需代码如下：-

/* Scheduler */
private long triggerDetection(String startDate, String endDate) {
for (UrlRequest request : urlRequests) {
                if (!validateRequests.isWhitelisted(request)) {
                    ContentDetectionClient.detectContent(request);
                }
            }
}

/* Client */
public void detectContent(UrlRequest urlRequest){
        Client client = new Client();
        URI uri = buildUrl(); /* It returns the URL of this dropwizard application's resource method provided below */

        ClientResponse response = client.resource(uri)
                .type(MediaType.APPLICATION_JSON_TYPE)
                .post(ClientResponse.class, urlRequest);

        Integer status = response.getStatus();
        if (status >= 200 && status < 300) {
            log.info("Completed request for url: {}", urlRequest.getUrl());

        }else{
            log.error("request failed for url: {}", urlRequest.getUrl());
        }
    }

    private URI buildUrl() {
        return UriBuilder
                .fromPath(uriConfiguration.getUrl())
                .build();
    }

/* Resource Method */
 @POST
    @Path("/pageDetection")
    @Consumes(MediaType.APPLICATION_JSON)
    @Produces(MediaType.APPLICATION_JSON)
    /**
     * Receives the url of the publisher, crawls the content of that url, applies a detector to check if the content is malicious.
     * @returns returns the probability of the page being malicious
     * @throws throws exception if the crawl call failed
     **/
    public DetectionScore detectContent(UrlRequest urlRequest) throws Exception {

        return contentAnalysisOrchestrator.detectContentPage(urlRequest);
    }

/* Orchestrator */
public DetectionScore detectContentPage(UrlRequest urlRequest) {
        try {

            Pair<Integer, HtmlPage> response =  crawler.rawLoad(urlRequest.getUrl());
            String content =   response.getValue().text();

            DetectionScore detectionScore = detector.getProbability(urlRequest.getUrl(), content);
            contentDetectionResultDao.insert(urlRequest.getAffiliateId(), urlRequest.getUrl(),detectionScore.getProbability()*1000,
                    detectionScore.getRecommendation(), urlRequest.getRequestsPerUrl(), -1, urlRequest.getCreatedAt() );

            return detectionScore;

        } catch (IOException e) {
            log.info("Error while analyzing the url : {}", e);
            throw new WebApplicationException(e, Response.Status.INTERNAL_SERVER_ERROR);
        }
    }

我正在考虑以下方法：-

• 我没有通过 POST 调用 dropwizard 资源方法，而是直接从调度程序调用 orchestrator.detectContent(urlRequest)。

• 编排器可以返回 detectionScore，我会将所有的 detectScore 存储在一个映射/表中并执行批量数据库插入，而不是像当前代码中那样单独插入。

我想要一些关于上述方法的 cmets 以及其他可以改进运行时间的技术。另外，我刚刚阅读了有关 Java 异步编程的内容，但似乎无法理解如何在上面的代码中使用它，所以也希望得到一些帮助。

谢谢。

编辑： 我能想到两个瓶颈：

• 网页下载
• 将结果插入数据库（数据库位于另一个系统中）
• 似乎一次处理 1 个 URL

系统有 8 GB 内存，其中 4 GB 似乎是空闲的

$ free -m
             total       used       free     shared    buffers     cached
Mem:          7843       4496       3346          0        193       2339
-/+ buffers/cache:       1964       5879 
Swap:         1952        489       1463 

CPU 使用率也很低：

top - 13:31:19 up 19 days, 15:39,  3 users,  load average: 0.00, 0.00, 0.00
Tasks: 215 total,   1 running, 214 sleeping,   0 stopped,   0 zombie
Cpu(s):  0.5%us,  0.0%sy,  0.0%ni, 99.4%id,  0.1%wa,  0.0%hi,  0.0%si,  0.0%st
Mem:   8031412k total,  4605196k used,  3426216k free,   198040k buffers
Swap:  1999868k total,   501020k used,  1498848k free,  2395344k cached

Answer 1

首先检查您大部分时间松动的地方。

我想大部分时间都丢失了下载网址。

如果下载 url 花费超过 90% 的时间，您可能无法改进您的应用程序，因为瓶颈不是 java，而是您的网络。

---

仅当下载时间在网络能力范围内时才考虑以下事项

如果下载时间不是那么长，您可能可以尝试提高您的性能。一种标准方法是使用生产者消费者链。详情请见here。

基本上你可以按如下方式拆分工作：

Downloading --> Parsing --> Saving 

下载是生产者，解析是下载过程的消费者，生产者保存过程，保存是消费者。

每个步骤可以由不同数量的线程执行。比如你可以有3个下载线程、5个解析线程和1个保存线程。

---

在 cmets 之后编辑

假设瓶颈不是cpu时间，所以对java代码进行干预并不重要。

如果您知道每天下载多少千兆字节，就可以查看它们是否接近您网络的最大带宽。

如果发生这种情况，有不同的可能性：

• 使用Content-Encoding: gzip 请求压缩内容（从而减少使用的带宽）
• 在工作于不同网络的不同节点之间拆分您的应用程序（因此在不同网络之间拆分带宽）
• 更新您的带宽（因此为您的网络增加带宽）
• 确保只下载请求的内容（如果没有请求，则不要下载 javascript、图片、css 等）（以尽量减少带宽的使用）
• 先前解决方案的组合

Answer 2

受 Davide 的（伟大的）回答启发，这里有一个示例，使用 simple-react （我写的库）将其并行化的简单方法。请注意，它略有不同，使用客户端驱动服务器上的并发。

示例

LazyReact streamBuilder = new LazyReact(15,15);

streamBuilder.fromIterable(urlRequests)
      .filter(urlReq->!validateRequests.isWhitelisted(urlReq))
      .forEach(request -> {
           ContentDetectionClient.detectContent(request);
       });

说明

看起来您可以从客户端驱动并发。这意味着您可以在服务器端跨线程分配工作，而无需额外工作。在此示例中，我们发出 15 个并发请求，但您可以将其设置为接近服务器可以处理的最大值。您的应用程序是 IO Bound，因此您可以使用大量线程来提高性能。

simple-react 用作​​期货流。所以这里我们为每次调用 ContentDetection 客户端创建一个异步任务。我们有 15 个线程可用，因此可以一次对服务器进行 15 个调用。

Java 7

有一个用于 Java 7 的 JDK 8 功能的反向移植，称为 StreamSupport，您还可以通过 RetroLambda 反向移植 Lambda 表达式。

要使用 CompletableFutures 实现相同的解决方案，我们可以为每个符合条件的 URL 创建一个未来任务。 UPDATE 我认为我们不需要对它们进行批处理，我们可以使用 Executor 来限制活动期货的数量。我们只需要在最后加入他们。

   Executor exec = Executors.newFixedThreadPool(maxActive);//15 threads
   List<CompletableFuture<Void>> futures= new ArrayList<>();

   for (UrlRequest request : urlRequests) {
            if (!validateRequests.isWhitelisted(request)) {
                futures.add(CompletableFuture.runAsync(()->ContentDetectionClient.detectContent(request), exec));
            }
        }
 CompletableFuture.allOf(futures.toArray())
                      .join();