Improve throughput in case of oracle database

jonasrutishauser · Jul 28, 2024 · 9b75af2 · 9b75af2
1 parent 4f99a65
commit 9b75af2
Show file tree

Hide file tree

Showing 3 changed files with 8 additions and 4 deletions.
diff --git a/.../main/java/com/github/jonasrutishauser/transactional/event/core/store/DispatcherImpl.java b/.../main/java/com/github/jonasrutishauser/transactional/event/core/store/DispatcherImpl.java
@@ -118,7 +118,7 @@ public synchronized void schedule() {
         } catch (RejectedExecutionException e) {
             LOGGER.warn("Failed to dispatch events: {}", e.getMessage());
         }
-        if (processed || maxAquire <= 0) {
+        if (processed || maxAquire <= configuration.getMaxConcurrentDispatching() / 10) {
             intervalSeconds = 0;
         } else {
             intervalSeconds = min(configuration.getMaxDispatchInterval(), max(intervalSeconds * 2, 1));

diff --git a/...in/java/com/github/jonasrutishauser/transactional/event/core/store/PendingEventStore.java b/...in/java/com/github/jonasrutishauser/transactional/event/core/store/PendingEventStore.java
@@ -260,7 +260,7 @@ public Set<String> aquire(int maxAquire) {
              PreparedStatement aquireStatement = connection.prepareStatement(aquireSQL);
              ResultSet resultSet = executeQuery(aquireStatement, lockOwner.getMinUntilForAquire(), limit);
              PreparedStatement updateStatement = connection.prepareStatement(updateSQL)) {
-            while (resultSet.next()) {
+            while (limit-- > 0 && resultSet.next()) {
                 result.add(resultSet.getString("id"));
                 updateStatement.setInt(1, resultSet.getInt("tries"));
                 updateStatement.setString(2, lockOwner.getId());
@@ -305,7 +305,11 @@ private ResultSet executeQuery(PreparedStatement statement, int intParam) throws
 
     private ResultSet executeQuery(PreparedStatement statement, long param1, int param2) throws SQLException {
         statement.setLong(1, param1);
-        statement.setInt(2, param2);
+        if (statement.getParameterMetaData().getParameterCount() > 1) {
+            statement.setInt(2, param2);
+        } else {
+            statement.setFetchSize(1);
+        }
         return statement.executeQuery();
     }
 

diff --git a/.../java/com/github/jonasrutishauser/transactional/event/core/store/QueryAdapterFactory.java b/.../java/com/github/jonasrutishauser/transactional/event/core/store/QueryAdapterFactory.java
@@ -91,7 +91,7 @@ public String addSkipLocked(String sql) {
     private static class SkipLockedQueryAdapter extends SimpleQueryAdapter {
         @Override
         public String addSkipLocked(String sql) {
-            return sql.replace("FOR UPDATE", "FOR UPDATE SKIP LOCKED");
+            return sql.replaceAll(LIMIT_EXPRESSION, "").replace("FOR UPDATE", "FOR UPDATE SKIP LOCKED");
         }
     }