EC0823/FLINKTAST-DEMO-01 at master · stevensu1/EC0823
EC0823/FLINKTASK-DEMO-02 at master · stevensu1/EC0823

流式处理（自定义支持多数库的连接池）：

DataSourceRouter：数据源路由器，提供智能的数据源选择策略，确保连接类型安全
DataSourceType：数据源类型枚举，定义系统支持的各种数据源类型
MultiDataSourceConfig：多数据源配置管理类，支持多种数据库的连接配置管理MultiDataSourceConnectionManager：多数据源连接管理器，支持管理多种数据库的连接池

具体代码见上面的仓库连接，下是使使用方法：

/**
 * 这使用一个flink执行环境，用于在一个执行环境中加载多个flink任务（使用不同的数据源
 *         LocalFlinkJob1.task(env);
 *         LocalFlinkJob2.task(env);
 *         ）
 */

package org.example.demo01;

import org.apache.flink.streaming.api.CheckpointingMode;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;


/**
 * 这使用一个flink执行环境，用于在一个执行环境中加载多个flink任务（使用不同的数据源
 *         LocalFlinkJob1.task(env);
 *         LocalFlinkJob2.task(env);
 *         ）
 */
public class LocalFlinkJob {

    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(8);

        // ✅ 开启 Checkpointing，支持 Exactly-Once
        env.enableCheckpointing(5000); // 每 5 秒做一次 checkpoint
        env.getCheckpointConfig().setCheckpointingMode(CheckpointingMode.EXACTLY_ONCE);
        env.getCheckpointConfig().setMinPauseBetweenCheckpoints(2000); // 避免频繁 checkpoint

        // 不再在主类中初始化 dbManager，避免序列化问题
        LocalFlinkJob1.task(env);
        LocalFlinkJob2.task(env);

        // ✅ 启动执行
        env.execute("Production JDBC Write Job");
    }

}

task 任务子类：

LocalFlinkJob1

package org.example.demo01;

import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.source.SourceFunction;
import org.example.demo01.sinkBD.MySQLSinkFunction;
import org.example.demo01.sourceDB.DataSourceType;
import org.example.demo01.sourceDB.MultiDataSourceConnectionManager;

import java.sql.Connection;
import java.sql.PreparedStatement;
import java.sql.ResultSet;

public class LocalFlinkJob1 {

    public static void task(StreamExecutionEnvironment env) throws Exception {
        DataStreamSource<MyRecordOne> dataStreamSource = env.addSource(
                new SourceFunction<MyRecordOne>() {
                    private volatile boolean isRunning = true;
                    private transient MultiDataSourceConnectionManager dbManager; // 标记为 transient

                    @Override
                    public void run(SourceContext<MyRecordOne> ctx) throws Exception {
                        // 在运行时初始化，避免序列化问题
                        dbManager = MultiDataSourceConnectionManager.getInstance();

                        while (isRunning) {
                            Connection connection = null;
                            try {
                                // 明确指定使用主MySQL数据源，确保类型安全
                                connection = dbManager.getConnection(DataSourceType.MYSQL_PRIMARY);

                                PreparedStatement statement = connection.prepareStatement(
                                        "SELECT id, name, value FROM my_record"
                                );
                                ResultSet resultSet = statement.executeQuery();

                                while (resultSet.next() && isRunning) {
                                    MyRecordOne record = new MyRecordOne(
                                            resultSet.getInt("id"),
                                            resultSet.getString("name"),
                                            resultSet.getString("value")
                                    );
                                    ctx.collect(record);
                                }

                                resultSet.close();
                                statement.close();

                            } catch (Exception e) {
                                System.err.println("数据库查询异常: " + e.getMessage());
                                throw new RuntimeException(e);
                            } finally {
                                if (connection != null) {
                                    connection.close(); // HikariCP会自动归还连接到池中
                                }
                            }

                            // 等待 5 秒再次查询
                            //  TimeUnit.SECONDS.sleep(5);
                        }
                    }

                    @Override
                    public void cancel() {
                        isRunning = false;
                    }
                }
                , TypeInformation.of(MyRecordOne.class)
        );

        // ✅ 打印验证（可选）
        dataStreamSource.print();
        dataStreamSource.addSink(new MySQLSinkFunction<>(MyRecordOne.class)).setParallelism(8);

    }


}

LocalFlinkJob2

package org.example.demo01;

import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.source.SourceFunction;
import org.example.demo01.sinkBD.MySQLSinkFunction;
import org.example.demo01.sourceDB.DataSourceType;
import org.example.demo01.sourceDB.MultiDataSourceConnectionManager;

import java.sql.Connection;
import java.sql.PreparedStatement;
import java.sql.ResultSet;
import java.util.concurrent.TimeUnit;

public class LocalFlinkJob2 {


    public static void task(StreamExecutionEnvironment env) throws Exception {
        DataStreamSource<MyRecordOne> dataStreamSource = env.addSource(
                new SourceFunction<MyRecordOne>() {
                    private volatile boolean isRunning = true;
                    private transient MultiDataSourceConnectionManager dbManager; // 标记为 transient

                    @Override
                    public void run(SourceContext<MyRecordOne> ctx) throws Exception {
                        // 在运行时初始化，避免序列化问题
                        dbManager = MultiDataSourceConnectionManager.getInstance();

                        while (isRunning) {
                            Connection connection = null;
                            try {
                                // 明确指定使用主MySQL数据源，确保类型安全
                                connection = dbManager.getConnection(DataSourceType.POSTGRESQL);

                                PreparedStatement statement = connection.prepareStatement(
                                        "SELECT id, name, value FROM my_record"
                                );
                                ResultSet resultSet = statement.executeQuery();

                                while (resultSet.next() && isRunning) {
                                    MyRecordOne record = new MyRecordOne(
                                            resultSet.getInt("id"),
                                            resultSet.getString("name"),
                                            resultSet.getString("value")
                                    );
                                    ctx.collect(record);
                                }

                                resultSet.close();
                                statement.close();

                            } catch (Exception e) {
                                System.err.println("数据库查询异常: " + e.getMessage());
                                throw new RuntimeException(e);
                            } finally {
                                if (connection != null) {
                                    connection.close(); // HikariCP会自动归还连接到池中
                                }
                            }

                            // 等待 5 秒再次查询
                            TimeUnit.SECONDS.sleep(5);
                        }
                    }

                    @Override
                    public void cancel() {
                        isRunning = false;
                    }
                }
                , TypeInformation.of(MyRecordOne.class)
        );

        // ✅ 打印验证（可选）
        dataStreamSource.print();
        dataStreamSource.addSink(new MySQLSinkFunction<>(MyRecordOne.class)).setParallelism(8);

    }

}

执行前需注意 加入jvm 参数：

--add-opens java.base/java.lang=ALL-UNNAMED
--add-opens java.base/java.util=ALL-UNNAMED
--add-opens java.base/java.lang.invoke=ALL-UNNAMED

下面是同时从两个库加载到的数据：

批次处理(使用mysql-flax 预加载数据)

整合mybatis-flax过程这里就略过了，主要注意将从数据获取的list<entity>加载到flink 环境：

执行结果：