記錄工作中groovy動態生成Flink任務

工作中的痛點：有一個計算的任務，需要配置成前端配置好一些簡單的信息，例如名字,計算間隔，計算規則（這個是需要提前寫好，開放給用戶選擇的），然后通過提交到我們的計算引擎中心生成對應的任務jar包提交到服務器上去計算（Flink+groovy)

 

我這里僅僅記錄流程和關鍵點。

 

1：將前端傳入引擎的信息進行封裝

// 解析定義的參數，並封裝運算腳本所需的其他參數值(key-value形式)。 
Map<String, Object> ruleParams = parseParamsToMap(taskDef.getRuleParams());

//接下來就是個中封裝數據 等等 ，下面粗略的給了一個腳本內容 內容就是計算的方法 scriptContentInfo就是計算方法中需要的一些參數
 ruleParams.put("ScriptContentInfo",taskDef.ScriptContentInfo);

2：自行構建一個groovy的腳本類

class GetRuleTask {

    static void getJobInfo(String ruleName, String scriptContentBody, Map<String, Object> inputParams, String gScriptPath){

// gScriptPath 是腳本所在的路徑 通常是idea是放在resource下面,ruleName是創建動態任務類的方法名，scripteContenBody就是類的方法體 inputParams就是方法中使用到的各種參數
        GroovyScriptEngine engine = new GroovyScriptEngine(gScriptPath);
        Binding bind = new Binding();
        bind.setVariable("ScriptContentBody",ScriptContentBody);
        bind.setVariable("ruleName", ruleName);
        bind.setVariable("inputParams", inputParams);
//CodeLoader下面會介紹，是一個將參數和類綁定在一起生成完整類的操作
        engine.run("CodeLoader.groovy",bind);

        AnnotationConfigApplicationContext cxt = new AnnotationConfigApplicationContext();
        cxt.scan("com.my.boke.boven.loader");
        cxt.refresh();
//這個地方的RuleSet.class 下面會介紹，就是動態生成的類
        RuleSet rules  = cxt.getBean(RuleSet.class);
//調用動態類的動態方法
        rules.invokeMethod(ruleName, null);
    
    

 

3：提前創建一個空類 RuleSet

package com.my.boke.boven.loader

class RuleSet {

}

4:CodeLoader介紹

ScriptEngineManager factory = new ScriptEngineManager()
ScriptEngine engine = factory.getEngineByName("groovy")
// 綁定規則的輸入參數
Bindings binds = engine.createBindings();
for (Entry<String, Object> entry : inputParams.entrySet()) {
    binds.put(entry.getKey(), entry.getValue());
}
//導入參數所用的到的所有包，其實也不多，也就10幾個。
def code_import = """import com.my.boke.boven.common.*;
import com.alibaba.fastjson.JSONObject;
...
...
...
;\n"""

//這個地方就是將方法包和方法體進行一個綁定，生成一個完整類
Reader reader = new StringReader(code_import + methodBody);
RuleSet.metaClass.'static'."${methodName}" = {
    ->
    engine.eval(reader, binds)
}

 

我故意將2步驟寫在了前面，就是想體現一點，其實我們的服務運行當中，如果沒有啟動任務的時候，類中的方法是不存在的，只有當我啟動的時候才會去動態的生成。

到這一步只需要remote將jar包提交到flink即可。

以上只提供思路，具體實現根據自己業務實現。