flink 代码中 定时任务查询mysql 数据库 如何关闭 ResultSet 防止内存泄露

在 Flink 中，可以使用 `RichFlatMapFunction` 或 `RichMapFunction` 来执行定时任务查询 MySQL 数据库。为了防止内存泄漏，需要在适当的时候关闭 ResultSet。

以下是一个示例代码，展示了如何在 Flink 中执行定时任务查询 MySQL 数据库并正确关闭 ResultSet：

import java.sql.*;

public class MySQLQueryFunction extends RichFlatMapFunction<String, String> {
    private transient Connection connection;
    private transient PreparedStatement statement;

    @Override
    public void open(Configuration parameters) throws Exception {
        super.open(parameters);
        connection = DriverManager.getConnection("jdbc:mysql://localhost:3306/db", "user", "password");
        statement = connection.prepareStatement("SELECT * FROM table WHERE condition");
    }

    @Override
    public void flatMap(String value, Collector<String> out) throws Exception {
        ResultSet resultSet = statement.executeQuery();
        try {
            while (resultSet.next()) {
                // 处理结果集
                String result = resultSet.getString("column");
                out.collect(result);
            }
        } finally {
            if (resultSet != null) {
                resultSet.close();
            }
        }
    }

    @Override
    public void close() throws Exception {
        super.close();
        if (statement != null) {
            statement.close();
        }
        if (connection != null) {
            connection.close();
        }
    }
}

在上述代码中，`open()` 方法中创建了数据库连接并准备了 PreparedStatement 对象。在 `flatMap()` 方法中执行查询，并在处理完结果集后通过 `finally` 块关闭 ResultSet。在 `close()` 方法中，通过 `try-with-resources` 语句块自动关闭 Statement 和 Connection 对象。

通过以上的代码实现，可以确保 ResultSet 在任务执行结束后被正确关闭，避免内存泄漏问题。