目錄

擴展閱讀? 異步IO

介紹

異步IO操作的需求

使用Aysnc I/O的前提條件

Async I/O API

案例演示

擴展閱讀 原理深入

AsyncDataStream

消息的順序性

---

擴展閱讀? 異步IO

介紹

異步IO操作的需求

Apache Flink 1.12 Documentation: Asynchronous I/O for External Data Access

Async I/O 是阿里巴巴貢獻給社區的一個呼聲非常高的特性，于1.2版本引入。主要目的是為了解決與外部系統交互時網絡延遲成為了系統瓶頸的問題。

流計算系統中經常需要與外部系統進行交互，我們通常的做法如向數據庫發送用戶a的查詢請求，然后等待結果返回，在這之前，我們的程序無法發送用戶b的查詢請求。這是一種同步訪問方式，如下圖所示

1. 左圖所示：通常實現方式是向數據庫發送用戶a的查詢請求（例如在MapFunction中），然后等待結果返回，在這之前，我們無法發送用戶b的查詢請求，這是一種同步訪問的模式，圖中棕色的長條標識等待時間，可以發現網絡等待時間極大的阻礙了吞吐和延遲
2. 右圖所示：為了解決同步訪問的問題，異步模式可以并發的處理多個請求和回復，可以連續的向數據庫發送用戶a、b、c、d等的請求，與此同時，哪個請求的回復先返回了就處理哪個回復，從而連續的請求之間不需要阻塞等待，這也正是Async I/O的實現原理。

使用Aysnc I/O的前提條件

1. 數據庫(或key/value存儲系統)提供支持異步請求的client。（如java的vertx）
2. 沒有異步請求客戶端的話也可以將同步客戶端丟到線程池中執行作為異步客戶端

Async I/O API

Async I/O API允許用戶在數據流中使用異步客戶端訪問外部存儲，該API處理與數據流的集成，以及消息順序性（Order），事件時間（EventTime），一致性（容錯）等臟活累活，用戶只專注于業務

如果目標數據庫中有異步客戶端，則三步即可實現異步流式轉換操作（針對該數據庫的異步）：

• 實現用來分發請求的AsyncFunction，用來向數據庫發送異步請求并設置回調
• 獲取操作結果的callback，并將它提交給ResultFuture
• 將異步I/O操作應用于DataStream

案例演示

兩種方式實現Flink異步IO查詢Mysql_優優我心的博客-CSDN博客

需求：

使用異步IO實現從MySQL中讀取數據

數據準備：

DROP TABLE IF EXISTS `t_category`;CREATE TABLE `t_category` (`id` int(11) NOT NULL,`name` varchar(255) DEFAULT NULL,PRIMARY KEY (`id`)) ENGINE=InnoDB DEFAULT CHARSET=utf8;-- ------------------------------ Records of t_category-- ----------------------------INSERT INTO `t_category` VALUES ('1', '手機');INSERT INTO `t_category` VALUES ('2', '電腦');INSERT INTO `t_category` VALUES ('3', '服裝');INSERT INTO `t_category` VALUES ('4', '化妝品');INSERT INTO `t_category` VALUES ('5', '食品');

代碼演示-異步IO讀取MySQL

package cn.i.extend;import io.vertx.core.AsyncResult;
import io.vertx.core.Handler;
import io.vertx.core.Vertx;
import io.vertx.core.VertxOptions;
import io.vertx.core.json.JsonObject;
import io.vertx.ext.jdbc.JDBCClient;
import io.vertx.ext.sql.SQLClient;
import io.vertx.ext.sql.SQLConnection;
import lombok.AllArgsConstructor;
import lombok.Data;
import lombok.NoArgsConstructor;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.AsyncDataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.async.ResultFuture;
import org.apache.flink.streaming.api.functions.async.RichAsyncFunction;
import org.apache.flink.streaming.api.functions.source.RichSourceFunction;import java.sql.*;
import java.util.Collections;
import java.util.List;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;/*** 使用異步io的先決條件* 1.數據庫(或key/value存儲)提供支持異步請求的client。* 2.沒有異步請求客戶端的話也可以將同步客戶端丟到線程池中執行作為異步客戶端。*/
public class ASyncIODemo {public static void main(String[] args) throws Exception {//1.envStreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();//2.Source//DataStreamSource[1,2,3,4,5]DataStreamSource<CategoryInfo> categoryDS = env.addSource(new RichSourceFunction<CategoryInfo>() {private Boolean flag = true;@Overridepublic void run(SourceContext<CategoryInfo> ctx) throws Exception {Integer[] ids = {1, 2, 3, 4, 5};for (Integer id : ids) {ctx.collect(new CategoryInfo(id, null));}}@Overridepublic void cancel() {this.flag = false;}});//3.Transformation//方式一：Java-vertx中提供的異步client實現異步IO//unorderedWait無序等待SingleOutputStreamOperator<CategoryInfo> result1 = AsyncDataStream.unorderedWait(categoryDS, new ASyncIOFunction1(), 1000, TimeUnit.SECONDS, 10);//方式二：MySQL中同步client+線程池模擬異步IO//unorderedWait無序等待SingleOutputStreamOperator<CategoryInfo> result2 = AsyncDataStream.unorderedWait(categoryDS, new ASyncIOFunction2(), 1000, TimeUnit.SECONDS, 10);//4.Sinkresult1.print("方式一：Java-vertx中提供的異步client實現異步IO \n");result2.print("方式二：MySQL中同步client+線程池模擬異步IO \n");//5.executeenv.execute();}
}@Data
@NoArgsConstructor
@AllArgsConstructor
class CategoryInfo {private Integer id;private String name;
}class MysqlSyncClient {private static transient Connection connection;private static final String JDBC_DRIVER = "com.mysql.jdbc.Driver";private static final String URL = "jdbc:mysql://localhost:3306/bigdata";private static final String USER = "root";private static final String PASSWORD = "root";static {init();}private static void init() {try {Class.forName(JDBC_DRIVER);} catch (ClassNotFoundException e) {System.out.println("Driver not found!" + e.getMessage());}try {connection = DriverManager.getConnection(URL, USER, PASSWORD);} catch (SQLException e) {System.out.println("init connection failed!" + e.getMessage());}}public void close() {try {if (connection != null) {connection.close();}} catch (SQLException e) {System.out.println("close connection failed!" + e.getMessage());}}public CategoryInfo query(CategoryInfo category) {try {String sql = "select id,name from t_category where id = "+ category.getId();Statement statement = connection.createStatement();ResultSet rs = statement.executeQuery(sql);if (rs != null && rs.next()) {category.setName(rs.getString("name"));}} catch (SQLException e) {System.out.println("query failed!" + e.getMessage());}return category;}
}/*** 方式一：Java-vertx中提供的異步client實現異步IO*/
class ASyncIOFunction1 extends RichAsyncFunction<CategoryInfo, CategoryInfo> {private transient SQLClient mySQLClient;@Overridepublic void open(Configuration parameters) throws Exception {JsonObject mySQLClientConfig = new JsonObject();mySQLClientConfig.put("driver_class", "com.mysql.jdbc.Driver").put("url", "jdbc:mysql://localhost:3306/bigdata").put("user", "root").put("password", "root").put("max_pool_size", 20);VertxOptions options = new VertxOptions();options.setEventLoopPoolSize(10);options.setWorkerPoolSize(20);Vertx vertx = Vertx.vertx(options);//根據上面的配置參數獲取異步請求客戶端mySQLClient = JDBCClient.createNonShared(vertx, mySQLClientConfig);}//使用異步客戶端發送異步請求@Overridepublic void asyncInvoke(CategoryInfo input, ResultFuture<CategoryInfo> resultFuture) throws Exception {mySQLClient.getConnection(new Handler<AsyncResult<SQLConnection>>() {@Overridepublic void handle(AsyncResult<SQLConnection> sqlConnectionAsyncResult) {if (sqlConnectionAsyncResult.failed()) {return;}SQLConnection connection = sqlConnectionAsyncResult.result();connection.query("select id,name from t_category where id = " +input.getId(), new Handler<AsyncResult<io.vertx.ext.sql.ResultSet>>() {@Overridepublic void handle(AsyncResult<io.vertx.ext.sql.ResultSet> resultSetAsyncResult) {if (resultSetAsyncResult.succeeded()) {List<JsonObject> rows = resultSetAsyncResult.result().getRows();for (JsonObject jsonObject : rows) {CategoryInfo categoryInfo = new CategoryInfo(jsonObject.getInteger("id"), jsonObject.getString("name"));resultFuture.complete(Collections.singletonList(categoryInfo));}}}});}});}@Overridepublic void close() throws Exception {mySQLClient.close();}@Overridepublic void timeout(CategoryInfo input, ResultFuture<CategoryInfo> resultFuture) throws Exception {System.out.println("async call time out!");input.setName("未知");resultFuture.complete(Collections.singleton(input));}
}/*** 方式二：同步調用+線程池模擬異步IO*/
class ASyncIOFunction2 extends RichAsyncFunction<CategoryInfo, CategoryInfo> {private transient MysqlSyncClient client;private ExecutorService executorService;//線程池@Overridepublic void open(Configuration parameters) throws Exception {super.open(parameters);client = new MysqlSyncClient();executorService = new ThreadPoolExecutor(10, 10, 0L, TimeUnit.MILLISECONDS, new LinkedBlockingQueue<Runnable>());}//異步發送請求@Overridepublic void asyncInvoke(CategoryInfo input, ResultFuture<CategoryInfo> resultFuture) throws Exception {executorService.execute(new Runnable() {@Overridepublic void run() {resultFuture.complete(Collections.singletonList((CategoryInfo) client.query(input)));}});}@Overridepublic void close() throws Exception {}@Overridepublic void timeout(CategoryInfo input, ResultFuture<CategoryInfo> resultFuture) throws Exception {System.out.println("async call time out!");input.setName("未知");resultFuture.complete(Collections.singleton(input));}
}

異步IO讀取Redis數據-沒必要！！！

package cn.it.extend;import io.vertx.core.Vertx;
import io.vertx.core.VertxOptions;
import io.vertx.redis.RedisClient;
import io.vertx.redis.RedisOptions;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.TimeCharacteristic;
import org.apache.flink.streaming.api.datastream.AsyncDataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.async.ResultFuture;
import org.apache.flink.streaming.api.functions.async.RichAsyncFunction;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import redis.clients.jedis.Jedis;
import redis.clients.jedis.JedisPool;
import redis.clients.jedis.JedisPoolConfig;import java.util.Collections;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.TimeUnit;
import java.util.function.Supplier;
/**
使用異步IO訪問redis
hset?AsyncReadRedis?beijing?1
hset?AsyncReadRedis?shanghai?2
hset?AsyncReadRedis?guangzhou?3
hset?AsyncReadRedis?shenzhen?4
hset?AsyncReadRedis?hangzhou?5
hset?AsyncReadRedis?wuhan?6
hset?AsyncReadRedis?chengdu?7
hset?AsyncReadRedis?tianjin?8
hset?AsyncReadRedis?chongqing?9city.txt
1,beijing
2,shanghai
3,guangzhou
4,shenzhen
5,hangzhou
6,wuhan
7,chengdu
8,tianjin
9,chongqing*/
public class AsyncIODemo_Redis {public static void main(String[] args) throws Exception {StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime);DataStreamSource<String> lines = env.readTextFile("data/input/city.txt");SingleOutputStreamOperator<String> result1 = AsyncDataStream.orderedWait(lines, new AsyncRedis(), 10, TimeUnit.SECONDS, 1);SingleOutputStreamOperator<String> result2 = AsyncDataStream.orderedWait(lines, new AsyncRedisByVertx(), 10, TimeUnit.SECONDS, 1);result1.print().setParallelism(1);result2.print().setParallelism(1);env.execute();}
}
/*** 使用異步的方式讀取redis的數據*/
class AsyncRedis extends RichAsyncFunction<String, String> {//定義redis的連接池對象private JedisPoolConfig config = null;private static String ADDR = "localhost";private static int PORT = 6379;//等待可用連接的最大時間，單位是毫秒，默認是-1，表示永不超時，如果超過等待時間，則會拋出異常private static int TIMEOUT = 10000;//定義redis的連接池實例private JedisPool jedisPool = null;//定義連接池的核心對象private Jedis jedis = null;//初始化redis的連接@Overridepublic void open(Configuration parameters) throws Exception {super.open(parameters);//定義連接池對象屬性配置config = new JedisPoolConfig();//初始化連接池對象jedisPool = new JedisPool(config, ADDR, PORT, TIMEOUT);//實例化連接對象（獲取一個可用的連接）jedis = jedisPool.getResource();}@Overridepublic void close() throws Exception {super.close();if(jedis.isConnected()){jedis.close();}}//異步調用redis@Overridepublic void asyncInvoke(String input, ResultFuture<String> resultFuture) throws Exception {System.out.println("input:"+input);//發起一個異步請求，返回結果CompletableFuture.supplyAsync(new Supplier<String>() {@Overridepublic String get() {String[] arrayData = input.split(",");String name = arrayData[1];String value = jedis.hget("AsyncReadRedis", name);System.out.println("output:"+value);return ?value;}}).thenAccept((String dbResult)->{//設置請求完成時的回調，將結果返回resultFuture.complete(Collections.singleton(dbResult));});}//連接超時的時候調用的方法，一般在該方法中輸出連接超時的錯誤日志，如果不重新該方法，連接超時后會拋出異常@Overridepublic void timeout(String input, ResultFuture<String> resultFuture) throws Exception {System.out.println("redis connect timeout!");}
}
/*** 使用高性能異步組件vertx實現類似于連接池的功能，效率比連接池要高* 1）在java版本中可以直接使用* 2）如果在scala版本中使用的話，需要scala的版本是2.12+*/
class AsyncRedisByVertx extends RichAsyncFunction<String,String> {//用transient關鍵字標記的成員變量不參與序列化過程private transient RedisClient redisClient;//獲取連接池的配置對象private JedisPoolConfig config = null;//獲取連接池JedisPool jedisPool = null;//獲取核心對象Jedis jedis = null;//Redis服務器IPprivate static String ADDR = "localhost";//Redis的端口號private static int PORT = 6379;//訪問密碼private static String AUTH = "XXXXXX";//等待可用連接的最大時間，單位毫秒，默認值為-1，表示永不超時。如果超過等待時間，則直接拋出JedisConnectionException；private static int TIMEOUT = 10000;private static final Logger logger = LoggerFactory.getLogger(AsyncRedis.class);//初始化連接@Overridepublic void open(Configuration parameters) throws Exception {super.open(parameters);config = new JedisPoolConfig();jedisPool = new JedisPool(config, ADDR, PORT, TIMEOUT);jedis = jedisPool.getResource();RedisOptions config = new RedisOptions();config.setHost(ADDR);config.setPort(PORT);VertxOptions vo = new VertxOptions();vo.setEventLoopPoolSize(10);vo.setWorkerPoolSize(20);Vertx vertx = Vertx.vertx(vo);redisClient = RedisClient.create(vertx, config);}//數據異步調用@Overridepublic void asyncInvoke(String input, ResultFuture<String> resultFuture) throws Exception {System.out.println("input:"+input);String[] split = input.split(",");String name = split[1];// 發起一個異步請求redisClient.hget("AsyncReadRedis", name, res->{if(res.succeeded()){String result = res.result();if(result== null){resultFuture.complete(null);return;}else {// 設置請求完成時的回調: 將結果傳遞給?collectorresultFuture.complete(Collections.singleton(result));}}else if(res.failed()) {resultFuture.complete(null);return;}});}@Overridepublic void timeout(String input, ResultFuture resultFuture) throws Exception {}@Overridepublic void close() throws Exception {super.close();if (redisClient != null) {redisClient.close(null);}}
}

擴展閱讀 原理深入

AsyncDataStream

AsyncDataStream是一個工具類，用于將AsyncFunction應用于DataStream，AsyncFunction發出的并發請求都是無序的，該順序基于哪個請求先完成，為了控制結果記錄的發出順序，flink提供了兩種模式，分別對應AsyncDataStream的兩個靜態方法，OrderedWait和unorderedWait

• orderedWait（有序）：消息的發送順序與接收到的順序相同（包括 watermark ），也就是先進先出。
• unorderWait（無序）：
  • 在ProcessingTime中，完全無序，即哪個請求先返回結果就先發送（最低延遲和最低消耗）。
  • 在EventTime中，以watermark為邊界，介于兩個watermark之間的消息可以亂序，但是watermark和消息之間不能亂序，這樣既認為在無序中又引入了有序，這樣就有了與有序一樣的開銷。

AsyncDataStream.(un)orderedWait?的主要工作就是創建了一個?AsyncWaitOperator。AsyncWaitOperator?是支持異步 IO 訪問的算子實現，該算子會運行?AsyncFunction?并處理異步返回的結果，其內部原理如下圖所示。

如圖所示，AsyncWaitOperator?主要由兩部分組成：

• StreamElementQueue?
• Emitter

StreamElementQueue 是一個 Promise 隊列，所謂 Promise 是一種異步抽象表示將來會有一個值（海底撈排隊給你的小票），這個隊列是未完成的 Promise 隊列，也就是進行中的請求隊列。Emitter 是一個單獨的線程，負責發送消息（收到的異步回復）給下游。

圖中E5表示進入該算子的第五個元素（”Element-5”）

• 在執行過程中首先會將其包裝成一個 “Promise”?P5，然后將P5放入隊列
• 最后調用?AsyncFunction?的?ayncInvoke?方法，該方法會向外部服務發起一個異步的請求，并注冊回調
• 該回調會在異步請求成功返回時調用?AsyncCollector.collect?方法將返回的結果交給框架處理。
• 實際上?AsyncCollector?是一個 Promise ，也就是?P5，在調用?collect?的時候會標記 Promise 為完成狀態，并通知 Emitter 線程有完成的消息可以發送了。
• Emitter 就會從隊列中拉取完成的 Promise ，并從 Promise 中取出消息發送給下游。

???????消息的順序性

上文提到 Async I/O 提供了兩種輸出模式。其實細分有三種模式:

• 有序
• ProcessingTime 無序
• EventTime 無序

Flink 使用隊列來實現不同的輸出模式，并抽象出一個隊列的接口（StreamElementQueue），這種分層設計使得AsyncWaitOperator和Emitter不用關心消息的順序問題。StreamElementQueue有兩種具體實現，分別是?OrderedStreamElementQueue?和UnorderedStreamElementQueue。UnorderedStreamElementQueue?比較有意思，它使用了一套邏輯巧妙地實現完全無序和 EventTime 無序。

• 有序

有序比較簡單，使用一個隊列就能實現。所有新進入該算子的元素（包括 watermark），都會包裝成 Promise 并按到達順序放入該隊列。如下圖所示，盡管P4的結果先返回，但并不會發送，只有?P1?（隊首）的結果返回了才會觸發 Emitter 拉取隊首元素進行發送。

• ProcessingTime 無序

ProcessingTime 無序也比較簡單，因為沒有 watermark，不需要協調 watermark 與消息的順序性，所以使用兩個隊列就能實現，一個?uncompletedQueue?一個?completedQueue。所有新進入該算子的元素，同樣的包裝成 Promise 并放入?uncompletedQueue?隊列，當uncompletedQueue隊列中任意的Promise返回了數據，則將該 Promise 移到?completedQueue?隊列中，并通知 Emitter 消費。如下圖所示：

• EventTime 無序

EventTime 無序類似于有序與 ProcessingTime 無序的結合體。因為有 watermark，需要協調 watermark與消息之間的順序性，所以uncompletedQueue中存放的元素從原先的 Promise 變成了 Promise 集合。

• 如果進入算子的是消息元素，則會包裝成 Promise 放入隊尾的集合中
• 如果進入算子的是 watermark，也會包裝成 Promise 并放到一個獨立的集合中，再將該集合加入到?uncompletedQueue?隊尾，最后再創建一個空集合加到?uncompletedQueue?隊尾
• 這樣，watermark 就成了消息順序的邊界。
• 只有處在隊首的集合中的 Promise 返回了數據，才能將該 Promise 移到completedQueue?
• 隊列中，由 Emitter 消費發往下游。
• 只有隊首集合空了，才能處理第二個集合。

這樣就保證了當且僅當某個 watermark 之前所有的消息都已經被發送了，該 watermark 才能被發送。過程如下圖所示：

總結

以上是生活随笔為你收集整理的2021年大数据Flink（四十六）：扩展阅读 异步IO的全部內容，希望文章能夠幫你解決所遇到的問題。

如果覺得生活随笔網站內容還不錯，歡迎將生活随笔推薦給好友。