Блог

Когда мне нужно работать с вводом-выводом (запрос к БД, вызов третьего API, …), я могу использовать RichAsyncFunction . Но мне нужно взаимодействовать с Google Sheet через API GG Sheet: https://developers.google.com/sheets/api/quickstart/java . Этот API является sync. Я написал ниже фрагмент кода:

 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 очень медленно, кажется, оно отправляется синхронно.

Большая часть кода, выполняемого пользователями, 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