Alawn/springside4
1
0
Fork 0
Code Issues Pull Requests Releases Wiki Activity

#332 endless refactoring, don't use pool ,just use jedis for setinel

Browse Source
This commit is contained in:
Calvin
2014-02-10 12:16:01 +08:00
parent 484941eb91
commit d9d30ec2ef
2 changed files with 68 additions and 84 deletions
Download Patch File Download Diff File Expand all files Collapse all files
modules/extension/src/main/java/org/springside/modules/nosql/redis/JedisUtils.java
+2 -2
View File
@@ -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();
modules/extension/src/main/java/org/springside/modules/nosql/redis/pool/JedisSentinelPool.java
+66 -82
View File
@@ -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<Jedis> {
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<JedisPool> sentinelPools = new ArrayList<JedisPool>();
private ConnectionInfo[] sentinelInfos;
private MasterSwitchListener masterSwitchListener;
private JedisPoolConfig masterPoolConfig;
@@ -56,10 +53,7 @@ public class JedisSentinelPool extends Pool<Jedis> {
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<Jedis> {
public void destroy() {
masterSwitchListener.shutdown();
for (JedisPool sentinelPool : sentinelPools) {
sentinelPool.destroy();
}
destroyInternalPool();
try {
@@ -133,8 +123,8 @@ public class JedisSentinelPool extends Pool<Jedis> {
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<Jedis> {
@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<Jedis> {
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<Jedis> 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<String>() {
@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<Jedis> sentinelPool) {
JedisTemplate sentinelTemplate = new JedisTemplate(sentinelPool);
List<String> address = sentinelTemplate.execute(new JedisAction<List<String>>() {
@Override
public List<String> action(Jedis jedis) {
return jedis.sentinelGetMasterAddrByName(masterName);
}
});
private ConnectionInfo queryMasterAddress() {
List<String> 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<Jedis> {
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<Jedis> {
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])) {