Implemented read-write separation based on JedisSentineled #4231
+525
−50
There are no files selected for viewing
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
|
|
@@ -3,6 +3,7 @@ | |
| import java.util.ArrayList; | ||
| import java.util.Collection; | ||
| import java.util.List; | ||
| import java.util.Map; | ||
| import java.util.Set; | ||
| import java.util.concurrent.atomic.AtomicBoolean; | ||
| import java.util.concurrent.locks.Lock; | ||
|
|
@@ -26,6 +27,15 @@ | |
| import redis.clients.jedis.util.IOUtils; | ||
|
|
||
| public class SentineledConnectionProvider implements ConnectionProvider { | ||
| class PoolInfo { | ||
| public String host; | ||
| public ConnectionPool pool; | ||
|
|
||
| public PoolInfo(String host, ConnectionPool pool) { | ||
| this.host = host; | ||
| this.pool = pool; | ||
| } | ||
| } | ||
|
|
||
| private static final Logger LOG = LoggerFactory.getLogger(SentineledConnectionProvider.class); | ||
|
|
||
|
|
@@ -51,6 +61,10 @@ public class SentineledConnectionProvider implements ConnectionProvider { | |
|
|
||
| private final Lock initPoolLock = new ReentrantLock(true); | ||
|
|
||
| private final List<PoolInfo> slavePools = new ArrayList<>(); | ||
|
|
||
| private int poolIndex; | ||
|
|
||
| public SentineledConnectionProvider(String masterName, final JedisClientConfig masterClientConfig, | ||
| Set<HostAndPort> sentinels, final JedisClientConfig sentinelClientConfig) { | ||
| this(masterName, masterClientConfig, null, null, sentinels, sentinelClientConfig); | ||
|
|
@@ -102,13 +116,52 @@ public SentineledConnectionProvider(String masterName, final JedisClientConfig m | |
| initMaster(master); | ||
| } | ||
|
|
||
| private Connection getSlaveResource() { | ||
| int startIdx; | ||
| synchronized (slavePools) { | ||
| poolIndex++; | ||
| if (poolIndex >= slavePools.size()) { | ||
| poolIndex = 0; | ||
| } | ||
| startIdx = poolIndex; | ||
| } | ||
| return _getSlaveResource(startIdx, 0); | ||
| } | ||
|
|
||
| private Connection _getSlaveResource(int idx, int cnt) { | ||
| PoolInfo poolInfo; | ||
| synchronized (slavePools) { | ||
| if (cnt >= slavePools.size()) { | ||
| return null; | ||
| } | ||
| poolInfo = slavePools.get(idx % slavePools.size()); | ||
| } | ||
| try { | ||
| Connection jedis = poolInfo.pool.getResource(); | ||
| return jedis; | ||
| } catch (Exception e) { | ||
| LOG.error("get connection fail:", e); | ||
| return _getSlaveResource(idx + 1, cnt + 1); | ||
| } | ||
| } | ||
|
|
||
| @Override | ||
| public Connection getConnection() { | ||
| return pool.getResource(); | ||
| } | ||
|
|
||
| @Override | ||
| public Connection getConnection(CommandArguments args) { | ||
| boolean readCommand = masterClientConfig.isReadCommand(args); | ||
| if (readCommand) { | ||
| Connection slaveConn = getSlaveResource(); | ||
| if (slaveConn != null) { | ||
| return slaveConn; | ||
| } | ||
| if (!masterClientConfig.isFallbackToMaster()) { | ||
| throw new JedisException("can not get Connection, all slave is invalid"); | ||
| } | ||
| } | ||
| return pool.getResource(); | ||
| } | ||
|
Comment on lines
193
to
231
Collaborator
There was a problem hiding this comment. See my comment about preserving backword compatibility |
||
|
|
||
|
|
@@ -117,6 +170,10 @@ public void close() { | |
| sentinelListeners.forEach(SentinelListener::shutdown); | ||
|
|
||
| pool.close(); | ||
|
|
||
| for (PoolInfo slavePool : slavePools) { | ||
| slavePool.pool.close(); | ||
| } | ||
| } | ||
|
|
||
| public HostAndPort getCurrentMaster() { | ||
|
|
@@ -167,6 +224,79 @@ private ConnectionPool createNodePool(HostAndPort master) { | |
| } | ||
| } | ||
|
|
||
| private void initSlaves(List<HostAndPort> slaves) { | ||
| List<PoolInfo> removedSlavePools = new ArrayList<>(); | ||
| try { | ||
| synchronized (slavePools) { | ||
|
||
| Loop: | ||
|
||
| for (int i = slavePools.size()-1; i >= 0; i--) { | ||
| PoolInfo poolInfo = slavePools.get(i); | ||
| for (HostAndPort slave : slaves) { | ||
| String host = slave.toString(); | ||
| if (poolInfo.host.equals(host)) { | ||
| continue Loop; | ||
| } | ||
| } | ||
| removedSlavePools.add(slavePools.remove(i)); | ||
| } | ||
|
|
||
| for (HostAndPort slave : slaves) { | ||
| addSlave(slave); | ||
| } | ||
| } | ||
| } finally { | ||
| if (!removedSlavePools.isEmpty() && clientSideCache != null) { | ||
| clientSideCache.flush(); | ||
| } | ||
|
|
||
| for (PoolInfo removedSlavePool : removedSlavePools) { | ||
| removedSlavePool.pool.destroy(); | ||
| } | ||
| } | ||
| } | ||
|
|
||
| private static boolean isHealthy(String flags) { | ||
| for (String flag : flags.split(",")) { | ||
| switch (flag.trim()) { | ||
| case "s_down": | ||
| case "o_down": | ||
| case "disconnected": | ||
| return false; | ||
| } | ||
| } | ||
| return true; | ||
| } | ||
|
|
||
| private void addSlave(HostAndPort slave) { | ||
| String newSlaveHost = slave.toString(); | ||
| synchronized (this.slavePools) { | ||
| for (int i = 0; i < this.slavePools.size(); i++) { | ||
| PoolInfo poolInfo = this.slavePools.get(i); | ||
| if (poolInfo.host.equals(newSlaveHost)) { | ||
| return; | ||
| } | ||
| } | ||
| slavePools.add(new PoolInfo(newSlaveHost, createNodePool(slave))); | ||
| } | ||
| } | ||
|
|
||
| private void removeSlave(HostAndPort slave) { | ||
| String newSlaveHost = slave.toString(); | ||
| PoolInfo removed = null; | ||
| synchronized (this.slavePools) { | ||
| for (int i = 0; i < this.slavePools.size(); i++) { | ||
| PoolInfo poolInfo = this.slavePools.get(i); | ||
| if (poolInfo.host.equals(newSlaveHost)) { | ||
| removed = slavePools.remove(i); | ||
| break; | ||
| } | ||
| } | ||
| } | ||
| if (removed != null) { | ||
| removed.pool.destroy(); | ||
| } | ||
| } | ||
|
|
||
| private HostAndPort initSentinels(Set<HostAndPort> sentinels) { | ||
|
|
||
| HostAndPort master = null; | ||
|
|
@@ -262,6 +392,24 @@ public void run() { | |
|
|
||
| sentinelJedis = new Jedis(node, sentinelClientConfig); | ||
|
|
||
| List<Map<String, String>> slaveInfos = sentinelJedis.sentinelSlaves(masterName); | ||
|
|
||
| List<HostAndPort> slaves = new ArrayList<>(); | ||
|
|
||
| for (int i = 0; i < slaveInfos.size(); i++) { | ||
| Map<String, String> slaveInfo = slaveInfos.get(i); | ||
| String flags = slaveInfo.get("flags"); | ||
| if (flags == null || !isHealthy(flags)) { | ||
| continue; | ||
| } | ||
| String ip = slaveInfo.get("ip"); | ||
| int port = Integer.parseInt(slaveInfo.get("port")); | ||
| HostAndPort slave = new HostAndPort(ip, port); | ||
| slaves.add(slave); | ||
| } | ||
|
|
||
| initSlaves(slaves); | ||
|
|
||
| // code for active refresh | ||
| List<String> masterAddr = sentinelJedis.sentinelGetMasterAddrByName(masterName); | ||
| if (masterAddr == null || masterAddr.size() != 2) { | ||
|
|
@@ -275,24 +423,58 @@ public void run() { | |
| public void onMessage(String channel, String message) { | ||
| LOG.debug("Sentinel {} published: {}.", node, message); | ||
|
|
||
| String[] switchMsg = message.split(" "); | ||
| String slaveIp; | ||
| int slavePort; | ||
| switch (channel) { | ||
| case "+switch-master": | ||
| if (switchMsg.length > 3) { | ||
| if (masterName.equals(switchMsg[0])) { | ||
| initMaster(toHostAndPort(switchMsg[3], switchMsg[4])); | ||
| } else { | ||
| LOG.debug( | ||
| "Ignoring message on +switch-master for master {}. Our master is {}.", | ||
| switchMsg[0], masterName); | ||
| } | ||
| } else { | ||
| LOG.error("Invalid message received on sentinel {} on channel +switch-master: {}.", | ||
| node, message); | ||
| } | ||
| break; | ||
| case "+sdown": | ||
| if (switchMsg[0].equals("master")) { | ||
| return; | ||
| } | ||
| if (!masterName.equals(switchMsg[5])) { | ||
| return; | ||
| } | ||
| slaveIp = switchMsg[2]; | ||
| slavePort = Integer.parseInt(switchMsg[3]); | ||
| removeSlave(new HostAndPort(slaveIp, slavePort)); | ||
| break; | ||
| case "-sdown": | ||
| if (!masterName.equals(switchMsg[5])) { | ||
| return; | ||
| } | ||
| slaveIp = switchMsg[2]; | ||
| slavePort = Integer.parseInt(switchMsg[3]); | ||
| addSlave(new HostAndPort(slaveIp, slavePort)); | ||
| break; | ||
| case "+slave": | ||
| if (!masterName.equals(switchMsg[5])) { | ||
| return; | ||
| } | ||
| slaveIp = switchMsg[2]; | ||
| slavePort = Integer.parseInt(switchMsg[3]); | ||
| addSlave(new HostAndPort(slaveIp, slavePort)); | ||
|
|
||
| String masterIp = switchMsg[6]; | ||
| int masterPort = Integer.parseInt(switchMsg[7]); | ||
| removeSlave(new HostAndPort(masterIp, masterPort)); | ||
| break; | ||
| } | ||
| } | ||
| }, "+switch-master", "+sdown", "-sdown", "+slave"); | ||
|
Collaborator
There was a problem hiding this comment. Based on High availability with Redis Sentinel Not clear the difference between Probably we should react on
Author
There was a problem hiding this comment. There is a problem here. When I kill a redis node, +odown is triggered, but when the node comes back online, there is no -odown event.
Author
There was a problem hiding this comment. And +odown will only be triggered when the master goes down, my redis version is 6.2.6 |
||
|
|
||
| } catch (JedisException e) { | ||
|
|
||
|
|
||
Oops, something went wrong.
Oops, something went wrong.
Add this suggestion to a batch that can be applied as a single commit.
This suggestion is invalid because no changes were made to the code.
Suggestions cannot be applied while the pull request is closed.
Suggestions cannot be applied while viewing a subset of changes.
Only one suggestion per line can be applied in a batch.
Add this suggestion to a batch that can be applied as a single commit.
Applying suggestions on deleted lines is not supported.
You must change the existing code in this line in order to create a valid suggestion.
Outdated suggestions cannot be applied.
This suggestion has been applied or marked resolved.
Suggestions cannot be applied from pending reviews.
Suggestions cannot be applied on multi-line comments.
Suggestions cannot be applied while the pull request is queued to merge.
Suggestion cannot be applied right now. Please check back later.
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
We are introducing support for READ commands only for the
JedisSentineledclient.Because of this, the read-from settings should be configured specifically in
JedisSentineled(or itsSentineledConnectionProvider) rather than being placed in the generic client configuration.In addition, I think it is better to determine whether a command is a READ command by exposing a configurable Predicate. This gives users flexibility and avoids the need to override methods. I like how this is achieved in Lettuce:
• Predicate interface: ReadOnlyPredicate
• Example implementation: ReadOnlyCommands
• Configuration: ClientOptions