flink rich function的open和close方法执行时机

我们在使用flink的时候，经常会有自定义函数的时候，我们可以继承相关的richXXXFunction类，这个类里面会有open，close方法进行相关初始化和关闭的操作，那么这些方法是什么时候执行的呢？带着这个问题，我们以自定义SourceFunction为例，进行研究。

我们可以自定义source，也比较方便，extends RichSourceFunction 这个类就可以实现，下面的例子就是我们定义一个mysql的source

public class MySource extends RichSourceFunction<Student> {PreparedStatement ps;private Connection connection;/*** open() 方法中建立连接，这样不用每次 invoke 的时候都要建立连接和释放连接。** @param parameters* @throws Exception*/@Overridepublic void open(Configuration parameters) throws Exception {super.open(parameters);connection = getConnection();String sql = "select * from stu;";ps = connection.prepareStatement(sql);}/*** 程序执行完毕就可以进行，关闭连接和释放资源的动作了** @throws Exception*/@Overridepublic void close() throws Exception {super.close();if (connection != null) { //关闭连接和释放资源connection.close();}if (ps != null) {ps.close();}}/*** DataStream 调用一次 run() 方法用来获取数据** @param ctx* @throws Exception*/@Overridepublic void run(SourceContext<Student> ctx) throws Exception {ResultSet resultSet = ps.executeQuery();while (resultSet.next()) {Student student =new Student(resultSet.getString("name"),resultSet.getInt("age"));ctx.collect(student);}}@Overridepublic void cancel() {}private static Connection getConnection() {Connection con = null;try {Class.forName("com.mysql.jdbc.Driver");con =DriverManager.getConnection("jdbc:mysql://localhost:3306/hyhdb?useUnicode=true&characterEncoding=UTF-8", "root", "root");} catch (Exception e) {System.out.println("-----------mysql get connection has exception , msg = " + e.getMessage());}return con;}
}

那么这个例子中的open和close方法什么时候执行呢？

我们可以通过源码了解到，AbstractUdfStreamOperator 这个类中的open和close方法调用的，

 @Overridepublic void open() throws Exception {super.open();FunctionUtils.openFunction(userFunction, new Configuration());}@Overridepublic void close() throws Exception {super.close();functionsClosed = true;FunctionUtils.closeFunction(userFunction);}

我们可以从代码看到open和close方法里面又调用了FunctionUtils的相关方法,FunctionUtils这个类是一个工具类，里面的代码实现如下：

     public static void openFunction(Function function, Configuration parameters) throws Exception {if (function instanceof RichFunction) {RichFunction richFunction = (RichFunction)function;richFunction.open(parameters);}}public static void closeFunction(Function function) throws Exception {if (function instanceof RichFunction) {RichFunction richFunction = (RichFunction)function;richFunction.close();}}

我们看到FunctionUtils里的相关方法，最后判断函数是不是RichFunction类型，如果是的话，那么久调用我们自定义函数的open和close方法。

通过代码我们知道最尾端的调用过程，那么AbstractUdfStreamOperator 这个类中的open和close方法是在什么地方被调用的呢？通过源码的调用过程，我们发现StreamTask类中的

 private void openAllOperators() throws Exception {for (StreamOperator<?> operator : operatorChain.getAllOperators()) {if (operator != null) {operator.open();}}}
private void closeAllOperators() throws Exception {// We need to close them first to last, since upstream operators in the chain might emit// elements in their close methods.StreamOperator<?>[] allOperators = operatorChain.getAllOperators();for (int i = allOperators.length - 1; i >= 0; i--) {StreamOperator<?> operator = allOperators[i];if (operator != null) {operator.close();}}}

我们知道我们的开发的程序，最后会转换成相应的task任务，在work上执行，那么flink的框架中，最后会调用到StreamTask，在这个类里面进行操作链的open和close方法执行。

关于操作链operatorChain的相关内容，后续进行分析

flink rich function的open和close方法执行时机相关推荐

hive初始化元数据的时候出现 Error：FUNCTION ‘NUCLEUS_ASCII‘ already exists解决方法
hive初始化元数据的时候出现 Error:FUNCTION 'NUCLEUS_ASCII' already exists解决方法执行 schematool --initSchema -dbType ...
Flink 算子Function实例化的坑
问题回顾关于一段代码: object MySingleObj{// 陷阱:// 单例对象中一个是可变引用,一个是可变数组var str:String = _val list = new ListBu ...
[Vue warn]: Failed to mount component: template or render function not defined. 错误解决方法
[Vue warn]: Failed to mount component: template or render function not defined. 错误解决方法参考文章: (1)[Vue ...
【JQuery】jQuery(document).ready(function($) { });的几种表示方法及load和ready的区别
jQuery中处理加载时机的几种方式第一种: jQuery(document).ready(function() {alert("你好"); }); //或 $(document ...
extjs 方法执行顺序_透析Extjs的Ext.js源码（二）能在定义时就能执行的方法的写法 function(){...}...
/** * 第二部分:能在定义时就能执行的方法的写法 function(){...}(); */ /** * 一.普通的方法的定义与执行 */ // 1-1.普通的方法定义,不带返回值的情况 fun ...
php undefined function 几个函数,PHP Fatal error: Call to undefined function 函数名() in 解决方法...
PHP Fatal error: Call to undefined function 函数名() in 解决方法发布于 2015-01-14 08:55:15 | 219 次阅读 | 评论: 1 ...
vue在一个方法执行完后执行另一个方法
vue在一个方法执行完后执行另一个方法用Promise来实现. Promise是ES6的新特性,用于处理异步操作逻辑,用过给Promise添加then和catch函数,处理成功和失败的情况 ES7中 ...
微信小程序-wx.createInnerAudioContext的方法执行多次问题
微信小程序-wx.createInnerAudioContext的方法执行多次问题在项目中用wx.createInnerAudioContext做语音播放这一块,测试的时候发现第一次播放的时候onP ...
Vue父组件方法和子组件方法执行优先顺序
首先,我遇到了一个两个分离的前端项目,该父类组件调用子类组件时,不论是写在mounted还是created中,明明两个方法应该按照顺序执行,和我想的完全不一样,子组件的方法优先执行,父类的方法永远在子 ...

flink rich function的open和close方法执行时机

flink rich function的open和close方法执行时机相关推荐

最新文章

热门文章