客户端高可用
我们直接来分析jedis的源码，找到相应的构造函数
public JedisSentinelPool(String masterName, Set<String> sentinels,
final GenericObjectPoolConfig poolConfig, final int connectionTimeout, final int soTimeout,
final String password, final int database, final String clientName) {
this.poolConfig = poolConfig;
this.connectionTimeout = connectionTimeout;
this.soTimeout = soTimeout;
this.password = password;
this.database = database;
this.clientName = clientName;
//这是最主要的代码
HostAndPort master = initSentinels(sentinels, masterName);
initPool(master);
}
初始化sentinel节点
private HostAndPort initSentinels(Set<String> sentinels, final String masterName) {
HostAndPort master = null;
boolean sentinelAvailable = false;
log.info("Trying to find master from available Sentinels...");
//对我们的sentinel集合进行遍历
for (String sentinel : sentinels) {
final HostAndPort hap = toHostAndPort(Arrays.asList(sentinel.split(":")));
log.fine("Connecting to Sentinel " + hap);
Jedis jedis = null;
try {
//尝试连接sentinel节点
jedis = new Jedis(hap.getHost(), hap.getPort());
//通过sentinel来获取指定master主节点的地址
List<String> masterAddr = jedis.sentinelGetMasterAddrByName(masterName);
// connected to sentinel...
sentinelAvailable = true;
//如果主节点地址为空或者不符合规则 continue 进行下一次循环，也就是连接下一个sentinel节点
if (masterAddr == null || masterAddr.size() != 2) {
log.warning("Can not get master addr, master name: " + masterName + ". Sentinel: " + hap
+ ".");
continue;
}
//获取master节点
master = toHostAndPort(masterAddr);
log.fine("Found Redis master at " + master);
break;
} catch (JedisException e) {
// resolves #1036, it should handle JedisException there's another chance
// of raising JedisDataException
log.warning("Cannot get master address from sentinel running @ " + hap
+ ". Reason: " + e + ". Trying next one.");
} finally {
if (jedis != null) {
jedis.close();
}
}
}
//遍历所有sentinel节点都无法获取master节点信息，抛出异常
if (master == null) {
if (sentinelAvailable) {
// can connect to sentinel, but master name seems to not
// monitored
throw new JedisException("Can connect to sentinel, but " + masterName
+ " seems to be not monitored...");
} else {
throw new JedisConnectionException("All sentinels down, cannot determine where is "
+ masterName + " master is running...");
}
}
log.info("Redis master running at " + master + ", starting Sentinel listeners...");
for (String sentinel : sentinels) {
final HostAndPort hap = toHostAndPort(Arrays.asList(sentinel.split(":")));
//初始化主节点监听器（是一个线程）
MasterListener masterListener = new MasterListener(masterName, hap.getHost(), hap.getPort());
// whether MasterListener threads are alive or not, process can be stopped
//设置为守护线程
masterListener.setDaemon(true)；
masterListeners.add(masterListener);
//开启监听主从的切换
masterListener.start();
}
//返回成功连接的主节点
return master;
}
下面是MasterListener 的run方法实现
protected AtomicBoolean running = new AtomicBoolean(false);
public void run() {
//设置线程状态，这里用的是原子变量
running.set(true);
while (running.get()) {
// 连接sentinel节点
j = new Jedis(host, port);
//DCL机制
try {
// double check that it is not being shutdown
if (!running.get()) {
break;
}
//订阅sentinel节点频道，主从切换是switch-master频道
j.subscribe(new JedisPubSub() {
@Override
public void onMessage(String channel, String message) {
log.fine("Sentinel " + host + ":" + port + " published: " + message + ".");
String[] switchMasterMsg = message.split(" ");
//当sentinel节点发布了switch-master的消息，也就是发生了主从切换
if (switchMasterMsg.length > 3) {
//如果当前主节点宕机
if (masterName.equals(switchMasterMsg[0])) {
//重新初始化连接池
initPool(toHostAndPort(Arrays.asList(switchMasterMsg[3], switchMasterMsg[4])));
} else {
log.fine("Ignoring message on +switch-master for master name "
+ switchMasterMsg[0] + ", our master name is " + masterName);
}
} else {
log.severe("Invalid message received on Sentinel " + host + ":" + port
+ " on channel +switch-master: " + message);
}
}
}, "+switch-master");
} catch (JedisConnectionException e) {
if (running.get()) {
log.log(Level.SEVERE, "Lost connection to Sentinel at " + host + ":" + port
+ ". Sleeping 5000ms and retrying.", e);
try {
Thread.sleep(subscribeRetryWaitTimeMillis);
} catch (InterruptedException e1) {
log.log(Level.SEVERE, "Sleep interrupted: ", e1);
}
} else {
log.fine("Unsubscribing from Sentinel at " + host + ":" + port);
}
} finally {
j.close();
}
}
}
初始化连接池
private void initPool(HostAndPort master) {
//判断当前主节点是否与新切换节点一致
if (!master.equals(currentHostMaster)) {
//设置当前主节点为新切换主节点
currentHostMaster = master;
if (factory == null) {
//初始化Jedis工厂
factory = new JedisFactory(master.getHost(), master.getPort(), connectionTimeout,
soTimeout, password, database, clientName);
//初始化连接池
initPool(poolConfig, factory);
} else {
//如果Jedis工厂已经有了，清除连接池中已有的连接
factory.setHostAndPort(currentHostMaster);
// although we clear the pool, we still have to check the
// returned object
// in getResource, this call only clears idle instances, not
// borrowed instances
internalPool.clear();
}
log.info("Created JedisPool to master at " + master);
}
}
从节点的作用
- 副本：高可用的基础
- 扩展：读能力
- 三个消息
- +switch-master：切换主节点（从节点晋升为主节点）
- +convert-to-slave：切换从节点（原主节点降为从节点）
-
+sdown：主观下线
从节点
客户端完成一个类似于JedisPool的连接池用于监视从节点的信息（上述三条）
网友评论