> {
return idleConnectionTimeout;
}
- /**
- * Reconnection attempt timeout to Redis server when
- * it has been excluded from internal list of available servers.
- *
- * On every such timeout event Redisson tries
- * to connect to disconnected Redis server.
- *
- * Default is 3000
- *
- * @see #failedAttempts
- *
- * @param slaveRetryTimeout - retry timeout in milliseconds
- * @return config
+ /*
+ * Use setFailedSlaveReconnectionInterval instead
*/
-
+ @Deprecated
public T setReconnectionTimeout(int slaveRetryTimeout) {
- this.reconnectionTimeout = slaveRetryTimeout;
+ log.warn("'reconnectionTimeout' setting in unavailable. Please use 'failedSlaveReconnectionInterval' setting instead!");
return (T) this;
}
- public int getReconnectionTimeout() {
- return reconnectionTimeout;
- }
-
- /**
- * Redis server will be excluded from the internal list of available nodes
- * when sequential unsuccessful execution attempts of any Redis command
- * on this server reaches failedAttempts.
- *
- * Default is 3
- *
- * @param slaveFailedAttempts - attempts
- * @return config
+ /*
+ * Use setFailedSlaveCheckInterval instead
*/
+ @Deprecated
public T setFailedAttempts(int slaveFailedAttempts) {
- this.failedAttempts = slaveFailedAttempts;
+ log.warn("'failedAttempts' setting in unavailable. Please use 'failedSlaveCheckInterval' setting instead!");
return (T) this;
}
-
- public int getFailedAttempts() {
- return failedAttempts;
- }
-
+
public boolean isSslEnableEndpointIdentification() {
return sslEnableEndpointIdentification;
}
diff --git a/redisson/src/main/java/org/redisson/config/BaseMasterSlaveServersConfig.java b/redisson/src/main/java/org/redisson/config/BaseMasterSlaveServersConfig.java
index 406f10e97..ff606d632 100644
--- a/redisson/src/main/java/org/redisson/config/BaseMasterSlaveServersConfig.java
+++ b/redisson/src/main/java/org/redisson/config/BaseMasterSlaveServersConfig.java
@@ -41,6 +41,10 @@ public class BaseMasterSlaveServersConfigeach slave node
*/
@@ -82,6 +86,8 @@ public class BaseMasterSlaveServersConfig
+ * On every such timeout event Redisson tries
+ * to connect to disconnected Redis server.
+ *
+ * Default is 3000
+ *
+ * @param failedSlavesReconnectionTimeout - retry timeout in milliseconds
+ * @return config
+ */
+
+ public T setFailedSlaveReconnectionInterval(int failedSlavesReconnectionTimeout) {
+ this.failedSlaveReconnectionInterval = failedSlavesReconnectionTimeout;
+ return (T) this;
+ }
+
+ public int getFailedSlaveReconnectionInterval() {
+ return failedSlaveReconnectionInterval;
+ }
+
+
+ /**
+ * Redis Slave node is excluded from the internal list of available nodes
+ * when the time interval from the moment of first Redis command execution failure
+ * on this server reaches slaveFailsInterval value.
+ *
+ * Default is 60000
+ *
+ * @param slaveFailsInterval - time interval in milliseconds
+ * @return config
+ */
+ public T setFailedSlaveCheckInterval(int slaveFailsInterval) {
+ this.failedSlaveCheckInterval = slaveFailsInterval;
+ return (T) this;
+ }
+ public int getFailedSlaveCheckInterval() {
+ return failedSlaveCheckInterval;
+ }
/**
* Redis 'master' server connection pool size.
diff --git a/redisson/src/main/java/org/redisson/connection/ClientConnectionsEntry.java b/redisson/src/main/java/org/redisson/connection/ClientConnectionsEntry.java
index 0d1dd6d5a..b622a457d 100644
--- a/redisson/src/main/java/org/redisson/connection/ClientConnectionsEntry.java
+++ b/redisson/src/main/java/org/redisson/connection/ClientConnectionsEntry.java
@@ -17,7 +17,7 @@ package org.redisson.connection;
import java.util.Queue;
import java.util.concurrent.ConcurrentLinkedQueue;
-import java.util.concurrent.atomic.AtomicInteger;
+import java.util.concurrent.atomic.AtomicLong;
import org.redisson.api.NodeType;
import org.redisson.api.RFuture;
@@ -57,7 +57,7 @@ public class ClientConnectionsEntry {
private volatile NodeType nodeType;
private ConnectionManager connectionManager;
- private final AtomicInteger failedAttempts = new AtomicInteger();
+ private final AtomicLong firstFailTime = new AtomicLong(0);
public ClientConnectionsEntry(RedisClient client, int poolMinSize, int poolMaxSize, int subscribePoolMinSize, int subscribePoolMaxSize,
ConnectionManager connectionManager, NodeType nodeType) {
@@ -80,16 +80,19 @@ public class ClientConnectionsEntry {
return nodeType;
}
- public void resetFailedAttempts() {
- failedAttempts.set(0);
+ public void resetFirstFail() {
+ firstFailTime.set(0);
}
- public int getFailedAttempts() {
- return failedAttempts.get();
+ public boolean isFailed() {
+ if (firstFailTime.get() != 0) {
+ return System.currentTimeMillis() - firstFailTime.get() > connectionManager.getConfig().getFailedSlaveCheckInterval();
+ }
+ return false;
}
-
- public int incFailedAttempts() {
- return failedAttempts.incrementAndGet();
+
+ public void trySetupFistFail() {
+ firstFailTime.compareAndSet(0, System.currentTimeMillis());
}
public RedisClient getClient() {
@@ -243,7 +246,7 @@ public class ClientConnectionsEntry {
+ ", freeSubscribeConnectionsCounter=" + freeSubscribeConnectionsCounter
+ ", freeConnectionsAmount=" + freeConnections.size() + ", freeConnectionsCounter="
+ freeConnectionsCounter + ", freezed=" + freezed + ", freezeReason=" + freezeReason
- + ", client=" + client + ", nodeType=" + nodeType + ", failedAttempts=" + failedAttempts
+ + ", client=" + client + ", nodeType=" + nodeType + ", firstFail=" + firstFailTime
+ "]";
}
diff --git a/redisson/src/main/java/org/redisson/connection/ConnectionManager.java b/redisson/src/main/java/org/redisson/connection/ConnectionManager.java
index cfab7553a..d6427cb24 100644
--- a/redisson/src/main/java/org/redisson/connection/ConnectionManager.java
+++ b/redisson/src/main/java/org/redisson/connection/ConnectionManager.java
@@ -29,6 +29,7 @@ import org.redisson.client.RedisConnection;
import org.redisson.client.RedisPubSubListener;
import org.redisson.client.codec.Codec;
import org.redisson.client.protocol.RedisCommand;
+import org.redisson.client.protocol.pubsub.PubSubType;
import org.redisson.command.CommandSyncService;
import org.redisson.config.Config;
import org.redisson.config.MasterSlaveServersConfig;
@@ -112,9 +113,7 @@ public interface ConnectionManager {
void unsubscribe(String channelName, AsyncSemaphore lock);
- RFuture unsubscribe(String channelName, boolean temporaryDown);
-
- RFuture punsubscribe(String channelName, boolean temporaryDown);
+ RFuture unsubscribe(String channelName, PubSubType topicType);
void punsubscribe(String channelName, AsyncSemaphore lock);
diff --git a/redisson/src/main/java/org/redisson/connection/MasterSlaveConnectionManager.java b/redisson/src/main/java/org/redisson/connection/MasterSlaveConnectionManager.java
index 1bfdffadb..10bfdebce 100644
--- a/redisson/src/main/java/org/redisson/connection/MasterSlaveConnectionManager.java
+++ b/redisson/src/main/java/org/redisson/connection/MasterSlaveConnectionManager.java
@@ -45,6 +45,8 @@ import org.redisson.client.RedisException;
import org.redisson.client.RedisNodeNotFoundException;
import org.redisson.client.RedisPubSubConnection;
import org.redisson.client.RedisPubSubListener;
+import org.redisson.client.RedisTimeoutException;
+import org.redisson.client.SubscribeListener;
import org.redisson.client.codec.Codec;
import org.redisson.client.protocol.RedisCommand;
import org.redisson.client.protocol.pubsub.PubSubType;
@@ -323,7 +325,7 @@ public class MasterSlaveConnectionManager implements ConnectionManager {
}
protected void initTimer(MasterSlaveServersConfig config) {
- int[] timeouts = new int[]{config.getRetryInterval(), config.getTimeout(), config.getReconnectionTimeout()};
+ int[] timeouts = new int[]{config.getRetryInterval(), config.getTimeout()};
Arrays.sort(timeouts);
int minTimeout = timeouts[0];
if (minTimeout % 100 != 0) {
@@ -366,16 +368,20 @@ public class MasterSlaveConnectionManager implements ConnectionManager {
addEntry(slot, entry);
}
- if (config.getDnsMonitoringInterval() != -1) {
- dnsMonitor = new DNSMonitor(this, f.getNow(),
- config.getSlaveAddresses(), config.getDnsMonitoringInterval(), resolverGroup);
- dnsMonitor.start();
- }
+ startDNSMonitoring(f.getNow());
} catch (RuntimeException e) {
stopThreads();
throw e;
}
}
+
+ protected void startDNSMonitoring(RedisClient masterHost) {
+ if (config.getDnsMonitoringInterval() != -1) {
+ dnsMonitor = new DNSMonitor(this, masterHost,
+ config.getSlaveAddresses(), config.getDnsMonitoringInterval(), resolverGroup);
+ dnsMonitor.start();
+ }
+ }
protected MasterSlaveEntry createMasterSlaveEntry(MasterSlaveServersConfig config,
HashSet slots) {
@@ -411,8 +417,8 @@ public class MasterSlaveConnectionManager implements ConnectionManager {
c.setConnectTimeout(cfg.getConnectTimeout());
c.setIdleConnectionTimeout(cfg.getIdleConnectionTimeout());
- c.setFailedAttempts(cfg.getFailedAttempts());
- c.setReconnectionTimeout(cfg.getReconnectionTimeout());
+ c.setFailedSlaveCheckInterval(cfg.getFailedSlaveCheckInterval());
+ c.setFailedSlaveReconnectionInterval(cfg.getFailedSlaveReconnectionInterval());
c.setMasterConnectionMinimumIdleSize(cfg.getMasterConnectionMinimumIdleSize());
c.setSlaveConnectionMinimumIdleSize(cfg.getSlaveConnectionMinimumIdleSize());
c.setSubscriptionConnectionMinimumIdleSize(cfg.getSubscriptionConnectionMinimumIdleSize());
@@ -500,7 +506,7 @@ public class MasterSlaveConnectionManager implements ConnectionManager {
@Override
public RFuture psubscribe(String channelName, Codec codec, RedisPubSubListener... listeners) {
- return subscribe(PubSubType.PSUBSCRIBE, codec, channelName, listeners);
+ return subscribe(PubSubType.PSUBSCRIBE, codec, channelName, new RedissonPromise(), listeners);
}
@Override
@@ -512,25 +518,36 @@ public class MasterSlaveConnectionManager implements ConnectionManager {
@Override
public RFuture subscribe(Codec codec, String channelName, RedisPubSubListener... listeners) {
- return subscribe(PubSubType.SUBSCRIBE, codec, channelName, listeners);
+ return subscribe(PubSubType.SUBSCRIBE, codec, channelName, new RedissonPromise(), listeners);
}
private RFuture subscribe(final PubSubType type, final Codec codec, final String channelName,
- final RedisPubSubListener... listeners) {
+ final RPromise promise, final RedisPubSubListener... listeners) {
final AsyncSemaphore lock = getSemaphore(channelName);
- final RPromise result = new RedissonPromise();
lock.acquire(new Runnable() {
@Override
public void run() {
- if (result.isDone()) {
+ if (promise.isDone()) {
lock.release();
return;
}
+ final RPromise result = new RedissonPromise();
+ result.addListener(new FutureListener() {
+ @Override
+ public void operationComplete(Future future) throws Exception {
+ if (!future.isSuccess()) {
+ subscribe(type, codec, channelName, promise, listeners);
+ return;
+ }
+
+ promise.trySuccess(result.getNow());
+ }
+ });
subscribe(codec, channelName, result, type, lock, listeners);
}
});
- return result;
+ return promise;
}
public RFuture subscribe(Codec codec, String channelName, AsyncSemaphore semaphore, RedisPubSubListener... listeners) {
@@ -604,7 +621,10 @@ public class MasterSlaveConnectionManager implements ConnectionManager {
for (RedisPubSubListener listener : listeners) {
connEntry.addListener(channelName, listener);
}
- connEntry.getSubscribeFuture(channelName, type).addListener(new FutureListener() {
+ SubscribeListener listener = connEntry.getSubscribeFuture(channelName, type);
+ final Future subscribeFuture = listener.getSuccessFuture();
+
+ subscribeFuture.addListener(new FutureListener() {
@Override
public void operationComplete(Future future) throws Exception {
if (!promise.trySuccess(connEntry)) {
@@ -621,6 +641,15 @@ public class MasterSlaveConnectionManager implements ConnectionManager {
}
}
});
+
+ newTimeout(new TimerTask() {
+ @Override
+ public void run(Timeout timeout) throws Exception {
+ if (promise.tryFailure(new RedisTimeoutException())) {
+ subscribeFuture.cancel(false);
+ }
+ }
+ }, config.getRetryInterval(), TimeUnit.MILLISECONDS);
}
private void connect(final Codec codec, final String channelName,
@@ -695,34 +724,54 @@ public class MasterSlaveConnectionManager implements ConnectionManager {
}
@Override
- public RFuture unsubscribe(final String channelName, boolean temporaryDown) {
- final PubSubConnectionEntry entry = name2PubSubConnection.remove(channelName);
- if (entry == null) {
- return null;
- }
- freePubSubConnections.remove(entry);
-
- final Codec entryCodec = entry.getConnection().getChannels().get(channelName);
- if (temporaryDown) {
- final RPromise result = new RedissonPromise();
- entry.unsubscribe(channelName, new BaseRedisPubSubListener() {
-
- @Override
- public boolean onStatus(PubSubType type, String channel) {
- if (type == PubSubType.UNSUBSCRIBE && channel.equals(channelName)) {
- result.trySuccess(entryCodec);
- return true;
- }
- return false;
+ public RFuture unsubscribe(final String channelName, final PubSubType topicType) {
+ final RPromise result = new RedissonPromise();
+ final AsyncSemaphore lock = getSemaphore(channelName);
+ lock.acquire(new Runnable() {
+ @Override
+ public void run() {
+ final PubSubConnectionEntry entry = name2PubSubConnection.remove(channelName);
+ if (entry == null) {
+ lock.release();
+ result.trySuccess(null);
+ return;
}
-
- });
- return result;
- }
- entry.unsubscribe(channelName, null);
- return RedissonPromise.newSucceededFuture(entryCodec);
+
+ freePubSubLock.acquire(new Runnable() {
+ @Override
+ public void run() {
+ freePubSubConnections.remove(entry);
+ freePubSubLock.release();
+
+ final Codec entryCodec = entry.getConnection().getChannels().get(channelName);
+ RedisPubSubListener