背景

代码生成技术适合在需要动态代码编译的场景中使用。比如大数据计算场景下，经常会要把flink sql 转成实际的执行计划

简单例子

代码是一个string 类型，直接用janino 编译后，就可以得到加载到jvm里的class 类

import org.codehaus.janino.SimpleCompiler;public class JaninoClassExample {public static void main(String[] args) throws Exception {// 创建一个 SimpleCompilerSimpleCompiler compiler = new SimpleCompiler();// 设置 Java 类代码String classCode ="public class HelloWorld {" +"    public String greet(String name) {" +"        return \"Hello, \" + name + \"!\";" +"    }" +"}";// 编译 Java 类代码compiler.cook(classCode);// 获取编译后的类Class<?> helloWorldClass = compiler.getClassLoader().loadClass("HelloWorld");// 创建类的实例并调用方法Object helloWorldInstance = helloWorldClass.getDeclaredConstructor().newInstance();String greeting = (String) helloWorldClass.getMethod("greet", String.class).invoke(helloWorldInstance, "Janino");System.out.println(greeting); // 输出 Hello, Janino!}
}

flink 例子

无group by 的sum 函数最后成的实现代码，这个类AggCodeGenHelper 里

public class LocalNoGroupingAggregateWithoutKeys$6 extends org.apache.flink.table.runtime.operators.TableStreamOperatorimplements org.apache.flink.streaming.api.operators.OneInputStreamOperator, org.apache.flink.streaming.api.operators.BoundedOneInput {private final Object[] references; // 引用对象数组（外部依赖）int hash_agg0_sum; // 聚合和boolean hash_agg0_sumIsNull; // 标记聚合和是否为nullorg.apache.flink.table.data.GenericRowData valueRow$5 = new org.apache.flink.table.data.GenericRowData(1); // 保存输出值的行数据private boolean hasInput = false; // 标记是否有输入org.apache.flink.streaming.runtime.streamrecord.StreamRecord element = new org.apache.flink.streaming.runtime.streamrecord.StreamRecord((Object) null); // 输入元素private final org.apache.flink.streaming.runtime.streamrecord.StreamRecord outElement = new org.apache.flink.streaming.runtime.streamrecord.StreamRecord(null); // 输出元素// 构造函数public LocalNoGroupingAggregateWithoutKeys$6(Object[] references,org.apache.flink.streaming.runtime.tasks.StreamTask task,org.apache.flink.streaming.api.graph.StreamConfig config,org.apache.flink.streaming.api.operators.Output output,org.apache.flink.streaming.runtime.tasks.ProcessingTimeService processingTimeService) throws Exception {this.references = references;this.setup(task, config, output); // 设置算子if (this instanceof org.apache.flink.streaming.api.operators.AbstractStreamOperator) {((org.apache.flink.streaming.api.operators.AbstractStreamOperator) this).setProcessingTimeService(processingTimeService); // 设置处理时间服务}}@Overridepublic void open() throws Exception {super.open(); // 调用父类的open方法}@Overridepublic void processElement(org.apache.flink.streaming.runtime.streamrecord.StreamRecord element) throws Exception {org.apache.flink.table.data.RowData in1 = (org.apache.flink.table.data.RowData) element.getValue(); // 获取输入行数据int field$0; // 输入字段的值boolean isNull$0; // 标记字段值是否为nullboolean isNull$1; // 中间null标记int result$2; // 中间结果if (!hasInput) { // 如果没有输入hasInput = true;// 初始化聚合缓冲区hash_agg0_sumIsNull = true;hash_agg0_sum = -1;}isNull$0 = in1.isNullAt(0); // 检查第一个字段是否为nullfield$0 = -1;if (!isNull$0) {field$0 = in1.getInt(0); // 获取第一个字段的整数值}int result$4 = -1; // 中间结果boolean isNull$4; // 中间null标记if (isNull$0) {// 处理null的情况isNull$4 = hash_agg0_sumIsNull;if (!isNull$4) {result$4 = hash_agg0_sum;}} else {int result$3 = -1; // 中间结果boolean isNull$3; // 中间null标记if (hash_agg0_sumIsNull) {// 处理聚合和为null的情况isNull$3 = isNull$0;if (!isNull$3) {result$3 = field$0;}} else {// 计算和isNull$1 = hash_agg0_sumIsNull || isNull$0;result$2 = -1;if (!isNull$1) {result$2 = (int) (hash_agg0_sum + field$0); // 将字段值加到聚合和上}isNull$3 = isNull$1;if (!isNull$3) {result$3 = result$2;}}isNull$4 = isNull$3;if (!isNull$4) {result$4 = result$3;}}hash_agg0_sumIsNull = isNull$4;if (!isNull$4) {// 复制结果值hash_agg0_sum = result$4;}}@Overridepublic void endInput() throws Exception {int field$0;boolean isNull$0;boolean isNull$1;int result$2;if (hasInput) {if (hash_agg0_sumIsNull) {valueRow$5.setField(0, null); // 如果聚合和为null，则将输出字段设置为null} else {valueRow$5.setField(0, hash_agg0_sum); // 将输出字段设置为聚合和}output.collect(outElement.replace(valueRow$5)); // 收集输出元素}}@Overridepublic void finish() throws Exception {super.finish(); // 调用父类的finish方法}@Overridepublic void close() throws Exception {super.close(); // 调用父类的close方法}
}