flink水槽只支持bio吗？

Question

接收器的调用方法似乎没有办法使异步io？例如返回Future？

例如，redis连接器使用jedis lib同步执行redis命令：

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

然后，它将阻止flink的任务线程等待redis服务器对每个命令的网络响应？！其他操作符是否可以与sink在同一线程中运行？如果是这样，那么它也会阻止他们吗？

我知道flink有asyncio api，但它似乎不是为接收器impl使用的？

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

Answer 1

正如@Dexter所提到的，您可以使用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);