中文字幕精品亚洲无线码二区,国产黄a三级三级三级看三级,亚洲七七久久桃花影院,丰满少妇被猛烈进入,国产小视频在线观看网站

# MySQL 配(pei)置(zhi)文件
# Linux：my.cnf配(pei)置(zhi)文件（/etc/mysql/）
# Window：my.ini配置文件（C:\ProgramData\MySQL\MySQL Server 5.7\）
# 開啟 Binlog
log_bin = mysql-bin

# 選擇(ze) ROW 模式（記錄(lu)行級(ji)變更(geng)）
binlog-format = ROW

# 配置數(shu)據庫唯一 ID（與 Canal 服務端的 slaveId 不同(tong)）
server-id = 1

# 打開命令提示符(cmd/services.msc)：
# 按 Win + R 鍵，輸入 cmd，然后按 Enter 鍵打開命令提示符窗口。
# 停止MySQL服(fu)務(wu)：
net stop MySQL57

# 啟動(dong)MySQL服(fu)務(wu)：
net start MySQL57

# 驗證(zheng)
SHOW VARIABLES LIKE 'log_bin';
SHOW VARIABLES LIKE 'binlog_format';

        <!-- Flink 核心 -->
        <dependency>
            <groupId>org.apache.flink</groupId>
            <artifactId>flink-java</artifactId>
            <version>1.18.1</version>
        </dependency>
        <dependency>
            <groupId>org.apache.flink</groupId>
            <artifactId>flink-streaming-java</artifactId>
            <version>1.18.1</version>
        </dependency>
        <dependency>
            <groupId>org.apache.flink</groupId>
            <artifactId>flink-clients</artifactId>
            <version>1.18.1</version>
        </dependency>
        <!-- Flink Table 核心依賴 -->
        <dependency>
            <groupId>org.apache.flink</groupId>
            <artifactId>flink-table-api-java-bridge</artifactId>
            <version>1.18.1</version>
        </dependency>
        <dependency>
            <groupId>org.apache.flink</groupId>
            <artifactId>flink-table-planner_2.12</artifactId>
            <version>1.18.1</version>
        </dependency>
        <!-- Flink Connector Base -->
        <dependency>
            <groupId>org.apache.flink</groupId>
            <artifactId>flink-connector-base</artifactId>
            <version>1.18.1</version>
        </dependency>
        <!-- Flink CDC MySQL -->
        <dependency>
            <groupId>com.ververica</groupId>
            <artifactId>flink-connector-mysql-cdc</artifactId>
            <version>3.0.1</version>
        </dependency>

flink:
  cdc:
    # 是否開啟(qi)CDC監聽
    auto-start: true
    # 自定義(yi)一個唯一的id
    server-id: "123456"
    # 數(shu)據(ju)庫配置
    mysql:
      hostname: localhost
      port: 3306
      username: root
      password: 123

/**
 * Flink CDC數據變更類型枚舉
 * 1、"c"表(biao)示(shi)創建
 * 2、"u"表(biao)示(shi)更新
 * 3、"d"表(biao)示(shi)刪除(chu)
 * 4、"r"表(biao)示(shi)讀取
 */
public enum DataChangeType {

    INSERT("c"),
    UPDATE("u"),
    DELETE("d"),
    READ("r");

    private final String code;

    DataChangeType(String code) {
        this.code = code;
    }

    public static DataChangeType getByCode(String code) {
        for (DataChangeType type : values()) {
            if (type.code.equals(code)) {
                return type;
            }
        }
        return null;
    }

}

import lombok.Data;
import org.springframework.boot.context.properties.ConfigurationProperties;
import org.springframework.stereotype.Component;



@Data
@Component
@ConfigurationProperties(prefix = "flink.cdc")
public class FlinkCdcProperties {

    /**
     * 是否自動啟動CDC監聽
     */
    private boolean autoStart;

    private String serverId;

    /**
     * MySQL配(pei)置(zhi)
     */
    private Mysql mysql = new Mysql();

    @Data
    public static class Mysql {
        private String hostname;
        private int port;
        private String username;
        private String password;
    }

}

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.iven.flinkcdcdemoservice.entity.DataChangeType;
import com.iven.flinkcdcdemoservice.entity.FlinkCdcProperties;
import com.iven.flinkcdcdemoservice.handler.FlinkCdcHandlerRegistry;
import com.iven.flinkcdcdemoservice.handler.FlinkCdcHandler;
import com.ververica.cdc.connectors.mysql.source.MySqlSource;
import com.ververica.cdc.connectors.mysql.table.StartupOptions;
import com.ververica.cdc.debezium.JsonDebeziumDeserializationSchema;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.restartstrategy.RestartStrategies;
import org.apache.flink.api.common.time.Time;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.ProcessFunction;
import org.apache.flink.util.Collector;
import org.springframework.boot.CommandLineRunner;
import org.springframework.stereotype.Component;

import java.io.Serializable;
import java.util.*;

@Slf4j
@Component
@RequiredArgsConstructor
public class FlinkCdcRunner implements CommandLineRunner {

    // 配置屬性
    private final FlinkCdcProperties properties;

    // 處理器注冊中心
    private final FlinkCdcHandlerRegistry handlerRegistry;

    @Override
    public void run(String... args) throws Exception {
        // 總開關關閉則(ze)不啟(qi)動
        if (!properties.isAutoStart()) {
            log.info("Flink CDC 總開(kai)關關閉，不啟動監聽");
            return;
        }

        // 沒有需要監聽的表則不啟(qi)動
        List<String> monitoredTables = handlerRegistry.getMonitoredTables();
        if (monitoredTables.isEmpty()) {
            log.warn("未發現需要監聽的表（未實現FlinkCdcTableHandler），不啟動監聽");
            return;
        }

        // 1. 創建Flink執行環境(jing)
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        // 設(she)置(zhi)并(bing)行度(du)為 1: server-id 的數量必須(xu) ≥ 并(bing)行度(du)
        env.setParallelism(1);
        // 啟用(yong)檢查點（可選）
        // env.enableCheckpointing(5000);
        // 配置檢查點存儲路徑（本地路徑或分布式存儲如HDFS）
        // env.getCheckpointConfig().setCheckpointStorage("file:///tmp/flink-cdc-checkpoints");
        // 檢查點超(chao)時(shi)時(shi)間（60秒未完成則取消）
        // env.getCheckpointConfig().setCheckpointTimeout(60000);
        // 允許檢查(cha)點失敗次數（默(mo)認0，即一(yi)次失敗則任務(wu)失敗）
        // env.getCheckpointConfig().setTolerableCheckpointFailureNumber(3);
        // 禁用檢(jian)查(cha)點(dian)
        env.getCheckpointConfig().disableCheckpointing();
        // 重試(shi)次數/重試(shi)間(jian)隔(ge)
        env.setRestartStrategy(RestartStrategies.fixedDelayRestart(3,  Time.seconds(10)));

        // 2. 配置(zhi)(zhi)MySQL CDC源（動態設置(zhi)(zhi)需要監聽的表(biao)）
        MySqlSource<String> mySqlSource = MySqlSource.<String>builder()
                .serverId(properties.getServerId())
                .hostname(properties.getMysql().getHostname())
                .port(properties.getMysql().getPort())
                .username(properties.getMysql().getUsername())
                .password(properties.getMysql().getPassword())
                // 從監聽的表中提取數(shu)據庫列(lie)表（去重）
                .databaseList(extractDatabases(monitoredTables))
                // 直接使用注冊中(zhong)心收集(ji)的表列表
                .tableList(monitoredTables.toArray(new String[0]))
                // 反序列化為JSON
                .deserializer(new JsonDebeziumDeserializationSchema())
                /* initial: 初始化(hua)快照,即全(quan)量導(dao)入(ru)后增量導(dao)入(ru)(檢(jian)測更(geng)新數(shu)據(ju)(ju)寫入(ru))
                 * latest: 只進(jin)行增量導(dao)入(ru)(不讀(du)取歷史變化(hua))
                 * timestamp: 指定(ding)(ding)時間(jian)(jian)戳進(jin)行數(shu)據(ju)(ju)導(dao)入(ru)(大于等于指定(ding)(ding)時間(jian)(jian)錯讀(du)取數(shu)據(ju)(ju))
                 */
                .startupOptions(StartupOptions.latest())
                .build();

        // 3. 讀取CDC數據流并(bing)處理(li)
        DataStreamSource<String> dataStream = env.fromSource(
                mySqlSource,
                WatermarkStrategy.noWatermarks(),
                "MySQL-CDC-Source"
        );

        // 4. 解析數據并路由(you)到(dao)對應(ying)處理器，使用靜態內(nei)(nei)部類(lei)代(dai)替匿名內(nei)(nei)部類(lei)
        dataStream.process(new CdcDataProcessFunction(handlerRegistry));

        // 5. 啟動(dong)Flink作業
        env.execute("Flink-CDC-動態監聽作業");
    }

    /**
     * 靜態內部類實(shi)現ProcessFunction，確保可序列化
     */
    private static class CdcDataProcessFunction extends ProcessFunction<String, Void> implements Serializable {
        private final FlinkCdcHandlerRegistry handlerRegistry;

        // 通過構造(zao)函數傳入依賴
        public CdcDataProcessFunction(FlinkCdcHandlerRegistry handlerRegistry) {
            this.handlerRegistry = handlerRegistry;
        }

        @Override
        public void processElement(String json, Context ctx, Collector<Void> out) {
            try {
                JSONObject cdcData = JSON.parseObject(json);
                // 操作(zuo)類型：c/u/d
                String op = cdcData.getString("op");
                JSONObject source = cdcData.getJSONObject("source");
                String dbName = source.getString("db");
                String tableName = source.getString("table");
                // 庫名(ming).表名(ming)
                String fullTableName = dbName + "." + tableName;

                // 找到(dao)對應(ying)表的(de)處理器
                FlinkCdcHandler handler = handlerRegistry.getHandler(fullTableName);
                if (handler == null) {
                    log.warn("表[{}]無(wu)處理(li)器，跳過處理(li)", fullTableName);
                    return;
                }

                // 按事(shi)件(jian)類型分發(fa)
                DataChangeType changeType = DataChangeType.getByCode(op);
                if (changeType == null) {
                    log.warn("未知操作類型：{}", op);
                    return;
                }

                switch (changeType) {
                    case INSERT:
                        List<Map<String, Object>> insertData = Collections.singletonList(
                                cdcData.getJSONObject("after").getInnerMap()
                        );
                        handler.handleInsert(insertData);
                        break;
                    case UPDATE:
                        List<Map<String, Object>> beforeData = Collections.singletonList(
                                cdcData.getJSONObject("before").getInnerMap()
                        );
                        List<Map<String, Object>> afterData = Collections.singletonList(
                                cdcData.getJSONObject("after").getInnerMap()
                        );
                        handler.handleUpdate(beforeData, afterData);
                        break;
                    case DELETE:
                        List<Map<String, Object>> deleteData = Collections.singletonList(
                                cdcData.getJSONObject("before").getInnerMap()
                        );
                        handler.handleDelete(deleteData);
                        break;
                    case READ:
                        // 可以忽略快照階(jie)段的(de)讀取操作，或根(gen)據需要處(chu)理
                        log.debug("處理快(kuai)照讀取操作: {}", fullTableName);
                        break;
                }
            } catch (Exception e) {
                log.error("Flink-CDC數(shu)據處理發生未(wei)預期異常", e);
            }
        }
    }

    /**
     * 從表(biao)(biao)名（庫名.表(biao)(biao)名）中提取數據庫列表(biao)(biao)（去重）
     *
     * @param tables
     * @return
     */
    private String[] extractDatabases(List<String> tables) {
        // 截取庫名（如demo.tb_user → demo）
        return tables.stream()
                .map(table -> table.split("\\.")[0])
                .distinct()
                .toArray(String[]::new);
    }

}

import lombok.extern.slf4j.Slf4j;
import org.springframework.context.ApplicationContext;
import org.springframework.context.ApplicationContextAware;
import org.springframework.stereotype.Component;

import java.io.Serializable;
import java.util.ArrayList;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;

/**
 * Flink CDC處(chu)理(li)器注冊中(zhong)心
 * 處(chu)理(li)器注冊中(zhong)心（自動掃描監聽(ting)表(biao)）
 */
@Slf4j
@Component
public class FlinkCdcHandlerRegistry implements ApplicationContextAware, Serializable {

    // 緩存(cun)：表名（庫名.表名）→ 處理器
    private final Map<String, FlinkCdcHandler> handlerMap = new ConcurrentHashMap<>();

    // 收集(ji)所有需要監聽的表（供Flink CDC配置使用）
    private List<String> monitoredTables;

    @Override
    public void setApplicationContext(ApplicationContext applicationContext) {
        // 掃描所有實(shi)現類
        Map<String, FlinkCdcHandler> beans = applicationContext.getBeansOfType(FlinkCdcHandler.class);
        beans.values().forEach(handler -> {
            String tableName = handler.getTableName();
            handlerMap.put(tableName, handler);
            log.info("注(zhu)冊監聽表：{} → 處理器：{}", tableName, handler.getClass().getSimpleName());
        });
        // 提取所有需要監(jian)聽的表(biao)
        monitoredTables = new ArrayList<>(handlerMap.keySet());
    }

    /**
     * 獲(huo)取指定(ding)表的處理(li)器
     *
     * @param tableName
     * @return
     */
    public FlinkCdcHandler getHandler(String tableName) {
        return handlerMap.get(tableName);
    }

    /**
     * 獲(huo)取所(suo)有需要監聽的表（供Flink CDC配置(zhi)）
     *
     * @return
     */
    public List<String> getMonitoredTables() {
        return monitoredTables;
    }

}

import java.util.List;
import java.util.Map;

/**
 * 表數據處理接口，每個監聽(ting)的表需實現此接口
 */
public interface FlinkCdcHandler {

    /**
     * 獲取監聽的表名(ming)(ming)（格(ge)式：庫名(ming)(ming).表名(ming)(ming)，如demo.tb_user）
     */
    String getTableName();

    /**
     * 處理新增數據
     */
    default void handleInsert(List<Map<String, Object>> dataList) {
        // 默認空(kong)實現，子(zi)類(lei)可(ke)重寫
    }

    /**
     * 處理更新(xin)數據(ju)（包含變(bian)更前(qian)和(he)變(bian)更后的數據(ju)）
     * @param beforeList 變(bian)更前(qian)數據(ju)
     * @param afterList 變(bian)更后數據(ju)
     */
    default void handleUpdate(List<Map<String, Object>> beforeList, List<Map<String, Object>> afterList) {
        // 默(mo)認空實現，子類可(ke)重寫(xie)
    }

    /**
     * 處理刪除數據(ju)
     */
    default void handleDelete(List<Map<String, Object>> dataList) {
        // 默(mo)認空實現，子類可重寫(xie)
    }

}

import lombok.extern.slf4j.Slf4j;
import org.springframework.stereotype.Component;

import java.io.Serializable;
import java.util.List;
import java.util.Map;

@Slf4j
@Component
public class TbUserFlinkCdcHandler implements FlinkCdcHandler, Serializable {
    @Override
    public String getTableName() {
        return "demo.tb_user";
    }

    @Override
    public void handleInsert(List<Map<String, Object>> dataList) {
        log.info("處(chu)理(li)tb_user新(xin)增數(shu)據，共{}條", dataList.size());
        dataList.forEach(data -> {
            String id = (String)data.get("id");
            String username = (String) data.get("name");
            // 業務邏輯：如同步到ES、緩存等
            log.info("新增(zeng)用戶：id={}, name={}", id, username);
        });
    }

    @Override
    public void handleUpdate(List<Map<String, Object>> beforeList, List<Map<String, Object>> afterList) {
        log.info("處理tb_user更(geng)新數據，共(gong){}條", afterList.size());
        for (int i = 0; i < afterList.size(); i++) {
            Map<String, Object> before = beforeList.get(i);
            Map<String, Object> after = afterList.get(i);
            log.info("更新用戶(hu)：id={}, 舊用戶(hu)名={}, 新用戶(hu)名={}",
                    after.get("id"), before.get("name"), after.get("name"));
        }
    }

    @Override
    public void handleDelete(List<Map<String, Object>> dataList) {
        log.info("處(chu)理(li)tb_user刪除數據，共{}條", dataList.size());
        dataList.forEach(data -> {
            log.info("刪除用戶：id={}", data.get("id"));
        });
    }

}