聊聊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