diff --git a/modules/extension/src/main/java/org/springside/modules/nosql/redis/JedisUtils.java b/modules/extension/src/main/java/org/springside/modules/nosql/redis/JedisUtils.java index c5c62f61..a2854858 100644 --- a/modules/extension/src/main/java/org/springside/modules/nosql/redis/JedisUtils.java +++ b/modules/extension/src/main/java/org/springside/modules/nosql/redis/JedisUtils.java @@ -49,10 +49,10 @@ public class JedisUtils { } /** - * 退出然后关闭Jedis连接。 + * 退出然后关闭Jedis连接。如果Jedis为null则无动作。 */ public static void closeJedis(Jedis jedis) { - if (jedis.isConnected()) { + if ((jedis != null) && jedis.isConnected()) { try { try { jedis.quit(); diff --git a/modules/extension/src/main/java/org/springside/modules/nosql/redis/pool/JedisSentinelPool.java b/modules/extension/src/main/java/org/springside/modules/nosql/redis/pool/JedisSentinelPool.java index 110b14da..ff4c479e 100644 --- a/modules/extension/src/main/java/org/springside/modules/nosql/redis/pool/JedisSentinelPool.java +++ b/modules/extension/src/main/java/org/springside/modules/nosql/redis/pool/JedisSentinelPool.java @@ -1,14 +1,11 @@ package org.springside.modules.nosql.redis.pool; -import java.util.ArrayList; import java.util.List; import java.util.concurrent.atomic.AtomicBoolean; import org.apache.commons.pool2.impl.GenericObjectPool; import org.slf4j.Logger; import org.slf4j.LoggerFactory; -import org.springside.modules.nosql.redis.JedisTemplate; -import org.springside.modules.nosql.redis.JedisTemplate.JedisAction; import org.springside.modules.nosql.redis.JedisUtils; import redis.clients.jedis.Jedis; @@ -23,13 +20,13 @@ import redis.clients.util.Pool; */ public class JedisSentinelPool extends Pool { - public static final String UNAVAILABLE_ADDRESS = "I dont know because no sentinel up"; + private static final String UNAVAILABLE_ADDRESS = "All sentinel down"; private static Logger logger = LoggerFactory.getLogger(JedisSentinelPool.class); private String masterName; - private List sentinelPools = new ArrayList(); + private ConnectionInfo[] sentinelInfos; private MasterSwitchListener masterSwitchListener; private JedisPoolConfig masterPoolConfig; @@ -56,10 +53,7 @@ public class JedisSentinelPool extends Pool { JedisPoolConfig masterPoolConfig) { // check and assign parameter, all parameter can't not be null or empty assertArgument(((sentinelInfos != null) && (sentinelInfos.length != 0)), "seintinelInfos is not set"); - for (ConnectionInfo sentinelInfo : sentinelInfos) { - JedisPool sentinelPool = new JedisPool(sentinelInfo, new JedisPoolConfig()); - sentinelPools.add(sentinelPool); - } + this.sentinelInfos = sentinelInfos; assertArgument(masterAddtionalInfo != null, "masterAddtionalInfo is not set"); this.masterAddtionalInfo = masterAddtionalInfo; @@ -79,10 +73,6 @@ public class JedisSentinelPool extends Pool { public void destroy() { masterSwitchListener.shutdown(); - for (JedisPool sentinelPool : sentinelPools) { - sentinelPool.destroy(); - } - destroyInternalPool(); try { @@ -133,8 +123,8 @@ public class JedisSentinelPool extends Pool { public static final String THREAD_NAME_PREFIX = "RedisMasterSwitchListener-"; public static final int RETRY_WAIT_TIME_MILLS = 1000; - private JedisPool currentSentinelPool; - private Jedis subscriberJedis; + private ConnectionInfo sentinelInfo; + private Jedis sentinelJedis; private JedisPubSub subscriber; private AtomicBoolean running = new AtomicBoolean(true); @@ -146,61 +136,61 @@ public class JedisSentinelPool extends Pool { @Override public void run() { - while (running.get()) { try { - currentSentinelPool = selectSentinel(); + boolean avalibeSentinelExist = selectSentinel(); + if (avalibeSentinelExist) { + try { + ConnectionInfo masterConnectionInfo = queryMasterAddress(); + if ((internalPool != null) && isMasterAddressChanged(masterConnectionInfo)) { + logger.info("The internalPool {} had changed, destroy it now.", + previousMasterConnectionInfo.getHostAndPort()); + destroyInternalPool(); + } - if (currentSentinelPool != null) { + if ((internalPool == null) || isMasterAddressChanged(masterConnectionInfo)) { + logger.info("The internalPool {} is not init or the address had changed, init it now.", + masterConnectionInfo.getHostAndPort()); + initInternalPool(masterConnectionInfo); + } - ConnectionInfo masterConnectionInfo = queryMasterAddress(currentSentinelPool); + previousMasterConnectionInfo = masterConnectionInfo; - if ((internalPool != null) && isMasterAddressChanged(masterConnectionInfo)) { - logger.info("The internalPool {} had changed, destroy it now.", - previousMasterConnectionInfo.getHostAndPort()); - destroyInternalPool(); + // blocking listen master switch message until exception happen. + subscriber = new MasterSwitchSubscriber(); + sentinelJedis.subscribe(subscriber, "+switch-master", "+redirect-to-master"); + } catch (JedisConnectionException e) { + + JedisUtils.closeJedis(sentinelJedis); + + if (running.get()) { + logger.error("Lost connection with Sentinel " + sentinelInfo.getHostAndPort() + + ", sleep 1000ms and try to connect other one."); + sleep(RETRY_WAIT_TIME_MILLS); + } + } catch (Exception e) { + JedisUtils.closeJedis(sentinelJedis); + + if (running.get()) { + logger.error( + "Unexpected Exception happen, current Sentinel is" + + sentinelInfo.getHostAndPort() + ", sleep 1000ms and try again.", e); + sleep(RETRY_WAIT_TIME_MILLS); + } } - - if ((internalPool == null) || isMasterAddressChanged(masterConnectionInfo)) { - logger.info("The internalPool {} is not init or the address had changed, init it now.", - masterConnectionInfo.getHostAndPort()); - initInternalPool(masterConnectionInfo); - } - - previousMasterConnectionInfo = masterConnectionInfo; - - // blocking listen master switch message until exception happen. - subscriber = new MasterSwitchSubscriber(); - subscriberJedis = currentSentinelPool.getResource(); - subscriberJedis.subscribe(subscriber, "+switch-master", "+redirect-to-master"); } else { logger.info("All sentinels down, sleep 1000ms and try to select again."); // when the system startup but the sentinels not yet, init an ugly address to prevent null point // exception. if (internalPool == null) { - ConnectionInfo currentMasterConnectionInfo = new ConnectionInfo(UNAVAILABLE_ADDRESS); - initInternalPool(currentMasterConnectionInfo); - previousMasterConnectionInfo = currentMasterConnectionInfo; + ConnectionInfo masterConnectionInfo = new ConnectionInfo(UNAVAILABLE_ADDRESS); + initInternalPool(masterConnectionInfo); + previousMasterConnectionInfo = masterConnectionInfo; } sleep(RETRY_WAIT_TIME_MILLS); } - } catch (JedisConnectionException e) { - if (subscriberJedis != null) { - currentSentinelPool.returnBrokenResource(subscriberJedis); - } - - if (running.get()) { - logger.error("Lost connection with Sentinel " + currentSentinelPool.getHostAndPort() - + ", sleep 1000ms and try to connect other one."); - sleep(RETRY_WAIT_TIME_MILLS); - } } catch (Exception e) { - if (running.get()) { - logger.error( - "Some Exception happen, current Sentinel is" + currentSentinelPool.getHostAndPort() - + ", sleep 1000ms and try again.", e); - sleep(RETRY_WAIT_TIME_MILLS); - } + logger.error("Unexpected exception happen", e); } } } @@ -217,54 +207,48 @@ public class JedisSentinelPool extends Pool { subscriber.unsubscribe(); } } finally { - if (subscriberJedis != null) { - JedisUtils.closeJedis(subscriberJedis); - } + JedisUtils.closeJedis(sentinelJedis); } } /** - * Pickup the first available sentinel, if all sentinel down, return null. + * Pickup the first available sentinel, if all sentinel down, return false. */ - private JedisPool selectSentinel() { - for (JedisPool sentinelPool : sentinelPools) { - if (ping(sentinelPool)) { - return sentinelPool; + private boolean selectSentinel() { + for (ConnectionInfo info : sentinelInfos) { + if (ping(info)) { + sentinelInfo = info; + sentinelJedis = new Jedis(sentinelInfo.getHost(), sentinelInfo.getPort()); + return true; } } - return null; + sentinelInfo = null; + sentinelJedis = null; + return false; } /** * Ping the jedis instance, return true if the result is PONG. */ - private boolean ping(Pool pool) { - JedisTemplate template = new JedisTemplate(pool); + private boolean ping(ConnectionInfo connectionInfo) { + Jedis jedis = new Jedis(connectionInfo.getHost(), connectionInfo.getPort()); try { - String result = template.execute(new JedisAction() { - @Override - public String action(Jedis jedis) { - return jedis.ping(); - } - }); + String result = jedis.ping(); return (result != null) && result.equals("PONG"); } catch (JedisException e) { return false; + } finally { + JedisUtils.closeJedis(jedis); } } /** * Query master address from sentinel. */ - private ConnectionInfo queryMasterAddress(Pool sentinelPool) { - JedisTemplate sentinelTemplate = new JedisTemplate(sentinelPool); - List address = sentinelTemplate.execute(new JedisAction>() { - @Override - public List action(Jedis jedis) { - return jedis.sentinelGetMasterAddrByName(masterName); - } - }); + private ConnectionInfo queryMasterAddress() { + + List address = sentinelJedis.sentinelGetMasterAddrByName(masterName); if ((address == null) || address.isEmpty()) { throw new IllegalArgumentException("Master name " + masterName + " is not in sentinel.conf"); @@ -287,7 +271,7 @@ public class JedisSentinelPool extends Pool { return true; } - return (previousMasterConnectionInfo.getHostAndPort().equals(currentMasterConnectionInfo.getHostAndPort())); + return !previousMasterConnectionInfo.getHostAndPort().equals(currentMasterConnectionInfo.getHostAndPort()); } private void sleep(int millseconds) { @@ -306,7 +290,7 @@ public class JedisSentinelPool extends Pool { public void onMessage(String channel, String message) { // message example: +switch-master: mymaster 127.0.0.1 6379 127.0.0.1 6380 // +redirect-to-master mymaster 127.0.0.1 6380 127.0.0.1 6381 (if slave-master fail-over quick enough) - logger.info("Sentinel " + currentSentinelPool.getHostAndPort() + " published: " + message); + logger.info("Sentinel " + sentinelInfo.getHostAndPort() + " published: " + message); String[] switchMasterMsg = message.split(" "); // if the switeched master name equals my master name, destroy the old pool and init a new pool if (masterName.equals(switchMasterMsg[0])) {