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

У нас есть служба на 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<Job> getNextJob(String queue) {
    Optional<Job> 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);
    }
}