flink sink 只支持 bio 吗？

Question

sink的invoke方法好像没办法做异步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，但它似乎不被 sink 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);