apache-flink - Flink 1.6 Async IO - 如何在丰富流时增加吞吐量，使用 REST 服务调用？

Question

我目前使用的是 Flink 1.6 版，并且面临 AsyncIO 的问题，其中性能达不到我的预期。

我确信我在执行过程中做错了什么，所以任何建议/建议都将不胜感激。

问题概要 - 我正在使用一个 id 流。对于每个 id，我需要调用一个 REST 服务。我已经实现了一个 RichAsyncFunction，它执行异步 REST 调用。

下面是相关的代码方法和asyncInvoke方法

// these are initialized in the open method
ExecutorService executorService = 
ExecutorService.newFixedThreadPool(n);
CloseableHttpAsyncClient client = ...
Gson gson = ...

public void asyncInvoke(String key, final ResultFuture<Item> resultFuture) throws Exception {

    executorService.submit(new Runnable() {

        client.execute(new HttpGet(new URI("http://myservice/" + key)), new FutureCallback<HttpResponse>() {
             @Override
                public void completed(final HttpResponse response) {
                    System.out.println("completed successfully");
                    Item item = gson.fromJson(EntityUtils.toString(response.getEntity), Item.class);
                    resultFuture.complete(Collections.singleton(item));
                }
        });
    });

}

通过上述实现，我尝试过：-

• 增加浓缩操作的并行度
• 增加执行器服务中的线程数
• 使用 apache http 异步客户端，我尝试调整连接管理器设置 - setDefaultMaxPerRoute 和 setMaxTotal。

我始终获得大约 100 个请求/秒的吞吐量。该服务每秒能够处理超过 5k。我做错了什么，我该如何改进？

标签： apache-flink