Flink使用richsinkfunction，写入PG，每隔半小时就报错和PG断开连接该怎么解决？-[阿里云_云淘科技]

匿名 • 2024年1月9日下午12:37 • 阿里云

Flink使用richsinkfunction，写入PG，每隔半小时就报错和PG断开连接（数据发送间隔时间很长，导致长时间会没有数据），该怎么解决啊？我用了druid连接池，配置了一些参数，但好像也没起作用？在invoke里也加了判断，如果connection断开了，就重新连接，但也没起作用

以下为热心网友提供的参考意见

这个问题可能是由于PG连接池的配置问题或者Flink的RichSinkFunction处理逻辑导致的。你可以尝试以下方法来解决这个问题：

检查PG连接池的配置，确保配置正确。例如，检查连接池的最大连接数、最小连接数、空闲连接超时时间等参数。
在RichSinkFunction中，尝试使用try-catch语句捕获异常，并在异常发生时重新建立连接。例如：

public class MyRichSinkFunction extends RichSinkFunction<MyEvent> {
    private Connection connection;

    @Override
    public void open(Configuration parameters) throws Exception {
        super.open(parameters);
        // 初始化连接
        connection = createConnection();
    }

    @Override
    public void close() throws Exception {
        super.close();
        // 关闭连接
        if (connection != null) {
            connection.close();
        }
    }

    @Override
    public void invoke(MyEvent value, Context context) {
        try {
            // 写入数据的逻辑
        } catch (Exception e) {
            // 发生异常时重新建立连接
            connection = createConnection();
        }
    }

    private Connection createConnection() {
        // 创建并返回一个新的数据库连接
    }
}