聊聊maxwell的Scripting

编程

本文主要研究一下maxwell的Scripting

Scripting

maxwell-1.25.1/src/main/java/com/zendesk/maxwell/scripting/Scripting.java

public class Scripting {

static final Logger LOGGER = LoggerFactory.getLogger(Scripting.class);

private final ScriptObjectMirror processRowFunc, processHeartbeatFunc, processDDLFunc;

private ScriptObjectMirror getFunc(ScriptEngine engine, String fName, String filename) {

ScriptObjectMirror f = (ScriptObjectMirror) engine.get(fName);

if ( f == null )

return null;

else if ( !f.isFunction() ) {

throw new RuntimeException("Expected " + fName + " to be a function!");

} else {

LOGGER.info("using function " + fName + " from " + filename);

}

return f;

}

public Scripting(String filename) throws IOException, ScriptException, NoSuchMethodException {

ScriptEngineManager manager = new ScriptEngineManager();

ScriptEngine engine = manager.getEngineByName("nashorn");

String externJS = new String(Files.readAllBytes(Paths.get(filename)));

engine.put("logger", LOGGER);

engine.eval(externJS);

processRowFunc = getFunc(engine, "process_row", filename);

processHeartbeatFunc = getFunc(engine, "process_heartbeat", filename);

processDDLFunc = getFunc(engine, "process_ddl", filename);

if ( processRowFunc == null && processHeartbeatFunc == null && processDDLFunc == null )

LOGGER.warn("expected " + filename + " to define at least one of: process_row,process_heartbeat,process_ddl");

}

public void invoke(RowMap row) {

if ( row instanceof HeartbeatRowMap && processHeartbeatFunc != null )

processHeartbeatFunc.call(null, new WrappedHeartbeatMap((HeartbeatRowMap) row));

else if ( row instanceof DDLMap && processDDLFunc != null )

processDDLFunc.call(null, new WrappedDDLMap((DDLMap) row));

else if ( row instanceof RowMap && processRowFunc != null )

processRowFunc.call(null, new WrappedRowMap(row));

}

private static ThreadLocal<ScriptEngine> stringifyEngineThreadLocal = ThreadLocal.withInitial(() -> {

ScriptEngineManager manager = new ScriptEngineManager();

return manager.getEngineByName("nashorn");

});

public static String stringify(ScriptObjectMirror mirror) throws ScriptException {

ScriptObjectMirror json = (ScriptObjectMirror) stringifyEngineThreadLocal.get().eval("JSON");

return (String) json.callMember("stringify", mirror);

}

}

  • Scripting定义了processRowFunc、processHeartbeatFunc、processDDLFunc属性;其构造器接收filename参数,它创建ScriptEngineManager,然后获取名为nashorn的ScriptEngine,之后通过filename获取externJS,并执行engine.eval(externJS),之后通过getFunc方法初始化化processRowFunc、processHeartbeatFunc、processDDLFunc;其getFunc方法通过engine.get(fName)获取ScriptObjectMirror;其invoke方法针对HeartbeatRowMap执行processHeartbeatFunc.call,针对DDLMap执行processDDLFunc.call,针对RowMap执行processRowFunc.call

MaxwellConfig

maxwell-1.25.1/src/main/java/com/zendesk/maxwell/MaxwellConfig.java

public class MaxwellConfig extends AbstractConfig {

static final Logger LOGGER = LoggerFactory.getLogger(MaxwellConfig.class);

//......

public String javascriptFile;

public Scripting scripting;

//......

public void validate() {

validatePartitionBy();

validateFilter();

//......

if ( this.javascriptFile != null ) {

try {

this.scripting = new Scripting(this.javascriptFile);

} catch ( Exception e ) {

LOGGER.error("Error setting up javascript: ", e);

System.exit(1);

}

}

}

//......

}

  • MaxwellConfig的validate方法在javascriptFile不为null的时候会创建Scripting,如果出现异常则执行System.exit(1)

小结

Scripting定义了processRowFunc、processHeartbeatFunc、processDDLFunc属性;其构造器接收filename参数,它创建ScriptEngineManager,然后获取名为nashorn的ScriptEngine,之后通过filename获取externJS,并执行engine.eval(externJS),之后通过getFunc方法初始化化processRowFunc、processHeartbeatFunc、processDDLFunc;其getFunc方法通过engine.get(fName)获取ScriptObjectMirror;其invoke方法针对HeartbeatRowMap执行processHeartbeatFunc.call,针对DDLMap执行processDDLFunc.call,针对RowMap执行processRowFunc.call

doc

  • Scripting

以上是 聊聊maxwell的Scripting 的全部内容, 来源链接: utcz.com/z/516311.html

回到顶部