Flink DataStream Async I/O(异步IO)

╰半夏微凉° 2021-12-15 05:21 290阅读 0赞

## Flink DataStream Async I/O(异步IO) ##

当我们在Operator算子中，需要与外部系统交互时（例如：查询数据库），如果我们使用同步的方式，那么容易造成该Operator计算延迟，吞吐量低。所以Flink 提供了`Async I/O`机制，可以通过异步操作，处理查询数据库等类似耗时的操作。

与数据库的异步交互意味着单个并行函数实例可以同时处理许多请求并同时接收响应。这样，等待时间可以覆盖发送其他请求和接收响应。至少，等待时间是在多个请求上摊销的。这导致大多数情况下流量吞吐量更高。如下图：

\[外链图片转存失败(img-b53beq63-1562223811981)(https://raw.githubusercontent.com/zhang3550545/image\_center/master/image-2019/flink-anyscio1.png)\]

#### 一、Async IO 示例 ####

public class AsyncIOFunctionTest { 
        public static void main(String[] args) throws Exception { 
    
            StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
            env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime);
            env.setParallelism(1);
    
            Properties p = new Properties();
            p.setProperty("bootstrap.servers", "localhost:9092");
    
            DataStreamSource<String> ds = env.addSource(new FlinkKafkaConsumer010<String>("order", new SimpleStringSchema(), p));
            ds.print();
    
            SingleOutputStreamOperator<Order> order = ds
                    .map(new MapFunction<String, Order>() { 
                        @Override
                        public Order map(String value) throws Exception { 
                            return new Gson().fromJson(value, Order.class);
                        }
                    })
                    .assignTimestampsAndWatermarks(new AscendingTimestampExtractor<Order>() { 
                        @Override
                        public long extractAscendingTimestamp(Order element) { 
                            try { 
                                return element.getOrderTime();
                            } catch (Exception e) { 
                                e.printStackTrace();
                            }
                            return 0;
                        }
                    })
                    .keyBy(new KeySelector<Order, String>() { 
                        @Override
                        public String getKey(Order value) throws Exception { 
                            return value.getUserId();
                        }
                    })
                    .window(TumblingEventTimeWindows.of(Time.minutes(10)))
                    .maxBy("orderTime");
    
            SingleOutputStreamOperator<Tuple7<String, String, Integer, String, String, String, Long>> operator = AsyncDataStream
                    .unorderedWait(order, new RichAsyncFunction<Order, Tuple7<String, String, Integer, String, String, String, Long>>() { 
    
                        private Connection connection;
    
                        @Override
                        public void open(Configuration parameters) throws Exception { 
                            super.open(parameters);
                            Class.forName("com.mysql.jdbc.Driver");
                            connection = DriverManager.getConnection("url", "user", "pwd");
                            connection.setAutoCommit(false);
                        }
    
                        @Override
                        public void asyncInvoke(Order input, ResultFuture<Tuple7<String, String, Integer, String, String, String, Long>> resultFuture) throws Exception { 
                            List<Tuple7<String, String, Integer, String, String, String, Long>> list = new ArrayList<>();
                            // 在 asyncInvoke 方法中异步查询数据库
                            String userId = input.getUserId();
                            Statement statement = connection.createStatement();
                            ResultSet resultSet = statement.executeQuery("select name,age,sex from user where userid=" + userId);
                            if (resultSet != null && resultSet.next()) { 
                                String name = resultSet.getString("name");
                                int age = resultSet.getInt("age");
                                String sex = resultSet.getString("sex");
                                Tuple7<String, String, Integer, String, String, String, Long> res = Tuple7.of(userId, name, age, sex, input.getOrderId(), input.getPrice(), input.getOrderTime());
                                list.add(res);
                            }
    
                            // 将数据搜集
                            resultFuture.complete(list);
                        }
    
                        @Override
                        public void close() throws Exception { 
                            super.close();
                            if (connection != null) { 
                                connection.close();
                            }
                        }
                    }, 5000, TimeUnit.MILLISECONDS,100);
    
            operator.print();
    
    
            env.execute("AsyncIOFunctionTest");
        }
    }

从上面示例中可看到，我们在`open()`中创建连接对象，在`close()`方法中关闭连接，在`RichAsyncFunction的asyncInvoke()`方法中，直接查询数据库操作，并将数据返回出去。

以上就算是完成了Async IO的操作。

**Async IO**处理的结果根据顺序可分为2种，一种是\*\*`unorderedWait()`**，另一种是**`orderedWait()`\*\*。

*  `unorderedWait()`：异步请求完成后立即发出结果记录。在异步I / O运算符之后，流中记录的顺序与以前不同。当使用*处理时间*作为基本时间特性时，此模式具有最低延迟和最低开销。
 *  `orderedWait()`：在这种情况下，保留流顺序。结果记录的发出顺序与触发异步请求的顺序相同（运算符输入记录的顺序）。为此，运算符缓冲结果记录，直到其所有先前记录被发出（或超时）。这通常会在检查点中引入一些额外的延迟和一些开销，因为与无序模式相比，记录或结果在检查点状态下保持更长的时间。

2个方法的参数都是一样的。参数有：

*  DataStream ds：需要通过异步处理的流对象
 *  AsyncFunction async：实际处理的异步方法，写异步操作逻辑
 *  long timeout：超时时间（如果超时，则中断异步操作，避免过多的操作连接占用资源，尤其是针对orderedWait）
 *  TimeUnit timeUnit：超时时间单位
 *  int capacity：一次触发异步请求的最大数量，默认100。