举个例子，你如果在算子中用到了MapFunction，写你想要对map中的每条处理记录都查询下数据库丰富当前处理记录。那么此时就可以使用异步访问系统
- 延迟问题。每次请求外部数据库都会带来很多的延迟(如果不采用异步的话MapFunction的大部分时间都是在等待查询结果)
- 而Flink提供的AsyncFunction(异步)可以有效的降低I/O所带来的延迟问题。
- 官网是这么说的：“AsyncFunction函数能够同时发出多个查询并对其结果进行异步处理”
- 我们可以抽象的概括为：传统的请求是在map中每循环一次都会发出一次请求并等待返回结果，然后再进行逻辑计算。如下方左图
- 但是通过异步的话，类似于是将集合中的数据统一进行一次并发请求，然后读取返回的结果，再进行逻辑计算。如下方右图
优点
– 1. AsyncFunction与检查点进行了良好的集成，即所有正在等待返回结果的记录都会被写入到检查点并支持在恢复时重新发送请求。
– 2. AsyncFunction也可以在事件事件处理模式下正确地工作，因为即使允许乱序，它也能够保证记录不会被水位线超过。
注意：
(如果外部数据源是Mysql，一般的jdbc连接都是同步机制的，因此我们可以使用异步的JDBC组件-Vertx)package flink
import java.sql.{Connection, DriverManager, Statement}
import java.util.concurrent.Executors
import org.apache.flink.configuration.Configuration
import org.apache.flink.streaming.api.functions.async.{ResultFuture, RichAsyncFunction}
import org.slf4j.{Logger, LoggerFactory}
import scala.collection.mutable.ListBuffer
import scala.concurrent.{ExecutionContext, Future}
import scala.util.{Failure, Success}
class AsyncFunction_ceshi extends RichAsyncFunction[class_infor, (Int, String)] {
val logger: Logger = LoggerFactory.getLogger(this.getClass)
private lazy val cachingPoolEXECTX = ExecutionContext.fromExecutor(Executors.newCachedThreadPool())
private lazy val direExec = ExecutionContext.fromExecutor(
org.apache.flink.runtime.concurrent.Executors.directExecutor()
)
var pre: Statement = _
var connect: Connection = _
// var result: ResultSet = _
override def open(parameters: Configuration): Unit = {
Class.forName("com.mysql.jdbc.Driver")
connect = DriverManager.getConnection("jdbc:mysql://master:3306/friends", "root", "msb_mk")
pre = connect.createStatement()
}
override def asyncInvoke(input: class_infor, resultFuture: ResultFuture[(Int, String)]): Unit = {
val age = input.age
val name: Future[String] = Future {
val result = pre.executeQuery(s"select name,age from ceshi where age = $age")
// Thread.sleep(2000)
val name = if (result.next()) result.getString("name") else "UNKNOW"
name
}(cachingPoolEXECTX)
import scala.collection.JavaConverters._
name.onComplete {
case Success(value) => resultFuture.complete(ListBuffer((age, value)).asJava)
// case Success(value) => resultFuture.complete(Collections.singletonList(age, value))
case Failure(value) => resultFuture.completeExceptionally(value)
}(direExec)
}
override def timeout(input: class_infor, resultFuture: ResultFuture[(Int, String)]): Unit = {
//在上面的asyncInvoke我自发的让其出现延迟问题；如果出现上述延迟问题则会在本函数中进行处理
logger.warn("Async function for mysql timeout... ...")
import scala.collection.JavaConverters._
resultFuture.complete(ListBuffer((input.age, "超时")).asJava)
}
override def close(): Unit = {
try {
// result.close()
pre.close()
connect.close()
} catch {
case e: Exception => e.printStackTrace()
}
}
}
package flink
import java.util.concurrent.TimeUnit
import org.apache.flink.streaming.api.datastream.{AsyncDataStream, DataStream, SingleOutputStreamOperator}
case class class_infor(name: String, age: Int)
object 异步访问外部系统 {
def main(args: Array[String]): Unit = {
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment
val streamLocal = StreamExecutionEnvironment.createLocalEnvironment(2)
val data: DataStream[class_infor] = streamLocal.fromElements(class_infor("a", 12), class_infor("b", 2))
val resultStream: SingleOutputStreamOperator[(Int, String)] = AsyncDataStream.orderedWait(data,
new AsyncFunction_ceshi(),
5, TimeUnit.SECONDS,
30) //最多30个并发请求
resultStream.print()
streamLocal.execute("ceshi")
}
}