フリンクシンクはバイオのみをサポートしますか？

Question

シンクのinvokeメソッドは、非同期ioを作る方法のようですか？例えば返信Future？例えば、Redisのコネクターが同期Redisのコマンドを実行するためにjedis libにを使用しています

：

https://github.com/apache/bahir-flink/blob/master/flink-connector-redis/src/main/java/org/apache/flink/streaming/connectors/redis/RedisSink.java

そして、それは、コマンドごとにRedisのサーバからネットワークの応答を待っているFLINKのタスクスレッドをブロックします！シンクを持つ同じスレッドで実行されている他の演算子は可能ですか？もしそうなら、それもそれをブロックするだろうか？

私はflinkがasyncio APIを持っていることを知っていますが、シンクimplで使われていないようですね？ @Dexterが述べたように

https://ci.apache.org/projects/flink/flink-docs-release-1.3/dev/stream/asyncio.html

Answer 1

、あなたはRichAsyncFunctionを使用することができます。ここではサンプルコードは（それを動作させるために更なるアップデートが必要な場合があります。）である

AsyncDataStream.orderedWait(ds, new RichAsyncFunction<Tuple2<String,MyEvent>, String>() { 
     transient private RedisClient client; 
     transient private RedisAsyncCommands<String, String> commands; 
     transient private ExecutorService executor; 

     @Override 
     public void open(Configuration parameters) throws Exception { 
      super.open(parameters); 

      client = RedisClient.create("redis://localhost"); 
      commands = client.connect().async(); 
      executor = Executors.newFixedThreadPool(10); 
     } 

     @Override 
     public void close() throws Exception { 
      // shut down the connection and thread pool. 
      client.shutdown(); 
      executor.shutdown(); 

      super.close(); 
     } 

     public void asyncInvoke(Tuple2<String, MyEvent> input, final AsyncCollector<String> collector) throws Exception { 
      // eg.g get something from redis in async 
      final RedisFuture<String> future = commands.get("key"); 
      future.thenAccept(new Consumer<String>() { 
       @Override 
       public void accept(String value) { 
        collector.collect(Collections.singletonList(future.get())); 
       } 
      }); 
     } 
    }, 1000, TimeUnit.MILLISECONDS);