@SpringBootApplication
public class MyApp {
  public static void main(String[] args) {
    SpringApplication.run(MyApp.class, args);
  }
  @Bean
  public CommandLineRunner commandLineRunner(ApplicationContext ctx) {
    return args -> {
      StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
      DebeziumSourceFunction<String> sourceFunction = MySqlSource.<String>builder()
              .hostname("localhost")
              .port(3306)
              .username("flinkuser")
              .password("flinkpw")
              .databaseList("mydb") // monitor all tables under "mydb" database
              .tableList("mydb.table1", "mydb.table2") // monitor only "table1" and "table2" under "mydb" database
              .deserializer(new StringDebeziumDeserializationSchema()) // converts SourceRecord to String
              .build();
      DataStreamSource<String> mysqlSource = env.addSource(sourceFunction);
      // formulate processing logic here, e.g., printing to standard output
      mysqlSource.print();
      // execute the Flink job within the Spring Boot application
      env.execute("Flink CDC");
    };
  }
}

@SpringBootApplication
public class FlinkCDCApplication implements ApplicationRunner {
    public static void main(String[] args) {
        SpringApplication.run(FlinkCDCApplication.class, args);
    }
    @Override
    public void run(ApplicationArguments args) throws Exception {
        final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        // Configure your Flink job here
        DebeziumSourceFunction<String> sourceFunction = MySqlSource.<String>builder()
                .hostname("localhost")
                .port(3306)
                .username("flinkuser")
                .password("flinkpw")
                .databaseList("mydb")
                // set other source options ...
                .deserializer(new StringDebeziumDeserializationSchema()) // Converts SourceRecord to String
                .build();
        DataStream<String> cdcStream = env.addSource(sourceFunction);
        // Implement your processing logic here
        // For example:
        cdcStream.print();
        // Start the Flink job within the Spring Boot application
        env.execute("Flink CDC with Spring Boot");
    }
}

<flink.version>1.13.6</flink.version>
<dependency>
   <groupId>org.apache.flink</groupId>
   <artifactId>flink-clients_2.12</artifactId>
   <version>${flink.version}</version>
</dependency>
   <dependency>
   <groupId>org.apache.flink</groupId>
   <artifactId>flink-java</artifactId>
   <version>${flink.version}</version>
</dependency>
<dependency>
   <groupId>org.apache.flink</groupId>
   <artifactId>flink-streaming-java_2.12</artifactId>
   <version>${flink.version}</version>
</dependency>
<!--mysql -cdc-->
<dependency>
   <groupId>com.ververica</groupId>
   <artifactId>flink-connector-mysql-cdc</artifactId>
   <version>2.0.0</version>
</dependency>
<dependency>
   <groupId>org.projectlombok</groupId>
   <artifactId>lombok</artifactId>
   <version>1.18.10</version>
</dependency>
<dependency>
   <groupId>cn.hutool</groupId>
   <artifactId>hutool-all</artifactId>
   <version>5.8.5</version>
</dependency>
<dependency>
    <groupId>org.apache.commons</groupId>
    <artifactId>commons-lang3</artifactId>
    <version>3.10</version>
</dependency>
<dependency>
   <groupId>com.alibaba</groupId>
   <artifactId>fastjson</artifactId>
   <version>2.0.42</version>
</dependency>

@Component
public class MysqlEventListener implements ApplicationRunner {
    private final DataChangeSink dataChangeSink;
    public MysqlEventListener(DataChangeSink dataChangeSink) {
        this.dataChangeSink = dataChangeSink;
    }
    @Override
    public void run(ApplicationArguments args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        DebeziumSourceFunction<DataChangeInfo> dataChangeInfoMySqlSource = buildDataChangeSourceRemote();
        DataStream<DataChangeInfo> streamSource = env
                .addSource(dataChangeInfoMySqlSource, "mysql-source")
                .setParallelism(1);
        streamSource.addSink(dataChangeSink);
        env.execute("mysql-stream-cdc");
    }
    private DebeziumSourceFunction<DataChangeInfo> buildDataChangeSourceLocal() {
        return MySqlSource.<DataChangeInfo>builder()
                .hostname("127.0.0.1")
                .port(3306)
                .username("root")
                .password("0507")
                .databaseList("flink-cdc-producer")
                .tableList("flink-cdc-producer.producer_content", "flink-cdc-producer.name_content")
                /*
                 * initial初始化快照,即全量導(dǎo)入后增量導(dǎo)入(檢測更新數(shù)據(jù)寫入)
                 * latest:只進(jìn)行增量導(dǎo)入(不讀取歷史變化)
                 * timestamp:指定時間戳進(jìn)行數(shù)據(jù)導(dǎo)入(大于等于指定時間錯讀取數(shù)據(jù))
                 */
                .startupOptions(StartupOptions.latest())
                .deserializer(new MysqlDeserialization())
                .serverTimeZone("GMT+8")
                .build();
    }
}

public class MysqlDeserialization implements DebeziumDeserializationSchema<DataChangeInfo> {
    public static final String TS_MS = "ts_ms";
    public static final String BIN_FILE = "file";
    public static final String POS = "pos";
    public static final String CREATE = "CREATE";
    public static final String BEFORE = "before";
    public static final String AFTER = "after";
    public static final String SOURCE = "source";
    public static final String UPDATE = "UPDATE";
    /**
     * 反序列化數(shù)據(jù),轉(zhuǎn)為變更JSON對象
     */
    @Override
    public void deserialize(SourceRecord sourceRecord, Collector<DataChangeInfo> collector) {
        String topic = sourceRecord.topic();
        String[] fields = topic.split("\\.");
        String database = fields[1];
        String tableName = fields[2];
        Struct struct = (Struct) sourceRecord.value();
        final Struct source = struct.getStruct(SOURCE);
        DataChangeInfo dataChangeInfo = new DataChangeInfo();
        dataChangeInfo.setBeforeData(getJsonObject(struct, BEFORE).toJSONString());
        dataChangeInfo.setAfterData(getJsonObject(struct, AFTER).toJSONString());
        //5.獲取操作類型  CREATE UPDATE DELETE
        Envelope.Operation operation = Envelope.operationFor(sourceRecord);
//        String type = operation.toString().toUpperCase();
//        int eventType = type.equals(CREATE) ? 1 : UPDATE.equals(type) ? 2 : 3;
        dataChangeInfo.setEventType(operation.name());
        dataChangeInfo.setFileName(Optional.ofNullable(source.get(BIN_FILE)).map(Object::toString).orElse(""));
        dataChangeInfo.setFilePos(Optional.ofNullable(source.get(POS)).map(x -> Integer.parseInt(x.toString())).orElse(0));
        dataChangeInfo.setDatabase(database);
        dataChangeInfo.setTableName(tableName);
        dataChangeInfo.setChangeTime(Optional.ofNullable(struct.get(TS_MS)).map(x -> Long.parseLong(x.toString())).orElseGet(System::currentTimeMillis));
        //7.輸出數(shù)據(jù)
        collector.collect(dataChangeInfo);
    }
    private Struct getStruct(Struct value, String fieldElement) {
        return value.getStruct(fieldElement);
    }
    /**
     * 從元數(shù)據(jù)獲取出變更之前或之后的數(shù)據(jù)
     */
    private JSONObject getJsonObject(Struct value, String fieldElement) {
        Struct element = value.getStruct(fieldElement);
        JSONObject jsonObject = new JSONObject();
        if (element != null) {
            Schema afterSchema = element.schema();
            List<Field> fieldList = afterSchema.fields();
            for (Field field : fieldList) {
                Object afterValue = element.get(field);
                jsonObject.put(field.name(), afterValue);
            }
        }
        return jsonObject;
    }
    @Override
    public TypeInformation<DataChangeInfo> getProducedType() {
        return TypeInformation.of(DataChangeInfo.class);
    }
}

@Data
public class DataChangeInfo implements Serializable {
    /**
     * 變更前數(shù)據(jù)
     */
    private String beforeData;
    /**
     * 變更后數(shù)據(jù)
     */
    private String afterData;
    /**
     * 變更類型 1新增 2修改 3刪除
     */
    private String eventType;
    /**
     * binlog文件名
     */
    private String fileName;
    /**
     * binlog當(dāng)前讀取點(diǎn)位
     */
    private Integer filePos;
    /**
     * 數(shù)據(jù)庫名
     */
    private String database;
    /**
     * 表名
     */
    private String tableName;
    /**
     * 變更時間
     */
    private Long changeTime;
}

@Component
@Slf4j
public class DataChangeSink extends RichSinkFunction<DataChangeInfo> {
    transient RabbitTemplate rabbitTemplate;
    transient ConfirmService confirmService;
    transient TableDataConvertService tableDataConvertService;
    @Override
    public void invoke(DataChangeInfo value, Context context) {
        log.info("收到變更原始數(shù)據(jù):{}", value);
        //轉(zhuǎn)換后發(fā)送到對應(yīng)的MQ
        if (MIGRATION_TABLE_CACHE.containsKey(value.getTableName())) {
            String routingKey = MIGRATION_TABLE_CACHE.get(value.getTableName());
            //可根據(jù)需要自行進(jìn)行confirmService的設(shè)計(jì)
            rabbitTemplate.setReturnsCallback(confirmService);
            rabbitTemplate.setConfirmCallback(confirmService);
            rabbitTemplate.convertAndSend(EXCHANGE_NAME, routingKey, tableDataConvertService.convertSqlByDataChangeInfo(value));
        }
    }
    /**
     * 在啟動SpringBoot項(xiàng)目是加載了Spring容器，其他地方可以使用@Autowired獲取Spring容器中的類；但是Flink啟動的項(xiàng)目中，
     * 默認(rèn)啟動了多線程執(zhí)行相關(guān)代碼，導(dǎo)致在其他線程無法獲取Spring容器，只有在Spring所在的線程才能使用@Autowired，
     * 故在Flink自定義的Sink的open()方法中初始化Spring容器
     */
    @Override
    public void open(Configuration parameters) throws Exception {
        super.open(parameters);
        this.rabbitTemplate = ApplicationContextUtil.getBean(RabbitTemplate.class);
        this.confirmService = ApplicationContextUtil.getBean(ConfirmService.class);
        this.tableDataConvertService = ApplicationContextUtil.getBean(TableDataConvertService.class);
    }
}

public interface TableDataConvertService {
    String convertSqlByDataChangeInfo(DataChangeInfo dataChangeInfo);
}

@Service
public class TableDataConvertServiceImpl implements TableDataConvertService {
    @Autowired
    Map<String, SqlGeneratorService> sqlGeneratorServiceMap;
    @Override
    public String convertSqlByDataChangeInfo(DataChangeInfo dataChangeInfo) {
        SqlGeneratorService sqlGeneratorService = sqlGeneratorServiceMap.get(dataChangeInfo.getEventType());
        return sqlGeneratorService.generatorSql(dataChangeInfo);
    }
}

public interface SqlGeneratorService {
    String generatorSql(DataChangeInfo dataChangeInfo);
}
public abstract class AbstractSqlGenerator implements SqlGeneratorService {
    @Override
    public String generatorSql(DataChangeInfo dataChangeInfo) {
        return null;
    }
     public String quoteIdentifier(String identifier) {
        // 對字段名進(jìn)行轉(zhuǎn)義處理，這里簡化為對其加反引號
        // 實(shí)際應(yīng)該處理數(shù)據(jù)庫標(biāo)識符的特殊字符
        return "`" + identifier + "`";
    }
}

@Service("DELETE")
@Slf4j
public class DeleteSqlGeneratorServiceImpl extends AbstractSqlGenerator {
    @Override
    public String generatorSql(DataChangeInfo dataChangeInfo) {
        String beforeData = dataChangeInfo.getBeforeData();
        Map<String, Object> beforeDataMap = JSONObjectUtils.JsonToMap(beforeData);
        StringBuilder wherePart = new StringBuilder();
        for (String key : beforeDataMap.keySet()) {
            Object beforeValue = beforeDataMap.get(key);
            if ("create_time".equals(key)){
                SimpleDateFormat dateFormat = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
                beforeValue = dateFormat.format(beforeValue);
            }
            if (wherePart.length() > 0) {
                // 不是第一個更改的字段，增加逗號分隔
                wherePart.append(", ");
            }
            wherePart.append(quoteIdentifier(key)).append(" = ").append(formatValue(beforeValue));
        }
        log.info("wherePart : {}", wherePart);
        return "DELETE FROM " + dataChangeInfo.getTableName() + " WHERE " + wherePart;
    }
}