From dc362bb0ca33f972f73a4d5ef867c458dab64300 Mon Sep 17 00:00:00 2001 From: Mark Fisher Date: Mon, 7 Jan 2008 19:31:21 +0000 Subject: [PATCH] Added dispatcher error handling to the message bus and refactored UnicastMessageDispatcher to properly iterate until success or rejection limit is reached. --- .../integration/bus/ConsumerPolicy.java | 18 +++--- .../integration/bus/MessageBus.java | 52 ++++++++++++++-- .../bus/UnicastMessageDispatcher.java | 62 ++++++++++--------- 3 files changed, 90 insertions(+), 42 deletions(-) diff --git a/spring-integration-core/src/main/java/org/springframework/integration/bus/ConsumerPolicy.java b/spring-integration-core/src/main/java/org/springframework/integration/bus/ConsumerPolicy.java index 547426b371..b1e5ca4011 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/bus/ConsumerPolicy.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/bus/ConsumerPolicy.java @@ -33,9 +33,9 @@ public class ConsumerPolicy { private static final int DEFAULT_MAX_MESSAGES_PER_TASK = 1; - private static final int DEFAULT_REJECTION_LIMIT = 10; + private static final int DEFAULT_REJECTION_LIMIT = 5; - private static final int DEFAULT_REJECTION_LIMIT_WAIT = 1000; + private static final int DEFAULT_RETRY_INTERVAL = 1000; private static final long DEFAULT_RECEIVE_TIMEOUT = 1000; @@ -48,7 +48,7 @@ public class ConsumerPolicy { private int rejectionLimit = DEFAULT_REJECTION_LIMIT; - private int rejectionLimitWait = DEFAULT_REJECTION_LIMIT_WAIT; + private int retryInterval = DEFAULT_RETRY_INTERVAL; private int initialDelay = 0; @@ -82,7 +82,7 @@ public class ConsumerPolicy { ConsumerPolicy policy = new ConsumerPolicy(); policy.setPeriod(-1); policy.setConcurrency(1); - policy.setMaxConcurrency(1); + policy.setMaxConcurrency(10); return policy; } @@ -157,17 +157,17 @@ public class ConsumerPolicy { public void setRejectionLimit(int rejectionLimit) { if (rejectionLimit < 1) { - throw new IllegalArgumentException("'idleTaskExecutionLimit' must be at least 1"); + throw new IllegalArgumentException("'rejectionLimit' must be at least 1"); } this.rejectionLimit = rejectionLimit; } - public int getRejectionLimitWait() { - return this.rejectionLimitWait; + public int getRetryInterval() { + return this.retryInterval; } - public void setRejectionLimitWait(int rejectionLimitWait) { - this.rejectionLimitWait = rejectionLimitWait; + public void setRetryInterval(int retryInterval) { + this.retryInterval = retryInterval; } public long getReceiveTimeout() { diff --git a/spring-integration-core/src/main/java/org/springframework/integration/bus/MessageBus.java b/spring-integration-core/src/main/java/org/springframework/integration/bus/MessageBus.java index ea11c3ec2d..3f5c0ac31b 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/bus/MessageBus.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/bus/MessageBus.java @@ -38,7 +38,9 @@ import org.springframework.integration.channel.DefaultChannelRegistry; import org.springframework.integration.channel.MessageChannel; import org.springframework.integration.channel.PointToPointChannel; import org.springframework.integration.endpoint.MessageEndpoint; +import org.springframework.integration.message.ErrorMessage; import org.springframework.integration.message.MessageReceiver; +import org.springframework.scheduling.concurrent.CustomizableThreadFactory; import org.springframework.util.Assert; import org.springframework.util.StringUtils; @@ -66,6 +68,8 @@ public class MessageBus implements ChannelRegistry, ApplicationContextAware, Lif private ScheduledThreadPoolExecutor dispatcherExecutor; + private int dispatcherPoolSize = 10; + private boolean autoCreateChannels; private boolean running; @@ -82,6 +86,21 @@ public class MessageBus implements ChannelRegistry, ApplicationContextAware, Lif this.activateSubscriptions(applicationContext); } + /** + * Set the size for the dispatcher thread pool. + */ + public void setDispatcherPoolSize(int dispatcherPoolSize) { + Assert.isTrue(dispatcherPoolSize > 0, "'dispatcherPoolSize' must be at least 1"); + this.dispatcherPoolSize = dispatcherPoolSize; + if (this.dispatcherExecutor != null) { + this.dispatcherExecutor.setCorePoolSize(dispatcherPoolSize); + } + } + + /** + * Set whether the bus should automatically create a channel when a + * subscription contains the name of a previously unregistered channel. + */ public void setAutoCreateChannels(boolean autoCreateChannels) { this.autoCreateChannels = autoCreateChannels; } @@ -136,7 +155,17 @@ public class MessageBus implements ChannelRegistry, ApplicationContextAware, Lif } public void initialize() { - this.dispatcherExecutor = new ScheduledThreadPoolExecutor(this.dispatcherTasks.size() > 0 ? this.dispatcherTasks.size() : 1); + initDispatcherExecutor(); + if (this.getInvalidMessageChannel() == null) { + this.setInvalidMessageChannel(new PointToPointChannel(Integer.MAX_VALUE)); + } + } + + private void initDispatcherExecutor() { + CustomizableThreadFactory threadFactory = new CustomizableThreadFactory(); + threadFactory.setThreadNamePrefix("dispatcher-executor-"); + threadFactory.setThreadGroup(new ThreadGroup("dispatcher-executors")); + this.dispatcherExecutor = new ScheduledThreadPoolExecutor(this.dispatcherPoolSize, threadFactory); } public MessageChannel getInvalidMessageChannel() { @@ -179,7 +208,6 @@ public class MessageBus implements ChannelRegistry, ApplicationContextAware, Lif } public void registerSourceAdapter(String name, SourceAdapter adapter) { - // TODO: use the name if (adapter instanceof MessageDispatcher) { MessageDispatcher dispatcher = (MessageDispatcher) adapter; ConsumerPolicy policy = dispatcher.getConsumerPolicy(); @@ -348,8 +376,19 @@ public class MessageBus implements ChannelRegistry, ApplicationContextAware, Lif } } + private void handleDispatchError(DispatcherTask task, Throwable t) { + try { + this.getInvalidMessageChannel().send(new ErrorMessage(t), 1000); + } + catch (Throwable ignore) { // message will be logged only + } + if (logger.isWarnEnabled()) { + logger.warn("failure occurred while dispatching message", t); + } + } - private static class DispatcherTask implements Runnable { + + private class DispatcherTask implements Runnable { private MessageDispatcher dispatcher; @@ -367,7 +406,12 @@ public class MessageBus implements ChannelRegistry, ApplicationContextAware, Lif } public void run() { - dispatcher.dispatch(); + try { + dispatcher.dispatch(); + } + catch (Throwable t) { + handleDispatchError(this, t); + } } } diff --git a/spring-integration-core/src/main/java/org/springframework/integration/bus/UnicastMessageDispatcher.java b/spring-integration-core/src/main/java/org/springframework/integration/bus/UnicastMessageDispatcher.java index fad085015d..132e8ffc3b 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/bus/UnicastMessageDispatcher.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/bus/UnicastMessageDispatcher.java @@ -19,6 +19,7 @@ package org.springframework.integration.bus; import java.util.Iterator; import java.util.concurrent.RejectedExecutionException; +import org.springframework.integration.MessageDeliveryException; import org.springframework.integration.message.Message; /** @@ -45,47 +46,50 @@ public class UnicastMessageDispatcher extends AbstractMessageDispatcher { @Override protected boolean dispatchMessage(Message message) { int attempts = 0; - Iterator iter = this.getExecutors().iterator(); - if (!iter.hasNext()) { - if (logger.isWarnEnabled()) { - logger.warn("dispatcher has no active executors"); + while (attempts < policy.getRejectionLimit()) { + if (attempts > 0) { + if (logger.isDebugEnabled()) { + logger.debug("executor(s) rejected message after " + attempts + + " attempt(s), will try again after 'retryInterval' of " + this.policy.getRetryInterval() + + " milliseconds"); + } + try { + Thread.sleep(policy.getRetryInterval()); + } + catch (InterruptedException iex) { + Thread.currentThread().interrupt(); + return false; + } } - return false; - } - while (iter.hasNext()) { - MessageReceivingExecutor executor = iter.next(); - try { + Iterator iter = this.getExecutors().iterator(); + if (!iter.hasNext()) { + if (logger.isWarnEnabled()) { + logger.warn("dispatcher has no active executors"); + } + return false; + } + while (iter.hasNext()) { + MessageReceivingExecutor executor = iter.next(); if (executor == null || !executor.isRunning()) { if (logger.isInfoEnabled()) { logger.info("skipping inactive executor"); } continue; } - executor.processMessage(message); - return true; - } - catch (RejectedExecutionException rex) { - attempts++; - if (attempts == policy.getRejectionLimit()) { - attempts = 0; + try { + executor.processMessage(message); + return true; + } + catch (RejectedExecutionException rex) { if (logger.isDebugEnabled()) { - logger.debug("reached rejected execution limit"); - } - try { - Thread.sleep(policy.getRejectionLimitWait()); - } - catch (InterruptedException iex) { - Thread.currentThread().interrupt(); + logger.debug("executor rejected task, continuing with other executors if available", rex); } } } - catch (Exception e) { - if (logger.isWarnEnabled()) { - logger.warn("error occurred during dispatch", e); - } - } + attempts++; } - return false; + throw new MessageDeliveryException("Dispatcher reached rejection limit of " + + this.policy.getRejectionLimit() + ". Consider increasing the concurrency and/or raising the limit."); } }