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

new-slave automatic discovery

Browse Source
pull/38/head
Nikita 11 years ago
parent 7d9eb607cf
commit e3da65ad87
4 changed files with 82 additions and 56 deletions
Show Stats Download Patch File Download Diff File

60
src/main/java/org/redisson/connection/BaseLoadBalancer.java
Unescape Escape View File

@ -17,7 +17,8 @@ package org.redisson.connection;
import java.net.InetSocketAddress; import java.net.InetSocketAddress;
import java.util.ArrayList; import java.util.ArrayList;
import java.util.Iterator; import java.util.Collection;
import java.util.Collections;
import java.util.List; import java.util.List;
import java.util.Queue; import java.util.Queue;
import java.util.concurrent.ConcurrentLinkedQueue; import java.util.concurrent.ConcurrentLinkedQueue;
@ -47,64 +48,56 @@ abstract class BaseLoadBalancer implements LoadBalancer {
this.password = password; this.password = password;
} }
public void add(ConnectionEntry entry) { public synchronized void add(ConnectionEntry entry) {
clients.add(entry); clients.add(entry);
clientsEmpty.open(); clientsEmpty.open();
} }
public void unfreeze(String host, int port) { public synchronized void unfreeze(String host, int port) {
InetSocketAddress addr = new InetSocketAddress(host, port); InetSocketAddress addr = new InetSocketAddress(host, port);
for (ConnectionEntry connectionEntry : clients) { for (ConnectionEntry connectionEntry : clients) {
if (!connectionEntry.getClient().getAddr().equals(addr)) { if (!connectionEntry.getClient().getAddr().equals(addr)) {
continue; continue;
} }
connectionEntry.setFreezed(false); connectionEntry.setFreezed(false);
clientsEmpty.open();
return; return;
} }
throw new IllegalStateException("Can't find " + addr + " in slaves!"); throw new IllegalStateException("Can't find " + addr + " in slaves!");
} }
public Queue<RedisPubSubConnection> freeze(String host, int port) { public synchronized Collection<RedisPubSubConnection> freeze(String host, int port) {
InetSocketAddress addr = new InetSocketAddress(host, port); InetSocketAddress addr = new InetSocketAddress(host, port);
for (ConnectionEntry connectionEntry : clients) { for (ConnectionEntry connectionEntry : clients) {
if (!connectionEntry.getClient().getAddr().equals(addr)) { if (connectionEntry.isFreezed()
|| !connectionEntry.getClient().getAddr().equals(addr)) {
continue; continue;
} }
log.debug("{} freezed", addr);
connectionEntry.setFreezed(true); connectionEntry.setFreezed(true);
return connectionEntry.getSubscribeConnections(); // TODO shutdown watchdog
}
throw new IllegalStateException("Can't find " + addr + " in slaves!");
}
public Queue<RedisPubSubConnection> remove(String host, int port) { boolean allFreezed = true;
InetSocketAddress addr = new InetSocketAddress(host, port); for (ConnectionEntry entry : clients) {
for (Iterator<ConnectionEntry> iterator = clients.iterator(); iterator.hasNext();) { if (!entry.isFreezed()) {
ConnectionEntry entry = iterator.next(); allFreezed = false;
if (!entry.getClient().getAddr().equals(addr)) { break;
continue;
} }
}
iterator.remove(); if (allFreezed) {
log.info("slave {} removed", entry.getClient().getAddr());
if (clients.isEmpty()) {
clientsEmpty.close(); clientsEmpty.close();
} }
entry.shutdown(); return connectionEntry.getSubscribeConnections();
log.info("slave {} shutdown", entry.getClient().getAddr());
return entry.getSubscribeConnections();
} }
throw new IllegalStateException("Can't find " + addr + " in slaves!");
return Collections.emptyList();
} }
@SuppressWarnings("unchecked") @SuppressWarnings("unchecked")
public RedisPubSubConnection nextPubSubConnection() { public RedisPubSubConnection nextPubSubConnection() {
List<ConnectionEntry> clientsCopy = new ArrayList<ConnectionEntry>(clients);
if (clientsCopy.isEmpty()) {
clientsEmpty.awaitUninterruptibly(); clientsEmpty.awaitUninterruptibly();
return nextPubSubConnection(); List<ConnectionEntry> clientsCopy = new ArrayList<ConnectionEntry>(clients);
}
while (true) { while (true) {
if (clientsCopy.isEmpty()) { if (clientsCopy.isEmpty()) {
// TODO refactor // TODO refactor
@ -142,11 +135,8 @@ abstract class BaseLoadBalancer implements LoadBalancer {
} }
public RedisConnection nextConnection() { public RedisConnection nextConnection() {
List<ConnectionEntry> clientsCopy = new ArrayList<ConnectionEntry>(clients);
if (clientsCopy.isEmpty()) {
clientsEmpty.awaitUninterruptibly(); clientsEmpty.awaitUninterruptibly();
return nextConnection(); List<ConnectionEntry> clientsCopy = new ArrayList<ConnectionEntry>(clients);
}
while (true) { while (true) {
if (clientsCopy.isEmpty()) { if (clientsCopy.isEmpty()) {
// TODO refactor // TODO refactor
@ -205,4 +195,10 @@ abstract class BaseLoadBalancer implements LoadBalancer {
} }
} }
public void shutdown() {
for (ConnectionEntry entry : clients) {
entry.getClient().shutdown();
}
}
} }

8
src/main/java/org/redisson/connection/LoadBalancer.java
Unescape Escape View File

@ -15,7 +15,7 @@
*/ */
package org.redisson.connection; package org.redisson.connection;
import java.util.Queue; import java.util.Collection;
import com.lambdaworks.redis.RedisConnection; import com.lambdaworks.redis.RedisConnection;
import com.lambdaworks.redis.codec.RedisCodec; import com.lambdaworks.redis.codec.RedisCodec;
@ -23,16 +23,16 @@ import com.lambdaworks.redis.pubsub.RedisPubSubConnection;
public interface LoadBalancer { public interface LoadBalancer {
void shutdown();
void unfreeze(String host, int port); void unfreeze(String host, int port);
Queue<RedisPubSubConnection> freeze(String host, int port); Collection<RedisPubSubConnection> freeze(String host, int port);
void init(RedisCodec codec, String password); void init(RedisCodec codec, String password);
void add(ConnectionEntry entry); void add(ConnectionEntry entry);
Queue<RedisPubSubConnection> remove(String host, int port);
RedisConnection nextConnection(); RedisConnection nextConnection();
RedisPubSubConnection nextPubSubConnection(); RedisPubSubConnection nextPubSubConnection();

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

@ -20,10 +20,8 @@ import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.util.concurrent.FutureListener; import io.netty.util.concurrent.FutureListener;
import java.net.URI; import java.net.URI;
import java.util.ArrayList;
import java.util.Collection; import java.util.Collection;
import java.util.HashSet; import java.util.HashSet;
import java.util.List;
import java.util.Map.Entry; import java.util.Map.Entry;
import java.util.Queue; import java.util.Queue;
import java.util.Set; import java.util.Set;
@ -63,7 +61,6 @@ public class MasterSlaveConnectionManager implements ConnectionManager {
private final ConcurrentMap<String, PubSubConnectionEntry> name2PubSubConnection = new ConcurrentHashMap<String, PubSubConnectionEntry>(); private final ConcurrentMap<String, PubSubConnectionEntry> name2PubSubConnection = new ConcurrentHashMap<String, PubSubConnectionEntry>();
protected LoadBalancer balancer; protected LoadBalancer balancer;
private final List<RedisClient> slaveClients = new ArrayList<RedisClient>();
protected volatile RedisClient masterClient; protected volatile RedisClient masterClient;
private Semaphore masterConnectionsSemaphore; private Semaphore masterConnectionsSemaphore;
@ -89,10 +86,13 @@ public class MasterSlaveConnectionManager implements ConnectionManager {
balancer.init(codec, config.getPassword()); balancer.init(codec, config.getPassword());
for (URI address : this.config.getSlaveAddresses()) { for (URI address : this.config.getSlaveAddresses()) {
RedisClient client = new RedisClient(group, address.getHost(), address.getPort()); RedisClient client = new RedisClient(group, address.getHost(), address.getPort());
slaveClients.add(client); ConnectionEntry entry = new ConnectionEntry(client,
balancer.add(new ConnectionEntry(client,
this.config.getSlaveConnectionPoolSize(), this.config.getSlaveConnectionPoolSize(),
this.config.getSlaveSubscriptionConnectionPoolSize())); this.config.getSlaveSubscriptionConnectionPoolSize());
balancer.add(entry);
}
if (this.config.getSlaveAddresses().size() > 1) {
slaveDown(this.config.getMasterAddress().getHost(), this.config.getMasterAddress().getPort());
} }
masterClient = new RedisClient(group, this.config.getMasterAddress().getHost(), this.config.getMasterAddress().getPort()); masterClient = new RedisClient(group, this.config.getMasterAddress().getHost(), this.config.getMasterAddress().getPort());
@ -105,16 +105,25 @@ public class MasterSlaveConnectionManager implements ConnectionManager {
} }
protected void slaveDown(String host, int port) { protected void slaveDown(String host, int port) {
Queue<RedisPubSubConnection> connections = balancer.freeze(host, port); Collection<RedisPubSubConnection> connections = balancer.freeze(host, port);
reattachListeners(connections); reattachListeners(connections);
} }
protected void addSlave(String host, int port) {
slaveDown(masterClient.getAddr().getHostName(), port);
RedisClient client = new RedisClient(group, host, port);
balancer.add(new ConnectionEntry(client,
this.config.getSlaveConnectionPoolSize(),
this.config.getSlaveSubscriptionConnectionPoolSize()));
}
protected void slaveUp(String host, int port) { protected void slaveUp(String host, int port) {
balancer.unfreeze(host, port); balancer.unfreeze(host, port);
} }
/** /**
* Remove slave with <code>host:port</code> from slaves list. * Freeze slave with <code>host:port</code> from slaves list.
* Re-attach pub/sub listeners from it to other slave. * Re-attach pub/sub listeners from it to other slave.
* Shutdown old master client. * Shutdown old master client.
* *
@ -122,12 +131,11 @@ public class MasterSlaveConnectionManager implements ConnectionManager {
protected void changeMaster(String host, int port) { protected void changeMaster(String host, int port) {
RedisClient oldMaster = masterClient; RedisClient oldMaster = masterClient;
masterClient = new RedisClient(group, host, port); masterClient = new RedisClient(group, host, port);
Queue<RedisPubSubConnection> connections = balancer.remove(host, port); slaveDown(host, port);
reattachListeners(connections);
oldMaster.shutdown(); oldMaster.shutdown();
} }
private void reattachListeners(Queue<RedisPubSubConnection> connections) { private void reattachListeners(Collection<RedisPubSubConnection> connections) {
for (Entry<String, PubSubConnectionEntry> mapEntry : name2PubSubConnection.entrySet()) { for (Entry<String, PubSubConnectionEntry> mapEntry : name2PubSubConnection.entrySet()) {
for (RedisPubSubConnection redisPubSubConnection : connections) { for (RedisPubSubConnection redisPubSubConnection : connections) {
PubSubConnectionEntry entry = mapEntry.getValue(); PubSubConnectionEntry entry = mapEntry.getValue();
@ -310,9 +318,7 @@ public class MasterSlaveConnectionManager implements ConnectionManager {
@Override @Override
public void shutdown() { public void shutdown() {
masterClient.shutdown(); masterClient.shutdown();
for (RedisClient client : slaveClients) { balancer.shutdown();
client.shutdown();
}
group.shutdownGracefully().syncUninterruptibly(); group.shutdownGracefully().syncUninterruptibly();
} }

34
src/main/java/org/redisson/connection/SentinelConnectionManager.java
Unescape Escape View File

@ -58,6 +58,7 @@ public class SentinelConnectionManager extends MasterSlaveConnectionManager {
String masterHost = master.get(0) + ":" + master.get(1); String masterHost = master.get(0) + ":" + master.get(1);
c.setMasterAddress(masterHost); c.setMasterAddress(masterHost);
log.info("master: {}", masterHost); log.info("master: {}", masterHost);
c.addSlaveAddress(masterHost);
// TODO async // TODO async
List<Map<String, String>> slaves = connection.slaves(cfg.getMasterName()).awaitUninterruptibly().getNow(); List<Map<String, String>> slaves = connection.slaves(cfg.getMasterName()).awaitUninterruptibly().getNow();
@ -67,10 +68,6 @@ public class SentinelConnectionManager extends MasterSlaveConnectionManager {
log.info("slave: {}:{}", ip, port); log.info("slave: {}:{}", ip, port);
c.addSlaveAddress(ip + ":" + port); c.addSlaveAddress(ip + ":" + port);
} }
if (slaves.isEmpty()) {
log.info("master added as slave");
c.addSlaveAddress(masterHost);
}
client.shutdown(); client.shutdown();
break; break;
@ -84,6 +81,8 @@ public class SentinelConnectionManager extends MasterSlaveConnectionManager {
private void monitorMasterChange(final SentinelServersConfig cfg) { private void monitorMasterChange(final SentinelServersConfig cfg) {
final AtomicReference<String> master = new AtomicReference<String>(); final AtomicReference<String> master = new AtomicReference<String>();
final Set<String> freezeSlaves = Collections.newSetFromMap(new ConcurrentHashMap<String, Boolean>()); final Set<String> freezeSlaves = Collections.newSetFromMap(new ConcurrentHashMap<String, Boolean>());
final Set<String> addedSlaves = Collections.newSetFromMap(new ConcurrentHashMap<String, Boolean>());
for (final URI addr : cfg.getSentinelAddresses()) { for (final URI addr : cfg.getSentinelAddresses()) {
RedisClient client = new RedisClient(group, addr.getHost(), addr.getPort()); RedisClient client = new RedisClient(group, addr.getHost(), addr.getPort());
sentinels.add(client); sentinels.add(client);
@ -97,6 +96,9 @@ public class SentinelConnectionManager extends MasterSlaveConnectionManager {
@Override @Override
public void message(String channel, String msg) { public void message(String channel, String msg) {
if ("+slave".equals(channel)) {
onSlaveAdded(addedSlaves, addr, msg);
}
if ("+sdown".equals(channel)) { if ("+sdown".equals(channel)) {
onSlaveDown(freezeSlaves, addr, msg); onSlaveDown(freezeSlaves, addr, msg);
} }
@ -109,7 +111,27 @@ public class SentinelConnectionManager extends MasterSlaveConnectionManager {
} }
}); });
pubsub.subscribe("+switch-master", "+sdown", "-sdown"); pubsub.subscribe("+switch-master", "+sdown", "-sdown", "+slave");
}
}
protected void onSlaveAdded(Set<String> addedSlaves, URI addr, String msg) {
String[] parts = msg.split(" ");
if (parts.length > 4
&& "slave".equals(parts[0])) {
String ip = parts[2];
String port = parts[3];
String slaveAddr = ip + ":" + port;
// to avoid addition twice
if (addedSlaves.add(slaveAddr)) {
log.debug("Slave has been added - {}", slaveAddr);
addSlave(ip, Integer.valueOf(port));
}
} else {
log.warn("Invalid message: {} from Sentinel {}:{}", msg, addr.getHost(), addr.getPort());
} }
} }
@ -122,6 +144,8 @@ public class SentinelConnectionManager extends MasterSlaveConnectionManager {
String port = parts[3]; String port = parts[3];
String slaveAddr = ip + ":" + port; String slaveAddr = ip + ":" + port;
// to avoid freeze twice
if (freezeSlaves.add(slaveAddr)) { if (freezeSlaves.add(slaveAddr)) {
log.debug("Slave has down - {}", slaveAddr); log.debug("Slave has down - {}", slaveAddr);
slaveDown(ip, Integer.valueOf(port)); slaveDown(ip, Integer.valueOf(port));

Write Preview
Loading…
Cancel
Save