You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@camel.apache.org by da...@apache.org on 2021/03/22 15:58:16 UTC
[camel] branch master updated: CAMEL-16383: camel-scheduler -
ConcurrentTasks option renamed to poolSize and avoid scheduler to cause
concurrent triggers that causes routing problems.
This is an automated email from the ASF dual-hosted git repository.
davsclaus pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/camel.git
The following commit(s) were added to refs/heads/master by this push:
new 07c60df CAMEL-16383: camel-scheduler - ConcurrentTasks option renamed to poolSize and avoid scheduler to cause concurrent triggers that causes routing problems.
07c60df is described below
commit 07c60df69280cfe282148d146d8c2775d43cd9b3
Author: Claus Ibsen <cl...@gmail.com>
AuthorDate: Mon Mar 22 16:57:48 2021 +0100
CAMEL-16383: camel-scheduler - ConcurrentTasks option renamed to poolSize and avoid scheduler to cause concurrent triggers that causes routing problems.
---
.../camel/component/aws2/sqs/Sqs2Endpoint.java | 2 +-
.../camel/component/ironmq/IronMQEndpoint.java | 2 +-
.../DefaultScheduledPollConsumerScheduler.java | 21 +++++++++++++++++----
3 files changed, 19 insertions(+), 6 deletions(-)
diff --git a/components/camel-aws/camel-aws2-sqs/src/main/java/org/apache/camel/component/aws2/sqs/Sqs2Endpoint.java b/components/camel-aws/camel-aws2-sqs/src/main/java/org/apache/camel/component/aws2/sqs/Sqs2Endpoint.java
index 1eb4014..40451c0 100644
--- a/components/camel-aws/camel-aws2-sqs/src/main/java/org/apache/camel/component/aws2/sqs/Sqs2Endpoint.java
+++ b/components/camel-aws/camel-aws2-sqs/src/main/java/org/apache/camel/component/aws2/sqs/Sqs2Endpoint.java
@@ -108,7 +108,7 @@ public class Sqs2Endpoint extends ScheduledPollEndpoint implements HeaderFilterS
scheduler.setUseFixedDelay(sqsConsumer.isUseFixedDelay());
scheduler.setInitialDelay(sqsConsumer.getInitialDelay());
scheduler.setTimeUnit(sqsConsumer.getTimeUnit());
- scheduler.setConcurrentTasks(configuration.getConcurrentConsumers());
+ scheduler.setConcurrentConsumers(configuration.getConcurrentConsumers());
sqsConsumer.setScheduler(scheduler);
return sqsConsumer;
}
diff --git a/components/camel-ironmq/src/main/java/org/apache/camel/component/ironmq/IronMQEndpoint.java b/components/camel-ironmq/src/main/java/org/apache/camel/component/ironmq/IronMQEndpoint.java
index 4e8c116..1eda2fc 100644
--- a/components/camel-ironmq/src/main/java/org/apache/camel/component/ironmq/IronMQEndpoint.java
+++ b/components/camel-ironmq/src/main/java/org/apache/camel/component/ironmq/IronMQEndpoint.java
@@ -66,7 +66,7 @@ public class IronMQEndpoint extends ScheduledPollEndpoint {
scheduler.setUseFixedDelay(ironMQConsumer.isUseFixedDelay());
scheduler.setInitialDelay(ironMQConsumer.getInitialDelay());
scheduler.setTimeUnit(ironMQConsumer.getTimeUnit());
- scheduler.setConcurrentTasks(configuration.getConcurrentConsumers());
+ scheduler.setConcurrentConsumers(configuration.getConcurrentConsumers());
ironMQConsumer.setScheduler(scheduler);
return ironMQConsumer;
}
diff --git a/core/camel-support/src/main/java/org/apache/camel/support/DefaultScheduledPollConsumerScheduler.java b/core/camel-support/src/main/java/org/apache/camel/support/DefaultScheduledPollConsumerScheduler.java
index af6aa81..3d77a6a 100644
--- a/core/camel-support/src/main/java/org/apache/camel/support/DefaultScheduledPollConsumerScheduler.java
+++ b/core/camel-support/src/main/java/org/apache/camel/support/DefaultScheduledPollConsumerScheduler.java
@@ -48,6 +48,7 @@ public class DefaultScheduledPollConsumerScheduler extends ServiceSupport implem
private volatile List<ScheduledFuture<?>> futures = new ArrayList<>();
private Runnable task;
private int poolSize = 1;
+ private int concurrentConsumers = 1;
private long initialDelay = -1;
private long delay = -1;
@@ -111,6 +112,14 @@ public class DefaultScheduledPollConsumerScheduler extends ServiceSupport implem
this.scheduledExecutorService = scheduledExecutorService;
}
+ public int getConcurrentConsumers() {
+ return concurrentConsumers;
+ }
+
+ public void setConcurrentConsumers(int concurrentConsumers) {
+ this.concurrentConsumers = concurrentConsumers;
+ }
+
public int getPoolSize() {
return poolSize;
}
@@ -170,16 +179,20 @@ public class DefaultScheduledPollConsumerScheduler extends ServiceSupport implem
currentInitialDelay, currentDelay, getTimeUnit().name().toLowerCase(Locale.ENGLISH),
consumer.getEndpoint());
}
- futures.add(scheduledExecutorService.scheduleWithFixedDelay(task, currentInitialDelay, currentDelay,
- getTimeUnit()));
+ for (int i = 0; i < concurrentConsumers; i++) {
+ futures.add(scheduledExecutorService.scheduleWithFixedDelay(task, currentInitialDelay, currentDelay,
+ getTimeUnit()));
+ }
} else {
if (LOG.isDebugEnabled()) {
LOG.debug("Scheduling poll (fixed rate) with initialDelay: {}, delay: {} ({}) for: {}",
currentInitialDelay, currentDelay, getTimeUnit().name().toLowerCase(Locale.ENGLISH),
consumer.getEndpoint());
}
- futures.add(scheduledExecutorService.scheduleAtFixedRate(task, currentInitialDelay, currentDelay,
- getTimeUnit()));
+ for (int i = 0; i < concurrentConsumers; i++) {
+ futures.add(scheduledExecutorService.scheduleAtFixedRate(task, currentInitialDelay, currentDelay,
+ getTimeUnit()));
+ }
}
}
}