为什么 Postgres 复制流在单独的函数中使用时不起作用？

Question

我正在研究postgres复制流API。在处理它时遇到了异常行为。

当我使用复制槽在主块内编写整个代码时，一切正常。

public class Server implements Config {
public static void main(String[] args) {
Properties prop = new Properties();
prop.load(new FileInputStream(System.getProperty("prop")));
String user = prop.getProperty("user");
String password = prop.getProperty("password");
String url = prop.getProperty("url");
Properties props = new Properties();
PGProperty.USER.set(props, user);
PGProperty.PASSWORD.set(props, password);
PGProperty.ASSUME_MIN_SERVER_VERSION.set(props, "9.4");
PGProperty.REPLICATION.set(props, "database");
PGProperty.PREFER_QUERY_MODE.set(props, "simple");
Connection conn= null;
PGConnection replicationConnection= null;
PGReplicationStream stream = null;
conn = DriverManager.getConnection(url, props);
replicationConnection = conn.unwrap(PGConnection.class);
stream = replicationConnection.getReplicationAPI().replicationStream().logical()
.withSlotName("replication_slot")
.withSlotOption("include-xids", true)
.withSlotOption("include-timestamp", "on")
.withSlotOption("skip-empty-xacts", true)
.withStatusInterval(20, TimeUnit.SECONDS).start();
while (true) {
ByteBuffer msg;
try {
msg = stream.readPending();         
if (msg == null) {
TimeUnit.MILLISECONDS.sleep(10L);
continue;
}
int offset = msg.arrayOffset();
byte[] source = msg.array();
int length = source.length - offset;
// convert byte buffer into string
String data = new String(source, offset, length);
// then convert it into bufferedreader
BufferedReader reader = new BufferedReader(new StringReader(data));
String line = reader.readLine();
while (line != null) {
System.out.println(line);
line = reader.readLine();
}
stream.setAppliedLSN(stream.getLastReceiveLSN());
stream.setFlushedLSN(stream.getLastReceiveLSN());
} catch (SQLException | IOException | InterruptedException e) {
// TODO Auto-generated catch block
e.printStackTrace();
}
}
}

但是当我尝试使用这样的单独方法分离流逻辑时

public class Server implements Config {
public static void main(String[] args) {
PGReplicationStream stream = getReplicationStream();
while (true) {
ByteBuffer msg;
try {
msg = stream.readPending();
if (msg == null) {
TimeUnit.MILLISECONDS.sleep(10L);
continue;
}
int offset = msg.arrayOffset();
byte[] source = msg.array();
int length = source.length - offset;
String data = new String(source, offset, length);
BufferedReader reader = new BufferedReader(new StringReader(data));
String line = reader.readLine();
while (line != null) {
System.out.println(line);
line = reader.readLine();
}
stream.setAppliedLSN(stream.getLastReceiveLSN());
stream.setFlushedLSN(stream.getLastReceiveLSN());
} catch (SQLException | IOException | InterruptedException e) {
e.printStackTrace();
}
}
}
public static PGReplicationStream getReplicationStream() {
Properties prop = new Properties();
try {
prop.load(new FileInputStream(System.getProperty("prop")));
} catch (IOException e1) {
// TODO Auto-generated catch block
e1.printStackTrace();
}
String user = prop.getProperty("user");
String password = prop.getProperty("password");
String url = prop.getProperty("url");
Properties props = new Properties();
PGProperty.USER.set(props, user);
PGProperty.PASSWORD.set(props, password);
PGProperty.ASSUME_MIN_SERVER_VERSION.set(props, "9.4");
PGProperty.REPLICATION.set(props, "database");
PGProperty.PREFER_QUERY_MODE.set(props, "simple");
Connection conn= null;
PGConnection replicationConnection= null;
PGReplicationStream stream = null;
try {
conn = DriverManager.getConnection(url, props);
replicationConnection = conn.unwrap(PGConnection.class);
stream = replicationConnection.getReplicationAPI().replicationStream().logical()
.withSlotName("replication_slot")
.withSlotOption("include-xids", true)
.withSlotOption("include-timestamp", "on")
.withSlotOption("skip-empty-xacts", true)
.withStatusInterval(20, TimeUnit.SECONDS).start();
} catch (SQLException e) {
e.printStackTrace();
}
return stream;
}

}

读取一些数据后，程序给出错误

org.postgresql.util.PSQLException: Database connection failed when reading from copy
at org.postgresql.core.v3.QueryExecutorImpl.readFromCopy(QueryExecutorImpl.java:1028)
at org.postgresql.core.v3.CopyDualImpl.readFromCopy(CopyDualImpl.java:41)
at org.postgresql.core.v3.replication.V3PGReplicationStream.receiveNextData(V3PGReplicationStream.java:155)
at org.postgresql.core.v3.replication.V3PGReplicationStream.readInternal(V3PGReplicationStream.java:124)
at org.postgresql.core.v3.replication.V3PGReplicationStream.readPending(V3PGReplicationStream.java:78)
at Server.main(Server.java:47)
Caused by: java.net.SocketException: Socket closed
at java.net.SocketInputStream.socketRead0(Native Method)
at java.net.SocketInputStream.socketRead(SocketInputStream.java:116)
at java.net.SocketInputStream.read(SocketInputStream.java:171)
at java.net.SocketInputStream.read(SocketInputStream.java:141)
at org.postgresql.core.VisibleBufferedInputStream.readMore(VisibleBufferedInputStream.java:140)
at org.postgresql.core.VisibleBufferedInputStream.ensureBytes(VisibleBufferedInputStream.java:109)
at org.postgresql.core.VisibleBufferedInputStream.read(VisibleBufferedInputStream.java:191)
at org.postgresql.core.PGStream.receive(PGStream.java:495)
at org.postgresql.core.PGStream.receive(PGStream.java:479)
at org.postgresql.core.v3.QueryExecutorImpl.processCopyResults(QueryExecutorImpl.java:1161)
at org.postgresql.core.v3.QueryExecutorImpl.readFromCopy(QueryExecutorImpl.java:1026)
... 5 more

我认为这两种方法之间没有任何区别。但是该程序的行为有所不同。有人可以解释一下，第二种方法有什么问题。

Answer 1

我相信你对Connection的问题.一旦您的函数返回，它就会超出范围，由垃圾收集器来收集它并最终确定。在 finalize 中，连接已关闭，然后您的程序可能会失败。尝试在 main 方法中可用的范围内移动连接和其他必需的中间变量，然后重试。

Answer 2

在执行 COPY 时关闭连接时存在一些适度的虚假行为。我想知道这是否可能涉及？

如果没记错的话，问题是终止消息(在 Connection.close(( 上发送，以指示客户端想要结束 协议连接(相对于 COPY协议操作不同步。这意味着从理论上讲，您的终止消息 可以填充到复制数据消息的中间。在实践中，我只看到服务器抱怨的错误 关于在 COPY 期间意外的消息类型(终止(，因为客户端应该在终止之前发送 CopyDone 或 CopyFail 而副本有效。不过，我认为有可能让消息内容被嚼碎。

我认为您的代码可能容易受到此影响，因为您在单独的函数中启动了一堆 COPY 操作，并且连接可能会被垃圾回收。

正如我在日志中看到的那样，当复制协议仍在读取数据时，套接字连接已关闭。

org.postgresql.util.PSQLException: Database connection failed when reading from copy
at org.postgresql.core.v3.QueryExecutorImpl.readFromCopy(QueryExecutorImpl.java:1028)
at org.postgresql.core.v3.CopyDualImpl.readFromCopy(CopyDualImpl.java:41)
at org.postgresql.core.v3.replication.V3PGReplicationStream.receiveNextData(V3PGReplicationStream.java:155)
at org.postgresql.core.v3.replication.V3PGReplicationStream.readInternal(V3PGReplicationStream.java:124)
at org.postgresql.core.v3.replication.V3PGReplicationStream.readPending(V3PGReplicationStream.java:78)
at Server.main(Server.java:47)
Caused by: java.net.SocketException: Socket closed
at java.net.SocketInputStream.socketRead0(Native Method)
at java.net.SocketInputStream.socketRead(SocketInputStream.java:116)
at java.net.SocketInputStream.read(SocketInputStream.java:171)
at java.net.SocketInputStream.read(SocketInputStream.java:141)
at org.postgresql.core.VisibleBufferedInputStream.readMore(VisibleBufferedInputStream.java:140)
at org.postgresql.core.VisibleBufferedInputStream.ensureBytes(VisibleBufferedInputStream.java:109)