我可以在RichAsyncFunction中编写同步代码吗

Question

当我需要使用I/O(查询数据库，调用第三个API，…(时，我可以使用RichAsyncFunction。但我需要通过GG Sheet API与Google Sheet互动：https://developers.google.com/sheets/api/quickstart/java.此API是同步的。我写了以下代码片段：

public class SendGGSheetFunction extends RichAsyncFunction<Obj, String> {
@Override
public void asyncInvoke(Obj message, final ResultFuture<String> resultFuture) {
CompletableFuture.supplyAsync(() -> {
syncSendToGGSheet(message);
return "";
}).thenAccept((String result) -> {
resultFuture.complete(Collections.singleton(result));
});
}
}

但我发现消息发送到GGSheet的速度很慢，它似乎是通过同步发送的。

Answer 1

用户在AsyncIO中执行的大多数代码最初都是同步的。您只需要确保它实际上是在一个单独的线程中执行的。最常见的是使用(静态共享的(ExecutorService。

private class SendGGSheetFunction extends RichAsyncFunction<Obj, String> {
private transient ExecutorService executorService;
@Override
public void open(Configuration parameters) throws Exception {
super.open(parameters);
executorService = Executors.newFixedThreadPool(30);
}
@Override
public void close() throws Exception {
super.close();
executorService.shutdownNow();
}
@Override
public void asyncInvoke(final Obj message, final ResultFuture<String> resultFuture) {
executorService.submit(() -> {
try {
resultFuture.complete(syncSendToGGSheet(message));
} catch (SQLException e) {
resultFuture.completeExceptionally(e);
}
});
}
}

以下是关于如何调整AsyncIO以提高吞吐量的一些注意事项：http://apache-flink-user-mailing-list-archive.2336050.n4.nabble.com/Flink-Async-IO-operator-tuning-micro-benchmarks-td35858.html