Управление соединениями JDBC с использованием виртуальных потоков Java ⇐ JAVA

[Anonymous]

У нас есть служба на Java 17, которая выполняет логику на нашей стороне и отправляет запрос в стороннюю систему. Время ответа от сторонней системы составляет около 800-1400 мс.
Для этого у нас есть ThreadPoolExecutor размером 12 (нельзя увеличивать его дальше из-за ограничений инфраструктуры)
Наша скорость отправки запросов составляла 5–8 запросов в секунду из-за времени ответа сторонней системы. Чтобы увеличить пропускную способность, мы обновляемся до Java 21, чтобы использовать виртуальные потоки. Однако во время тестирования разработки, похоже, возникла еще одна проблема.
Наша скорость отправки запросов увеличилась примерно до 50 запросов/с. Но мы сталкиваемся с исключениями, связанными с сбоем запроса на соединение JDBC. Размер моего пула подключений к базе данных должен быть около 100.
Когда моя служба запускается, к моему микросервису принадлежит 10 подключений.
При отправке запросов с помощью ThreadPoolExecutor: 13 дополнительных подключений, всего 23 .
При использовании newVirtualThreadPerTaskExecutor: всего 94 соединения.
Я использую следующий запрос для мониторинга своей базы данных PostGre V13.9
SELECT pid, datname, usename, application_name, client_addr, client_port, backend_start, query_start, state_change, query, state
FROM pg_stat_activity where application_name ='PostgreSQL JDBC Driver';

Исключения:
org.springframework.transaction.CannotCreateTransactionException: Could not open JPA EntityManager for transaction
at org.springframework.orm.jpa.JpaTransactionManager.doBegin(JpaTransactionManager.java:466) ~[spring-orm-6.1.2.jar:6.1.2]
at org.springframework.transaction.support.AbstractPlatformTransactionManager.startTransaction(AbstractPlatformTransactionManager.java:531) ~[spring-tx-6.1.2.jar:6.1.2]
at org.springframework.transaction.support.AbstractPlatformTransactionManager.getTransaction(AbstractPlatformTransactionManager.java:405) ~[spring-tx-6.1.2.jar:6.1.2]
at org.springframework.transaction.interceptor.TransactionAspectSupport.createTransactionIfNecessary(TransactionAspectSupport.java:610) ~[spring-tx-6.1.2.jar:6.1.2]

at java.base/java.util.concurrent.ThreadPerTaskExecutor$TaskRunner.run(ThreadPerTaskExecutor.java:314) ~[na:na]
at java.base/java.lang.VirtualThread.run(VirtualThread.java:309) ~[na:na]
Caused by: org.hibernate.exception.GenericJDBCException: Unable to acquire JDBC Connection [FATAL: remaining connection slots are reserved for non-replication superuser connections] [n/a]

at java.base/java.util.concurrent.ThreadPerTaskExecutor$TaskRunner.run(ThreadPerTaskExecutor.java:314) ~[na:na]
at java.base/java.lang.VirtualThread.run(VirtualThread.java:309) ~[na:na]
Caused by: org.hibernate.exception.GenericJDBCException: Unable to acquire JDBC Connection [FATAL: remaining connection slots are reserved for non-replication superuser connections] [n/a]
at org.hibernate.exception.internal.StandardSQLExceptionConverter.convert(StandardSQLExceptionConverter.java:63) ~[hibernate-core-6.4.1.Final.jar:6.4.1.Final]

Я понимаю, что при использовании ThreadPoolExecutor из 12 потоков каждый поток устанавливает соединение и использует его. При использовании виртуальных потоков, как только какой-либо виртуальный поток занят ожиданием ответа или Thread.sleep, другой виртуальный поток берет на себя инициативу. Как только я запускаю нагрузочный тест, я замечаю всплеск подключений, некоторые из которых приводят к исключениям (например, 11 запросов не удалось из 1500 (из-за ошибки JDBC)), а затем процесс продвигается плавно, используя максимум 94 соединения. на данный момент у меня есть.
Вопрос в том, как мне решить эту проблему и ограничить количество процессов, запускаемых одновременно. И что еще более важно, ограничьте количество подключений/подключений повторного использования, создаваемых к базе данных. У меня есть несколько других приложений в базе данных, поэтому я не смогу их значительно увеличить.
Я пробовал использовать системное свойство "jdk.virtualThreadScheduler.maxPoolSize", передав аргумент VM -Djdk .virtualThreadScheduler.maxPoolSize=5
Но исключение все равно возникает, даже если установлено -Djdk.virtualThreadScheduler.maxPoolSize=1
Что мне не хватает? Следует ли использовать виртуальные потоки для основных задач ввода-вывода и задач ЦП, а не для связи со сторонними системами?
Некоторые коды того, как инициируются мои виртуальные потоки:
this.executorService = Executors.newSingleThreadScheduledExecutor();
this.taskExecutorService = Executors.newVirtualThreadPerTaskExecutor();

this.executorService.scheduleAtFixedRate(this, 0L, this.pollingTime.toMillis(), TimeUnit.MILLISECONDS); //Called once when service starts

public void run() {
this.taskExecutorService.execute(//my runnable job);
}

Я также пробовал использовать счетчики/семафоры для ограничения виртуальных потоков. Это снижает пропускную способность, но по-прежнему создает больше новых соединений по сравнению с ThreadPoolExecutor. Это спроектировано таким образом? Я подозреваю, что каждый виртуальный поток/процесс создает новое соединение, тогда как в ThreadPoolExecutor он использует соединение, назначенное его PID. Есть какие-нибудь разъяснения или предложения?
Изменить: (код соединения JDBC)
Я вижу большинство запросов на удаление в открытых соединениях, поскольку именно это и выполняется. при отправке запросов через виртуальные потоки. Ниже приведен код.
public Optional getNextJob(String queue) {
Optional job = Optional.empty();

try {
Connection connection = this.dataSource.getConnection();

try {
String sql = "DELETE FROM scheduler_task WHERE id = (SELECT id FROM scheduler_task WHERE queue = ? and trigger_date < now() LIMIT 1 FOR UPDATE SKIP LOCKED) RETURNING id, queue, reference_id, trigger_date";
PreparedStatement stmt = connection.prepareStatement(sql);

try {
ResultSet resultSet = stmt.executeQuery();

try {
if (resultSet.next()) {
job = Optional.of(this.mapToJob(resultSet));
}
} catch (Throwable var14) {
if (resultSet != null) {
try {
resultSet.close();
} catch (Throwable var13) {
var14.addSuppressed(var13);
}
}
throw var14;
}

if (resultSet != null) {
resultSet.close();
}
} catch (Throwable var15) {
if (stmt != null) {
try {
stmt.close();
} catch (Throwable var12) {
var15.addSuppressed(var12);
}
}

throw var15;
}
if (stmt != null) {
stmt.close();
}
} catch (Throwable var16) {
if (connection != null) {
try {
connection.close();
} catch (Throwable var11) {
var16.addSuppressed(var11);
}
}

throw var16;
}
if (connection != null) {
connection.close();
}

return job;
} catch (SQLException var17) {
log.error("error", var17);
throw new Exception(var17);
}
}


Подробнее здесь: https://stackoverflow.com/questions/784 ... al-threads