vue做单页面网站3322免费域名注册
导读
在流计算中，如果以事件流为主，关联一些维度信息，就需要根据每个事件中的关键信息去数据库执行一次查询。正常的思路可能是通过mapFunction以阻塞的方式查询数据库，等待数据结果返回，然后执行下一个步骤。如果数据库查询时间很长，那有可能会阻塞流计算的整体流程。因此可以考虑异步的方式请求数据库，当数据返回时，该事件再继续执行下面的操作。这样提升了流计算的并发度，但是也增加了数据库的访问以及网络带宽的压力。
1 Flink中的异步IO
在Flink中提供了一种异步IO的模式，不需要使用map函数阻塞式的加载数据，而是使用异步方法同时处理大量请求。不过这就需要数据库支持异步请求，如果不支持异步请求也可以手动维护线程池调用，只不过效率上没有原生的异步client更高效。比如Mysql可以通过Vertx支持异步查询，HBase2.x也支持异步查询。
一般要实现Flink的异步查询需要自定义几个方法：
class MyAsyncReq extends RichAsyncFunction{
@Override
public void open(..) throws Exception {}
@Override
public void close() throws Exception {}
@Override
public void asyncInvoke(..) throws Exception {}
}
其中open中需要定义连接或者连接池，close中进行释放，asyncInvoke执行异步查询。
AsyncDataStream.unorderedWait(
stream,
new MyAsyncReq(),
1000,
TimeUnit.MILLISECONDS,
100);
使用的使用执行下面的方法即可，
stream为主要的事件流，
myasyncreq是异步IO类，
1000为异步请求的超时时间，
100是同时进行异步请求的最大数量
另外，由于是异步请求，所以可能请求结束后顺序与原来的顺序就不一致了。使用unordered时会以异步请求结束的时间为准，ordered会以事件时间为准。
2 基于Vertx实现的Mysql异步IO
如果外部数据源是Mysql，一般的jdbc连接都是同步机制的，看浪尖大大的文章，推荐了一个异步JDBC组件——Vertx，下面就以Vertx为例作为异步IO的Client。
maven引入除flink之外其他的jar：
mysql
mysql-connector-java
8.0.13
io.vertx
vertx-jdbc-client
3.8.3
io.vertx
vertx-core
3.8.3
先在open中创建SQLClient，它内部维护了自己的异步请求服务；然后再close中关闭client；在asyncInvoke中调用获取connection，执行查询，并释放连接。
public class JDBCAsyncFunction extends RichAsyncFunction {
private SQLClient client;
@Override
public void open(Configuration parameters) throws Exception {
Vertx vertx = Vertx.vertx(new VertxOptions()
.setWorkerPoolSize(10)
.setEventLoopPoolSize(10));
JsonObject config = new JsonObject()
.put("url", "jdbc:mysql://xx:3306/base")
.put("driver_class", "com.mysql.cj.jdbc.Driver")
.put("max_pool_size", 10)
.put("user", "x")
.put("password", "x");
client = JDBCClient.createShared(vertx, config);
}
@Override
public void close() throws Exception {
client.close();
}
@Override
public void asyncInvoke(Click input, ResultFuture resultFuture) throws Exception {
client.getConnection(conn -> {
if (conn.failed()) {
return;
}
final SQLConnection connection = conn.result();
connection.query("select id, name from t where id = " + input.getId(), res2 -> {
ResultSet rs = new ResultSet();
if (res2.succeeded()) {
rs = res2.result();
}
List stores = new ArrayList<>();
for (JsonObject json : rs.getRows()) {
Store s = new Store();
s.setId(json.getInteger("id"));
s.setName(json.getString("name"));
stores.add(s);
}
connection.close();
resultFuture.complete(stores);
});
});
}
}
注意，一定要在query的返回调用方法中手动释放connection，不然马上就会报连接池耗尽的异常。使用时就没什么区别了：
AsyncDataStream
.unorderedWait(clicks,new JDBCPoolFunction(), 100,TimeUnit.SECONDS,10)
.print();
3 参考
1 vertx:
2 设计思想参考：