github-forks
/
redisson
mirror of https://github.com/redisson/redisson.git
1
0
Fork 0
Code Issues Projects Releases Wiki Activity

Merge branch 'master' into 3.0.0

Browse Source
pull/1303/head
Nikita 7 years ago
parent 373807b2b1 09b2724c44
commit 70af89bff3
9 changed files with 221 additions and 113 deletions
Show Stats Download Patch File Download Diff File

27
redisson/src/main/java/org/redisson/client/RedisClient.java
Unescape Escape View File

@ -57,10 +57,10 @@ import io.netty.util.concurrent.FutureListener;
*/
public class RedisClient {
private final AtomicReference<RFuture<InetSocketAddress>> resolveFuture = new AtomicReference<RFuture<InetSocketAddress>>();
private final AtomicReference<RFuture<InetSocketAddress>> resolvedAddrFuture = new AtomicReference<RFuture<InetSocketAddress>>();
private final Bootstrap bootstrap;
private final Bootstrap pubSubBootstrap;
private final URI addr;
private final URI uri;
private InetSocketAddress resolvedAddr;
private final ChannelGroup channels;
@ -105,7 +105,12 @@ public class RedisClient {
this.executor = copy.getExecutor();
this.timer = copy.getTimer();
addr = copy.getAddress();
uri = copy.getAddress();
resolvedAddr = copy.getAddr();
if (resolvedAddr != null) {
resolvedAddrFuture.set(RedissonPromise.newSucceededFuture(resolvedAddr));
}
channels = new DefaultChannelGroup(copy.getGroup().next());
bootstrap = createBootstrap(copy, Type.PLAIN);
@ -147,18 +152,22 @@ public class RedisClient {
try {
return connectAsync().syncUninterruptibly().getNow();
} catch (Exception e) {
throw new RedisConnectionException("Unable to connect to: " + addr, e);
throw new RedisConnectionException("Unable to connect to: " + uri, e);
}
}
public RFuture<InetSocketAddress> resolveAddr() {
if (resolvedAddrFuture.get() != null) {
return resolvedAddrFuture.get();
}
final RPromise<InetSocketAddress> promise = new RedissonPromise<InetSocketAddress>();
if (!resolveFuture.compareAndSet(null, promise)) {
return resolveFuture.get();
if (!resolvedAddrFuture.compareAndSet(null, promise)) {
return resolvedAddrFuture.get();
}
AddressResolver<InetSocketAddress> resolver = (AddressResolver<InetSocketAddress>) bootstrap.config().resolver().getResolver(bootstrap.config().group().next());
Future<InetSocketAddress> resolveFuture = resolver.resolve(InetSocketAddress.createUnresolved(addr.getHost(), addr.getPort()));
Future<InetSocketAddress> resolveFuture = resolver.resolve(InetSocketAddress.createUnresolved(uri.getHost(), uri.getPort()));
resolveFuture.addListener(new FutureListener<InetSocketAddress>() {
@Override
public void operationComplete(Future<InetSocketAddress> future) throws Exception {
@ -229,7 +238,7 @@ public class RedisClient {
try {
return connectPubSubAsync().syncUninterruptibly().getNow();
} catch (Exception e) {
throw new RedisConnectionException("Unable to connect to: " + addr, e);
throw new RedisConnectionException("Unable to connect to: " + uri, e);
}
}
@ -342,7 +351,7 @@ public class RedisClient {
@Override
public String toString() {
return "[addr=" + addr + "]";
return "[addr=" + uri + "]";
}
}

8
redisson/src/main/java/org/redisson/client/RedisClientConfig.java
Unescape Escape View File

@ -15,7 +15,7 @@
*/
package org.redisson.client;
import java.net.InetAddress;
import java.net.InetSocketAddress;
import java.net.URI;
import java.util.concurrent.ExecutorService;
@ -36,7 +36,7 @@ import io.netty.util.Timer;
public class RedisClientConfig {
private URI address;
private InetAddress addr;
private InetSocketAddress addr;
private Timer timer;
private ExecutorService executor;
@ -99,7 +99,7 @@ public class RedisClientConfig {
this.address = URIBuilder.create(address);
return this;
}
public RedisClientConfig setAddress(InetAddress addr, URI address) {
public RedisClientConfig setAddress(InetSocketAddress addr, URI address) {
this.addr = addr;
this.address = address;
return this;
@ -111,7 +111,7 @@ public class RedisClientConfig {
public URI getAddress() {
return address;
}
public InetAddress getAddr() {
public InetSocketAddress getAddr() {
return addr;
}

4
redisson/src/main/java/org/redisson/connection/ConnectionManager.java
Unescape Escape View File

@ -88,6 +88,8 @@ public interface ConnectionManager {
MasterSlaveEntry getEntry(int slot);
MasterSlaveEntry getEntry(InetSocketAddress address);
<R> RPromise<R> newPromise();
void releaseRead(NodeSource source, RedisConnection connection);
@ -100,6 +102,8 @@ public interface ConnectionManager {
RedisClient createClient(NodeType type, URI address, int timeout, int commandTimeout);
RedisClient createClient(NodeType type, InetSocketAddress address, URI uri);
RedisClient createClient(NodeType type, URI address);
MasterSlaveEntry getEntry(RedisClient redisClient);

68
redisson/src/main/java/org/redisson/connection/DNSMonitor.java
Unescape Escape View File

@ -15,7 +15,6 @@
*/
package org.redisson.connection;
import java.net.InetAddress;
import java.net.InetSocketAddress;
import java.net.URI;
import java.util.Collection;
@ -26,8 +25,8 @@ import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;
import org.redisson.api.RFuture;
import org.redisson.client.RedisClient;
import org.redisson.connection.ClientConnectionsEntry.FreezeReason;
import org.redisson.misc.URIBuilder;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@ -49,22 +48,22 @@ public class DNSMonitor {
private final AddressResolver<InetSocketAddress> resolver;
private final ConnectionManager connectionManager;
private final Map<URI, InetAddress> masters = new HashMap<URI, InetAddress>();
private final Map<URI, InetAddress> slaves = new HashMap<URI, InetAddress>();
private final Map<URI, InetSocketAddress> masters = new HashMap<URI, InetSocketAddress>();
private final Map<URI, InetSocketAddress> slaves = new HashMap<URI, InetSocketAddress>();
private ScheduledFuture<?> dnsMonitorFuture;
private long dnsMonitoringInterval;
public DNSMonitor(ConnectionManager connectionManager, InetSocketAddress masterHost, Collection<URI> slaveHosts, long dnsMonitoringInterval, DnsAddressResolverGroup resolverGroup) {
public DNSMonitor(ConnectionManager connectionManager, RedisClient masterHost, Collection<URI> slaveHosts, long dnsMonitoringInterval, DnsAddressResolverGroup resolverGroup) {
this.resolver = resolverGroup.getResolver(connectionManager.getGroup().next());
URI uri = URIBuilder.create("redis://" + masterHost.getAddress().getHostAddress() + ":" + masterHost.getPort());
masters.put(uri, masterHost.getAddress());
masterHost.resolveAddr().syncUninterruptibly();
masters.put(masterHost.getConfig().getAddress(), masterHost.getAddr());
for (URI host : slaveHosts) {
Future<InetSocketAddress> resolveFuture = resolver.resolve(InetSocketAddress.createUnresolved(host.getHost(), 0));
Future<InetSocketAddress> resolveFuture = resolver.resolve(InetSocketAddress.createUnresolved(host.getHost(), host.getPort()));
resolveFuture.syncUninterruptibly();
slaves.put(host, resolveFuture.getNow().getAddress());
slaves.put(host, resolveFuture.getNow());
}
this.connectionManager = connectionManager;
this.dnsMonitoringInterval = dnsMonitoringInterval;
@ -86,8 +85,8 @@ public class DNSMonitor {
@Override
public void run() {
final AtomicInteger counter = new AtomicInteger(masters.size() + slaves.size());
for (final Entry<URI, InetAddress> entry : masters.entrySet()) {
Future<InetSocketAddress> resolveFuture = resolver.resolve(InetSocketAddress.createUnresolved(entry.getKey().getHost(), 0));
for (final Entry<URI, InetSocketAddress> entry : masters.entrySet()) {
Future<InetSocketAddress> resolveFuture = resolver.resolve(InetSocketAddress.createUnresolved(entry.getKey().getHost(), entry.getKey().getPort()));
resolveFuture.addListener(new FutureListener<InetSocketAddress>() {
@Override
public void operationComplete(Future<InetSocketAddress> future) throws Exception {
@ -100,25 +99,25 @@ public class DNSMonitor {
return;
}
InetAddress master = entry.getValue();
InetAddress now = future.get().getAddress();
if (!now.getHostAddress().equals(master.getHostAddress())) {
log.info("Detected DNS change. Master {} has changed ip from {} to {}", entry.getKey(), master.getHostAddress(), now.getHostAddress());
for (MasterSlaveEntry entrySet : connectionManager.getEntrySet()) {
if (entrySet.getClient().getAddr().getHostName().equals(entry.getKey().getHost())
&& entrySet.getClient().getAddr().getPort() == entry.getKey().getPort()) {
entrySet.changeMaster(entry.getKey());
break;
}
InetSocketAddress currentMasterAddr = entry.getValue();
InetSocketAddress newMasterAddr = future.getNow();
if (!newMasterAddr.getAddress().equals(currentMasterAddr.getAddress())) {
log.info("Detected DNS change. Master {} has changed ip from {} to {}",
entry.getKey(), currentMasterAddr.getAddress().getHostAddress(), newMasterAddr.getAddress().getHostAddress());
MasterSlaveEntry masterSlaveEntry = connectionManager.getEntry(currentMasterAddr);
if (masterSlaveEntry == null) {
log.error("Unable to find master entry for {}", currentMasterAddr);
return;
}
masters.put(entry.getKey(), now);
masterSlaveEntry.changeMaster(newMasterAddr, entry.getKey());
masters.put(entry.getKey(), newMasterAddr);
}
}
});
}
for (final Entry<URI, InetAddress> entry : slaves.entrySet()) {
Future<InetSocketAddress> resolveFuture = resolver.resolve(InetSocketAddress.createUnresolved(entry.getKey().getHost(), 0));
for (final Entry<URI, InetSocketAddress> entry : slaves.entrySet()) {
Future<InetSocketAddress> resolveFuture = resolver.resolve(InetSocketAddress.createUnresolved(entry.getKey().getHost(), entry.getKey().getPort()));
resolveFuture.addListener(new FutureListener<InetSocketAddress>() {
@Override
public void operationComplete(Future<InetSocketAddress> future) throws Exception {
@ -131,30 +130,29 @@ public class DNSMonitor {
return;
}
InetAddress slave = entry.getValue();
final InetAddress updatedSlave = future.get().getAddress();
if (!updatedSlave.getHostAddress().equals(slave.getHostAddress())) {
log.info("Detected DNS change. Slave {} has changed ip from {} to {}", entry.getKey().getHost(), slave.getHostAddress(), updatedSlave.getHostAddress());
final InetSocketAddress currentSlaveAddr = entry.getValue();
final InetSocketAddress newSlaveAddr = future.getNow();
if (!newSlaveAddr.getAddress().equals(currentSlaveAddr.getAddress())) {
log.info("Detected DNS change. Slave {} has changed ip from {} to {}",
entry.getKey().getHost(), currentSlaveAddr.getAddress().getHostAddress(), newSlaveAddr.getAddress().getHostAddress());
for (final MasterSlaveEntry masterSlaveEntry : connectionManager.getEntrySet()) {
final URI uri = URIBuilder.create(slave.getHostAddress() + ":" + entry.getKey().getPort());
if (masterSlaveEntry.hasSlave(uri)) {
RFuture<Void> addFuture = masterSlaveEntry.addSlave(entry.getKey());
if (masterSlaveEntry.hasSlave(currentSlaveAddr)) {
RFuture<Void> addFuture = masterSlaveEntry.addSlave(newSlaveAddr, entry.getKey());
addFuture.addListener(new FutureListener<Void>() {
@Override
public void operationComplete(Future<Void> future) throws Exception {
if (!future.isSuccess()) {
log.error("Can't add slave: " + updatedSlave, future.cause());
log.error("Can't add slave: " + newSlaveAddr, future.cause());
return;
}
masterSlaveEntry.slaveDown(uri, FreezeReason.MANAGER);
masterSlaveEntry.slaveDown(currentSlaveAddr, FreezeReason.MANAGER);
}
});
break;
}
}
slaves.put(entry.getKey(), updatedSlave);
slaves.put(entry.getKey(), newSlaveAddr);
}
}
});

89
redisson/src/main/java/org/redisson/connection/MasterSlaveConnectionManager.java
Unescape Escape View File

@ -286,9 +286,8 @@ public class MasterSlaveConnectionManager implements ConnectionManager {
addEntry(slot, entry);
}
InetSocketAddress masterHost = f.getNow().resolveAddr().syncUninterruptibly().getNow();
if (config.getDnsMonitoringInterval() != -1) {
dnsMonitor = new DNSMonitor(this, masterHost,
dnsMonitor = new DNSMonitor(this, f.getNow(),
config.getSlaveAddresses(), config.getDnsMonitoringInterval(), resolverGroup);
dnsMonitor.start();
}
@ -349,6 +348,13 @@ public class MasterSlaveConnectionManager implements ConnectionManager {
clientEntries.put(client, new RedisClientEntry(client, commandExecutor, type));
return client;
}
@Override
public RedisClient createClient(NodeType type, InetSocketAddress address, URI uri) {
RedisClient client = createClient(type, address, uri, config.getConnectTimeout(), config.getRetryInterval() * config.getRetryAttempts());
clientEntries.put(client, new RedisClientEntry(client, commandExecutor, type));
return client;
}
@Override
public void shutdownAsync(RedisClient client) {
@ -363,6 +369,13 @@ public class MasterSlaveConnectionManager implements ConnectionManager {
RedisClientConfig redisConfig = createRedisConfig(type, address, timeout, commandTimeout);
return RedisClient.create(redisConfig);
}
private RedisClient createClient(NodeType type, InetSocketAddress address, URI uri, int timeout, int commandTimeout) {
RedisClientConfig redisConfig = createRedisConfig(type, null, timeout, commandTimeout);
redisConfig.setAddress(address, uri);
return RedisClient.create(redisConfig);
}
protected RedisClientConfig createRedisConfig(NodeType type, URI address, int timeout, int commandTimeout) {
RedisClientConfig redisConfig = new RedisClientConfig();
@ -677,7 +690,17 @@ public class MasterSlaveConnectionManager implements ConnectionManager {
return newSucceededFuture(entryCodec);
}
public MasterSlaveEntry getEntry(URI addr) {
public MasterSlaveEntry getEntry(InetSocketAddress address) {
for (MasterSlaveEntry entry : client2entry.values()) {
InetSocketAddress addr = entry.getClient().getAddr();
if (addr.getAddress().equals(address.getAddress()) && addr.getPort() == address.getPort()) {
return entry;
}
}
return null;
}
private MasterSlaveEntry getEntry(URI addr) {
for (MasterSlaveEntry entry : client2entry.values()) {
if (URIBuilder.compare(entry.getClient().getAddr(), addr)) {
return entry;
@ -732,28 +755,19 @@ public class MasterSlaveConnectionManager implements ConnectionManager {
@Override
public RFuture<RedisConnection> connectionWriteOp(NodeSource source, RedisCommand<?> command) {
MasterSlaveEntry entry = source.getEntry();
if (entry == null) {
entry = getEntry(source);
}
MasterSlaveEntry entry = getEntry(source);
if (entry == null) {
RedisNodeNotFoundException ex = new RedisNodeNotFoundException("Node: " + source.getAddr() + " for slot: " + source.getSlot() + " hasn't been discovered yet");
RedisNodeNotFoundException ex = new RedisNodeNotFoundException("Node: " + source + " hasn't been discovered yet");
return RedissonPromise.newFailedFuture(ex);
}
return entry.connectionWriteOp(command);
}
private MasterSlaveEntry getEntry(NodeSource source) {
// slots handling during migration state
if (source.getRedirect() != null) {
return getEntry(source.getAddr());
}
return getEntry(source.getSlot());
}
@Override
public RFuture<RedisConnection> connectionReadOp(NodeSource source, RedisCommand<?> command) {
MasterSlaveEntry entry = source.getEntry();
if (entry == null && source.getSlot() != null) {
entry = getEntry(source.getSlot());
@ -761,30 +775,21 @@ public class MasterSlaveConnectionManager implements ConnectionManager {
if (source.getRedisClient() != null) {
entry = getEntry(source.getRedisClient());
}
if (source.getAddr() != null) {
entry = getEntry(source.getAddr());
if (entry == null) {
for (MasterSlaveEntry e : getEntrySet()) {
if (e.hasSlave(source.getAddr())) {
entry = e;
break;
}
}
}
if (entry == null) {
RedisNodeNotFoundException ex = new RedisNodeNotFoundException("Node: " + source.getAddr() + " for slot: " + source.getSlot() + " hasn't been discovered yet");
return RedissonPromise.newFailedFuture(ex);
}
return entry.connectionReadOp(command, source.getAddr());
}
return entry;
}
@Override
public RFuture<RedisConnection> connectionReadOp(NodeSource source, RedisCommand<?> command) {
MasterSlaveEntry entry = getEntry(source);
if (entry == null) {
RedisNodeNotFoundException ex = new RedisNodeNotFoundException("Node: " + source.getAddr() + " for slot: " + source.getSlot() + " hasn't been discovered yet");
RedisNodeNotFoundException ex = new RedisNodeNotFoundException("Node: " + source + " hasn't been discovered yet");
return RedissonPromise.newFailedFuture(ex);
}
if (source.getRedirect() != null) {
return entry.connectionReadOp(command, source.getAddr());
}
return entry.connectionReadOp(command);
}
@ -800,7 +805,7 @@ public class MasterSlaveConnectionManager implements ConnectionManager {
protected void releaseSubscribeConnection(int slot, PubSubConnectionEntry pubSubEntry) {
MasterSlaveEntry entry = getEntry(slot);
if (entry == null) {
log.error("Node for slot: " + slot + " hasn't been discovered yet");
log.error("Node for slot: " + slot + " can't be found");
} else {
entry.returnPubSubConnection(pubSubEntry);
}
@ -808,12 +813,9 @@ public class MasterSlaveConnectionManager implements ConnectionManager {
@Override
public void releaseWrite(NodeSource source, RedisConnection connection) {
MasterSlaveEntry entry = source.getEntry();
MasterSlaveEntry entry = getEntry(source);
if (entry == null) {
entry = getEntry(source);
}
if (entry == null) {
log.error("Node: " + source.getAddr() + " for slot: " + source.getSlot() + " hasn't been discovered yet");
log.error("Node: " + source + " can't be found");
} else {
entry.releaseWrite(connection);
}
@ -821,12 +823,9 @@ public class MasterSlaveConnectionManager implements ConnectionManager {
@Override
public void releaseRead(NodeSource source, RedisConnection connection) {
MasterSlaveEntry entry = source.getEntry();
if (entry == null) {
entry = getEntry(source);
}
MasterSlaveEntry entry = getEntry(source);
if (entry == null) {
log.error("Node: " + source.getAddr() + " for slot: " + source.getSlot() + " hasn't been discovered yet");
log.error("Node: " + source + " can't be found");
} else {
entry.releaseRead(connection);
}

92
redisson/src/main/java/org/redisson/connection/MasterSlaveEntry.java
Unescape Escape View File

@ -108,11 +108,20 @@ public class MasterSlaveEntry {
}
return result;
}
public RFuture<RedisClient> setupMasterEntry(InetSocketAddress address, URI uri) {
RedisClient client = connectionManager.createClient(NodeType.MASTER, address, uri);
return setupMasterEntry(client);
}
public RPromise<RedisClient> setupMasterEntry(URI address) {
final RPromise<RedisClient> result = new RedissonPromise<RedisClient>();
public RFuture<RedisClient> setupMasterEntry(URI address) {
RedisClient client = connectionManager.createClient(NodeType.MASTER, address);
return setupMasterEntry(client);
}
final RedisClient client = connectionManager.createClient(NodeType.MASTER, address);
private RFuture<RedisClient> setupMasterEntry(final RedisClient client) {
final RPromise<RedisClient> result = new RedissonPromise<RedisClient>();
RFuture<InetSocketAddress> addrFuture = client.resolveAddr();
addrFuture.addListener(new FutureListener<InetSocketAddress>() {
@ -148,7 +157,16 @@ public class MasterSlaveEntry {
return result;
}
public boolean slaveDown(URI address, FreezeReason freezeReason) {
public boolean slaveDown(ClientConnectionsEntry entry, FreezeReason freezeReason) {
ClientConnectionsEntry e = slaveBalancer.freeze(entry, freezeReason);
if (e == null) {
return false;
}
return slaveDown(entry, freezeReason == FreezeReason.SYSTEM);
}
public boolean slaveDown(InetSocketAddress address, FreezeReason freezeReason) {
ClientConnectionsEntry entry = slaveBalancer.freeze(address, freezeReason);
if (entry == null) {
return false;
@ -157,8 +175,8 @@ public class MasterSlaveEntry {
return slaveDown(entry, freezeReason == FreezeReason.SYSTEM);
}
public boolean slaveDown(RedisClient redisClient, FreezeReason freezeReason) {
ClientConnectionsEntry entry = slaveBalancer.freeze(redisClient, freezeReason);
public boolean slaveDown(URI address, FreezeReason freezeReason) {
ClientConnectionsEntry entry = slaveBalancer.freeze(address, freezeReason);
if (entry == null) {
return false;
}
@ -169,9 +187,8 @@ public class MasterSlaveEntry {
private boolean slaveDown(ClientConnectionsEntry entry, boolean temporaryDown) {
// add master as slave if no more slaves available
if (!config.checkSkipSlavesInit() && slaveBalancer.getAvailableClients() == 0) {
URI addr = masterEntry.getClient().getConfig().getAddress();
if (slaveUp(addr, FreezeReason.SYSTEM)) {
log.info("master {} used as slave", addr);
if (slaveBalancer.unfreeze(masterEntry.getClient().getAddr(), FreezeReason.SYSTEM)) {
log.info("master {} used as slave", masterEntry.getClient().getAddr());
}
}
@ -344,6 +361,10 @@ public class MasterSlaveEntry {
return slaveBalancer.contains(redisClient);
}
public boolean hasSlave(InetSocketAddress addr) {
return slaveBalancer.contains(addr);
}
public boolean hasSlave(URI addr) {
return slaveBalancer.contains(addr);
}
@ -352,9 +373,11 @@ public class MasterSlaveEntry {
return addSlave(address, false, NodeType.SLAVE);
}
private RFuture<Void> addSlave(URI address, final boolean freezed, final NodeType nodeType) {
final RedisClient client = connectionManager.createClient(NodeType.SLAVE, address);
public RFuture<Void> addSlave(InetSocketAddress address, URI uri) {
return addSlave(address, uri, false, NodeType.SLAVE);
}
private RFuture<Void> addSlave(final RedisClient client, final boolean freezed, final NodeType nodeType) {
final RPromise<Void> result = new RedissonPromise<Void>();
RFuture<InetSocketAddress> addrFuture = client.resolveAddr();
addrFuture.addListener(new FutureListener<InetSocketAddress>() {
@ -364,7 +387,7 @@ public class MasterSlaveEntry {
result.tryFailure(future.cause());
return;
}
ClientConnectionsEntry entry = new ClientConnectionsEntry(client,
config.getSlaveConnectionMinimumIdleSize(),
config.getSlaveConnectionPoolSize(),
@ -382,11 +405,36 @@ public class MasterSlaveEntry {
});
return result;
}
private RFuture<Void> addSlave(InetSocketAddress address, URI uri, final boolean freezed, final NodeType nodeType) {
RedisClient client = connectionManager.createClient(NodeType.SLAVE, address, uri);
return addSlave(client, freezed, nodeType);
}
private RFuture<Void> addSlave(URI address, final boolean freezed, final NodeType nodeType) {
RedisClient client = connectionManager.createClient(NodeType.SLAVE, address);
return addSlave(client, freezed, nodeType);
}
public RedisClient getClient() {
return masterEntry.getClient();
}
public boolean slaveUp(ClientConnectionsEntry entry, FreezeReason freezeReason) {
if (!slaveBalancer.unfreeze(entry, freezeReason)) {
return false;
}
InetSocketAddress addr = masterEntry.getClient().getAddr();
// exclude master from slaves
if (!config.checkSkipSlavesInit()
&& !addr.equals(entry.getClient().getAddr())) {
slaveDown(masterEntry.getClient().getAddr(), FreezeReason.SYSTEM);
log.info("master {} excluded from slaves", addr);
}
return true;
}
public boolean slaveUp(URI address, FreezeReason freezeReason) {
if (!slaveBalancer.unfreeze(address, freezeReason)) {
return false;
@ -396,7 +444,7 @@ public class MasterSlaveEntry {
// exclude master from slaves
if (!config.checkSkipSlavesInit()
&& !URIBuilder.compare(addr, address)) {
slaveDown(masterEntry.getClient().getConfig().getAddress(), FreezeReason.SYSTEM);
slaveDown(masterEntry.getClient().getAddr(), FreezeReason.SYSTEM);
log.info("master {} excluded from slaves", addr);
}
return true;
@ -409,9 +457,21 @@ public class MasterSlaveEntry {
*
* @param address of Redis
*/
public void changeMaster(final URI address) {
public void changeMaster(URI address) {
final ClientConnectionsEntry oldMaster = masterEntry;
RFuture<RedisClient> future = setupMasterEntry(address);
changeMaster(address, oldMaster, future);
}
public void changeMaster(InetSocketAddress address, URI uri) {
final ClientConnectionsEntry oldMaster = masterEntry;
RFuture<RedisClient> future = setupMasterEntry(address, uri);
changeMaster(uri, oldMaster, future);
}
private void changeMaster(final URI address, final ClientConnectionsEntry oldMaster,
RFuture<RedisClient> future) {
future.addListener(new FutureListener<RedisClient>() {
@Override
public void operationComplete(Future<RedisClient> future) throws Exception {
@ -434,7 +494,7 @@ public class MasterSlaveEntry {
// more than one slave available, so master can be removed from slaves
if (!config.checkSkipSlavesInit()
&& slaveBalancer.getAvailableClients() > 1) {
slaveDown(newMasterClient, FreezeReason.SYSTEM);
slaveDown(newMasterClient.getAddr(), FreezeReason.SYSTEM);
}
connectionManager.shutdownAsync(oldMaster.getClient());
log.info("master {} has changed to {}", oldMaster.getClient().getAddr(), masterEntry.getClient().getAddr());

5
redisson/src/main/java/org/redisson/connection/NodeSource.java
Unescape Escape View File

@ -80,7 +80,10 @@ public class NodeSource {
@Override
public String toString() {
return "NodeSource [slot=" + slot + ", addr=" + addr + ", redirect=" + redirect + "]";
return "NodeSource [slot=" + slot + ", addr=" + addr + ", redisClient=" + redisClient + ", redirect=" + redirect
+ ", entry=" + entry + "]";
}
}

37
redisson/src/main/java/org/redisson/connection/balancer/LoadBalancerManager.java
Unescape Escape View File

@ -121,6 +121,20 @@ public class LoadBalancerManager {
throw new IllegalStateException("Can't find " + address + " in slaves!");
}
return unfreeze(entry, freezeReason);
}
public boolean unfreeze(InetSocketAddress address, FreezeReason freezeReason) {
ClientConnectionsEntry entry = getEntry(address);
if (entry == null) {
throw new IllegalStateException("Can't find " + address + " in slaves!");
}
return unfreeze(entry, freezeReason);
}
public boolean unfreeze(ClientConnectionsEntry entry, FreezeReason freezeReason) {
synchronized (entry) {
if (!entry.isFreezed()) {
return false;
@ -136,12 +150,18 @@ public class LoadBalancerManager {
}
return false;
}
public ClientConnectionsEntry freeze(URI address, FreezeReason freezeReason) {
ClientConnectionsEntry connectionEntry = getEntry(address);
return freeze(connectionEntry, freezeReason);
}
public ClientConnectionsEntry freeze(InetSocketAddress address, FreezeReason freezeReason) {
ClientConnectionsEntry connectionEntry = getEntry(address);
return freeze(connectionEntry, freezeReason);
}
public ClientConnectionsEntry freeze(RedisClient redisClient, FreezeReason freezeReason) {
ClientConnectionsEntry connectionEntry = getEntry(redisClient);
return freeze(connectionEntry, freezeReason);
@ -172,6 +192,10 @@ public class LoadBalancerManager {
return pubSubConnectionPool.get();
}
public boolean contains(InetSocketAddress addr) {
return getEntry(addr) != null;
}
public boolean contains(URI addr) {
return getEntry(addr) != null;
}
@ -190,6 +214,17 @@ public class LoadBalancerManager {
return null;
}
protected ClientConnectionsEntry getEntry(InetSocketAddress address) {
for (ClientConnectionsEntry entry : client2Entry.values()) {
InetSocketAddress addr = entry.getClient().getAddr();
if (addr.getAddress().equals(address.getAddress()) && addr.getPort() == address.getPort()) {
return entry;
}
}
return null;
}
protected ClientConnectionsEntry getEntry(RedisClient redisClient) {
return client2Entry.get(redisClient);
}

4
redisson/src/main/java/org/redisson/connection/pool/ConnectionPool.java
Unescape Escape View File

@ -325,7 +325,7 @@ abstract class ConnectionPool<T extends RedisConnection> {
private void checkForReconnect(ClientConnectionsEntry entry, Throwable cause) {
if (entry.getNodeType() == NodeType.SLAVE) {
masterSlaveEntry.slaveDown(entry.getClient().getConfig().getAddress(), FreezeReason.RECONNECT);
masterSlaveEntry.slaveDown(entry, FreezeReason.RECONNECT);
log.error("slave " + entry.getClient().getAddr() + " disconnected due to failedAttempts=" + config.getFailedAttempts() + " limit reached", cause);
scheduleCheck(entry);
} else {
@ -392,7 +392,7 @@ abstract class ConnectionPool<T extends RedisConnection> {
public void operationComplete(Future<Void> future)
throws Exception {
if (entry.getNodeType() == NodeType.SLAVE) {
masterSlaveEntry.slaveUp(entry.getClient().getConfig().getAddress(), FreezeReason.RECONNECT);
masterSlaveEntry.slaveUp(entry, FreezeReason.RECONNECT);
log.info("slave {} has been successfully reconnected", entry.getClient().getAddr());
} else {
synchronized (entry) {

Write Preview
Loading…
Cancel
Save