Postgres/jdbc/逻辑复制 - 出于内存问题

Question

我正在开发一个连接到逻辑复制插槽的应用程序，以消耗WAL事件。然后将这些WAL活动转发给MQ经纪人。这很棒，但是我注意到一段时间后我的记忆力都用完了。我设法将问题最小化，负责获得WAL事件的守则。它在以下代码中发生：

final Properties properties = new Properties();
PGProperty.USER.set(properties, "user");
PGProperty.PASSWORD.set(properties, "password");
PGProperty.ASSUME_MIN_SERVER_VERSION.set(properties, "9.4");
PGProperty.REPLICATION.set(properties, "database");
PGProperty.PREFER_QUERY_MODE.set(properties, "simple");
while (true) {
    Connection          connection   = null;
    PGConnection        PGConnection = null;
    PGReplicationStream stream       = null;
    try {
        connection = DriverManager.getConnection("jdbc:postgresql://localhost:5432/db", properties);
        PGConnection = connection.unwrap(PGConnection.class);
        stream = PGConnection.getReplicationAPI().replicationStream().logical().withSlotName("slot").start();
        while (true) {
            final ByteBuffer buffer = stream.read();
            // ... logic here ... (disabled during memory test)
            stream.setAppliedLSN(stream.getLastReceiveLSN());
            stream.setFlushedLSN(stream.getLastReceiveLSN());
        }
    } catch (final SQLException e1) {
        Logger.getLogger(getClass()).error(e1);
        if (stream != null) {
            try {
                stream.close();
            } catch (final SQLException e2) {
                Logger.getLogger(getClass()).error(e2);
            }
        }
        if (connection != null) {
            try {
                connection.close();
            } catch (final SQLException e2) {
                Logger.getLogger(getClass()).error(e2);
            }
        }
    }
}

我评论了解析消息并将其转发给MQ经纪人的逻辑，因为在没有此事的情况下也发生了内存。

我还尝试使用轮询方法readPending()而不是阻止方法read()（如https://jdbc.postgresql.org/documentation/head/replication.html），但问题仍然存在。

我还注意到，一段时间后，该应用程序以100％的CPU使用。这必须是由基础库引起的，因为read()当时仍在正常处理（也就是说，它依次处理每个WAL事件）。

在这些测试期间，我要执行INSERT和UPDATE查询，以较低的速率。

我正在使用以下依赖性：

<dependency>
    <groupId>org.postgresql</groupId>
    <artifactId>postgresql</artifactId>
    <version>42.1.4</version>
</dependency>

该应用程序在tomcat8容器中作为战争运行。

任何想法发生了什么事？

更新1

我弄清楚了发生了什么，但无法解释到目前为止。我会详细介绍。

每10秒钟，我进行INSERT和UPDATE查询，如所述。这些查询导致645个WAL事件。因此，每10秒钟，我必须进行read() 645事件。一开始，这需要0（或有时为1）毫秒到read()一个事件。一段时间后，它需要1毫秒。然后，一段时间后，它需要2毫秒。等等...

因此，一段时间后，我无法在10秒内read() 645事件，因为read()所需的时间不断增加。这解释了100％的CPU用法和不记忆。

我仍然不确定如何解释以及如何解决此问题。我将继续调查。

更新2

我尝试在循环末端添加buffer.clear()，但没有成功。我仍遇到100％CPU和内存问题。这是预期的，因为缓冲区是局部变量，因此每次循环后都会gced。但是我认为无论如何测试都是一个好主意。

Answer 1

我找到了我的原因。我正在使用decoderbufs解码输出插件测试，https://github.com/xstevens/decoderbufs。用内置的test插件或wal2json（https://github.com/eulerto/wal2json）替换时，我没有这些问题。

我将尝试通知作者decoderbufs插件。