From 4550bd43d73584a35377d2bb5a16ac40160aade4 Mon Sep 17 00:00:00 2001 From: Jennifer Hickey Date: Thu, 22 Aug 2013 14:16:12 -0700 Subject: [PATCH] Minor enhancements to listener resubscribe DATAREDIS-231 - Do not stop the container on Exception, as it will attempt unnecessary unsubscribe - Retry connect indefinitely until container is stopped - Ensure connection is closed after subscription failure - Set listening to false if SubscriptionTask errors out for any reason - Change default recovery interval to 5 seconds --- .../RedisMessageListenerContainer.java | 110 +++++++++--------- 1 file changed, 54 insertions(+), 56 deletions(-) diff --git a/src/main/java/org/springframework/data/redis/listener/RedisMessageListenerContainer.java b/src/main/java/org/springframework/data/redis/listener/RedisMessageListenerContainer.java index 1f7c8784b..b2dae5252 100644 --- a/src/main/java/org/springframework/data/redis/listener/RedisMessageListenerContainer.java +++ b/src/main/java/org/springframework/data/redis/listener/RedisMessageListenerContainer.java @@ -66,6 +66,7 @@ import org.springframework.util.ErrorHandler; * * * @author Costin Leau + * @author Jennifer Hickey * @author Way Joke */ public class RedisMessageListenerContainer implements InitializingBean, DisposableBean, BeanNameAware, SmartLifecycle { @@ -80,6 +81,11 @@ public class RedisMessageListenerContainer implements InitializingBean, Disposab public static final String DEFAULT_THREAD_NAME_PREFIX = ClassUtils.getShortName(RedisMessageListenerContainer.class) + "-"; + /** + * The default recovery interval: 5000 ms = 5 seconds. + */ + public static final long DEFAULT_RECOVERY_INTERVAL = 5000; + private long initWait = TimeUnit.SECONDS.toMillis(5); private Executor subscriptionExecutor; @@ -119,12 +125,7 @@ public class RedisMessageListenerContainer implements InitializingBean, Disposab private volatile RedisSerializer serializer = new StringRedisSerializer(); - // will recovery redis listening after recovery tnterval setting - private long recoveryInterval = 3000L; - // will recovery redis listening until reach the limit ( -1 will always try again) - private int multipleRetries = -1; - // the retried times - private int numberOfRetries = 0; + private long recoveryInterval = DEFAULT_RECOVERY_INTERVAL; public void afterPropertiesSet() { if (taskExecutor == null) { @@ -640,41 +641,37 @@ public class RedisMessageListenerContainer implements InitializingBean, Disposab } /** - * Handle subscription task exception, will recovery redis listening on RedisConnectionFailureExceptions. - * @param t Throwable exception + * Handle subscription task exception. Will attempt to restart the subscription + * if the Exception is a connection failure (for example, Redis was restarted). + * @param ex Throwable exception */ - protected void refreshConnectionUntilSuccessful(Throwable t) { - // we need to specifically deal with RedisConnectionFailureExceptions - // so we aren't attempting to reconnect if other types of failures occur. - if (t instanceof RedisConnectionFailureException) { - - if (multipleRetries > 0 && numberOfRetries > multipleRetries) { - logger.error("Retry count has reached the upper limit " + multipleRetries); - return; + protected void handleSubscriptionException(Throwable ex) { + listening = false; + subscriptionTask.closeConnection(); + if(ex instanceof RedisConnectionFailureException) { + if(isRunning()) { + logger.error("Connection failure occurred. Restarting subscription task after " + + recoveryInterval + " ms"); + sleepBeforeRecoveryAttempt(); + lazyListen(); } + } else { + logger.error("SubscriptionTask aborted with exception:", ex); + } + } - subscriptionExecutor.execute(new Runnable() { - - public void run() { - if (multipleRetries > 0) { - numberOfRetries++; - } - try { - stop(); - } catch (Throwable ex) { - // as we know the stop will throw JedisConnectionException on jedis, - // catch Throwable will not need to import any jedis classes. - // The exception will be ignored. - } - logger.error("Restarting subscription task after " + recoveryInterval + "ms."); - try { - Thread.sleep(recoveryInterval); - } catch (InterruptedException e) { - return; - } - start(); - } - }); + /** + * Sleep according to the specified recovery interval. + * Called between recovery attempts. + */ + protected void sleepBeforeRecoveryAttempt() { + if (this.recoveryInterval > 0) { + try { + Thread.sleep(this.recoveryInterval); + } + catch (InterruptedException interEx) { + logger.debug("Thread interrupted while sleeping the recovery interval"); + } } } @@ -743,8 +740,8 @@ public class RedisMessageListenerContainer implements InitializingBean, Disposab synchronized (localMonitor) { subscriptionTaskRunning = true; } - connection = connectionFactory.getConnection(); try { + connection = connectionFactory.getConnection(); if (connection.isSubscribed()) { throw new IllegalStateException("Retrieved connection is already subscribed; aborting listening"); } @@ -767,8 +764,7 @@ public class RedisMessageListenerContainer implements InitializingBean, Disposab connection.pSubscribe(new DispatchMessageListener(), unwrap(patternMapping.keySet())); } } catch (Throwable t) { - logger.error("SubscriptionTask abort with exception:", t); - refreshConnectionUntilSuccessful(t); + handleSubscriptionException(t); } finally { // this block is executed once the subscription thread has ended, this may or may not mean // the connection has been unsubscribed, depending on driver @@ -818,8 +814,7 @@ public class RedisMessageListenerContainer implements InitializingBean, Disposab } } if(!subscriptionTaskRunning) { - logger.trace("Closing connection"); - connection.close(); + closeConnection(); }else { logger.warn("Unable to close connection. Subscription task still running"); } @@ -828,6 +823,18 @@ public class RedisMessageListenerContainer implements InitializingBean, Disposab } } + void closeConnection() { + if(connection != null) { + logger.trace("Closing connection"); + try { + connection.close(); + } catch(Exception e) { + logger.warn("Error closing subscription connection", e); + } + connection = null; + } + } + void subscribeChannel(byte[]... channels) { if (channels != null && channels.length > 0) { if (connection != null) { @@ -920,20 +927,11 @@ public class RedisMessageListenerContainer implements InitializingBean, Disposab } /** - * Set the recovery interval value(ms), must be greater than zero - * @param recoveryInterval + * Specify the interval between recovery attempts, in milliseconds. + * The default is 5000 ms, that is, 5 seconds. + * @see #handleSubscriptionException */ public void setRecoveryInterval(long recoveryInterval) { - if (recoveryInterval > 0) { - this.recoveryInterval = recoveryInterval; - } - } - - /** - * Set the multiple retries times, less than zero on behalf of the infinite. - * @param multipleRetries - */ - public void setMultipleRetries(int multipleRetries) { - this.multipleRetries = multipleRetries; + this.recoveryInterval = recoveryInterval; } } \ No newline at end of file