揭秘字节跳动埋点数据实时动态处理引擎（附源码）

 env.addSource(new UserDefinedSource())
    .process(new ProcessFunction<ClientLogSource, ClientLogSink>() {
        // 动态规则配置中心
        private ZkBasedConfigCenter zkBasedConfigCenter;
        // kafka producer 管理中心
        private KafkaProducerCenter kafkaProducerCenter;

        @Override
        public void open(Configuration parameters) throws Exception {
            super.open(parameters);
            this.zkBasedConfigCenter = ZkBasedConfigCenter.getInstance();
            this.kafkaProducerCenter = KafkaProducerCenter.getInstance();
        }

        @Override
        public void processElement(ClientLogSource clientLogSource, Context context, Collector<ClientLogSink> collector)
                throws Exception {
            
            // 遍历所有的动态规则
            this.zkBasedConfigCenter.getMap().forEach(new BiConsumer<Long, DynamicProducerRule>() {
                @Override
                public void accept(Long id, DynamicProducerRule dynamicProducerRule) {
                    // 验证该条数据是否符合该条规则
                    if (dynamicProducerRule.eval(clientLogSource)) {
                        // 将符合规则的数据发向对应规则的 topic 中
                        kafkaProducerCenter.send(dynamicProducerRule.getTargetTopic(), clientLogSource.toString());
                    }

                }
            });
        }

        @Override
        public void close() throws Exception {
            super.close();
            // 关闭规则池
            this.zkBasedConfigCenter.close();
            // 关闭 producer 池
            this.kafkaProducerCenter.close();
        }
    });

env.execute();

{
    "id-数值类型 string": {
  "condition-过滤条件": "1==1",
  "targetTopic-目标 topic 名称": "tuzisir1"
 }
 "1": {
  "condition": "clientLogSource.getId() > 300 && clientLogSource.getPage().equals(\"首页\")",
  "targetTopic": "topic_id_bigger_than_300_and_main_page"
 },
 "2": {
  "condition": "clientLogSource.getPage().equals(\"个人主页\")",
  "targetTopic": "topic_profile_page"
 }
}

public class DynamicProducerRule implements Evaluable {

    // 具体过滤规则
    private String condition;

    // 具体写入 topic
    private String targetTopic;

    // 使用 janino 编译的规则过滤器
    private Evaluable evaluable;

    public void init(Long id) {
        try {
            // 使用 janino 初始化规则
            Class<Evaluable> clazz = JaninoUtils.genCodeAndGetClazz(id, targetTopic, condition);
            this.evaluable = clazz.newInstance();
        } catch (Exception e) {
            throw new RuntimeException(e);
        }
    }

    @Override
    public boolean eval(ClientLogSource clientLogSource) {
        return this.evaluable.eval(clientLogSource);
    }

}

public interface Evaluable {

    // 动态规则接口过滤方法
    boolean eval(ClientLogSource clientLogSource);

}

public class ZkBasedConfigCenter {
    
    // zk config 变化监听器
    private TreeCache treeCache;

    // zk 客户端
    private CuratorFramework zkClient;

    private ZkBasedConfigCenter() {
        try {
            open();
        } catch (Exception e) {
            throw new RuntimeException(e);
        }
    }

    // ！！！规则池！！！规则池！！！规则池
    private ConcurrentMap<Long, DynamicProducerRule> map = new ConcurrentHashMap<>();

    private void open() throws Exception {

        // 初始化规则
        // 初始化 zk config 监听器
        // 当有配置变更时
        // 调用 private void update(String json) 更新规则

    }

    public void close() {
        this.treeCache.close();
        this.zkClient.close();
    }

    private void update(String json) {

        Map<Long, DynamicProducerRule>
                result = getNewMap(json);

        // 1.将新增规则添加进规则池
        // 2.将下线规则从规则池删除
    }

    private Map<Long, DynamicProducerRule> getNewMap(String json) {
        // 将新规则解析，并使用 janino 进行初始化
    }
}

// ClientLogSource 是原始日志
boolean eval(flink.examples.datastream._01.bytedance.split.model.ClientLogSource clientLogSource) {
    return String.valueOf(clientLogSource.getId()).equals("1");
}

java:847 ms
janino:745 ms
groovy:4110 ms

java:1097 ms
janino:1170 ms
groovy:4052 ms

java:916 ms
janino:1117 ms
groovy:4311 ms

java:915 ms
janino:1112 ms
groovy:4382 ms

java:921 ms
janino:1104 ms
groovy:4321 ms

public class JaninoUtils {

    public static Class<Evaluable> genCodeAndGetClazz(Long id, String topic, String condition) throws Exception {
        // 动态生成代码
        // 初始化 Class<Evaluable> 并返回
    }

}

public class KafkaProducerCenter {

    // kafka producer 池
    private final ConcurrentMap<String, Producer<String, String>> producerConcurrentMap
            = new ConcurrentHashMap<>();

    private Producer<String, String> getProducer(String topicName) {

        // 如果 kafka producer 池中有当前 topic 的 producer，则直接返回
        // 如果没有，则初始化一个新的 producer 然后返回

    }

    public void send(String topicName, String message) {

        final ProducerRecord<String, String> record = new ProducerRecord<>(topicName,
                "", message);
        try {
            RecordMetadata metadata = getProducer(topicName).send(record).get();
        } catch (Exception e) {
            throw new RuntimeException(e);
        }
    }

    public void close() {
        // 关闭所有 producer 连接
    }
}