flink cdc采集的时候，写mybatis查询，cancel job后，为何数据库连接没有释放？

在您使用Flink CDC进行数据采集时，同时自定义了MyBatis查询来获取配置数据，并遇到取消作业（cancel job）后MyBatis的数据库连接未被释放的问题。这可能是由于以下几个原因导致的：

1. 资源管理不当：Flink作业在取消时，其内部的资源通常会进行清理，但自定义的MyBatis查询部分可能没有正确集成到Flink的资源生命周期管理中。这意味着当Flink作业结束时，MyBatis创建的数据库连接可能没有被显式关闭。

2. 连接池配置问题：如果您的MyBatis配置使用了连接池（如上述示例中的POOLED类型），连接池可能没有配置合适的连接回收策略。即使作业结束，连接池可能仍然持有这些连接，直到达到预设的超时时间才释放。

3. 异常处理缺失：在执行MyBatis查询的过程中，如果遇到未被捕获的异常，可能导致连接没有通过finally块或try-with-resources语句正确关闭。

解决方案

为了确保MyBatis的数据库连接在Flink作业取消后能够被正确释放，您可以采取以下措施：

• 集成资源释放逻辑：在Flink作业的生命周期管理中明确地添加代码来管理MyBatis的数据库连接。例如，在open()方法中初始化数据库连接，在close()方法中确保所有资源被正确关闭。对于DataStream API，这对应于RichFunction的生命周期方法；对于Table API，则可能需要自定义函数并实现相应的生命周期接口。

• 优化连接池配置：

  • 确保连接池配置包含合理的最大空闲时间和超时时间，以便在无活动连接时及时回收。
  • 使用dataSource配置中的closeConnection相关参数，确保连接在不再需要时能被及时关闭。

• 增强异常处理：确保所有数据库操作都包裹在适当的异常处理逻辑中，确保即使发生异常也能释放资源。可以使用try-with-resources语句来自动管理资源生命周期。

示例代码片段

在Flink作业中使用自定义函数时，确保资源管理得当，如下所示（伪代码示例）：

public class MyDatabaseLookupFunction extends RichFlatMapFunction<...> {
    private DataSource dataSource;
    private MyMapper myMapper;

    @Override
    public void open(Configuration parameters) throws Exception {
        super.open(parameters);
        // 初始化数据源和MyBatis Mapper
        dataSource = ...; // 根据实际情况初始化数据源
        SqlSessionFactory sqlSessionFactory = new SqlSessionFactoryBuilder().build(Resources.getResourceAsStream("mybatis-config.xml"));
        SqlSession sqlSession = sqlSessionFactory.openSession(dataSource);
        myMapper = sqlSession.getMapper(MyMapper.class);
    }

    @Override
    public void flatMap(..., Collector<...> out) throws Exception {
        // 执行MyBatis查询逻辑
        ...
    }

    @Override
    public void close() throws Exception {
        if (myMapper != null && myMapper.sqlSession != null) {
            myMapper.sqlSession.close();
        }
        if (dataSource != null) {
            // 根据数据源类型，执行适当的关闭逻辑
            ...
        }
        super.close();
    }
}

注意事项

• 资源泄露检测：在开发和测试阶段，考虑使用工具或日志记录来监控和检测潜在的资源泄露问题。
• 连接池监控：定期检查连接池的使用情况，确保配置合理且运行高效。

通过上述措施，您可以有效解决Flink CDC作业取消后MyBatis数据库连接未释放的问题。