Jedis的JedisSentinelPool源代码分析-监听切换事件
https://segmentfault.com/a/1190000002690506java
本专栏与Redis相关的文章redis
Redis Sentinel机制与用法（一）
Redis Sentinel机制与用法（二）
Jedis的JedisSentinelPool源代码分析
Jedis的Sharded源代码分析
Redis 主从 Replication 的配置
详解Redis SORT命令
JedisCommand接口说明数据库
概述
Jedis是Redis官方推荐的Java客户端，更多Redis的客户端能够参考Redis官网客户端列表。Redis-Sentinel做为官方推荐的HA解决方案，Jedis也在客户端角度实现了对Sentinel的支持，主要实如今JedisSentinelPool.java这个类中，下文会分析这个类的实现。apache
属性
JedisSentinelPool类里有如下的属性：segmentfault
//基于apache的commom-pool2的对象池配置
protected GenericObjectPoolConfig poolConfig;
//超时时间，默认是2000
protected int timeout = Protocol.DEFAULT_TIMEOUT;
//sentinel的密码
protected String password;
//redis数据库的数目
protected int database = Protocol.DEFAULT_DATABASE;
//master监听器，当master的地址发生改变时，会触发这些监听者
protected Set<MasterListener> masterListeners = new HashSet<MasterListener>();
protected Logger log = Logger.getLogger(getClass().getName());
//Jedis实例建立工厂
private volatile JedisFactory factory;
//当前的master，HostAndPort是一个简单的包装了ip和port的模型类
private volatile HostAndPort currentHostMaster;
构造器
构造器的代码以下：多线程
public JedisSentinelPool(String masterName, Set<String> sentinels, final GenericObjectPoolConfig poolConfig, int timeout, final String password, final int database) {
this.poolConfig = poolConfig;
this.timeout = timeout;
this.password = password;
this.database = database;
HostAndPort master = initSentinels(sentinels, masterName);
initPool(master);
}
构造器一开始对实例变量进行赋值，参数sentinels是客户端所须要打交道的Redis-Sentinel，容许有多个，用一个集合来盛装。ide
而后经过initSentinels方法与sentinel沟通后，肯定当前sentinel所监视的master是哪个。而后经过master来建立好对象池，以便后续从对象池中取出一个Jedis实例，来对master进行操做。memcached
initSentinels方法
initSentinels方法的代码以下所示，我加了一些注释：函数
private HostAndPort initSentinels(Set<String> sentinels, final String masterName) {
HostAndPort master = null;
boolean sentinelAvailable = false;
log.info("Trying to find master from available Sentinels...");
// 有多个sentinels,遍历这些个sentinels
for (String sentinel : sentinels) {
// host:port表示的sentinel地址转化为一个HostAndPort对象。
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());
// 根据masterName获得master的地址，返回一个list，host= list[0], port =
// list[1]
List<String> masterAddr = jedis.sentinelGetMasterAddrByName(masterName);
// connected to sentinel...
sentinelAvailable = true;
if (masterAddr == null || masterAddr.size() != 2) {
log.warning("Can not get master addr, master name: " + masterName
+ ". Sentinel: " + hap + ".");
continue;
}
master = toHostAndPort(masterAddr);
log.fine("Found Redis master at " + master);
// 若是在任何一个sentinel中找到了master，再也不遍历sentinels
break;
} catch (JedisConnectionException e) {
log.warning("Cannot connect to sentinel running @ " + hap
+ ". Trying next one.");
} finally {
// 关闭与sentinel的链接
if (jedis != null) {
jedis.close();
}
}
}
// 到这里，若是master为null，则说明有两种状况，一种是全部的sentinels节点都down掉了，一种是master节点没有被存活的sentinels监控到
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...");
}
}
//若是走到这里，说明找到了master的地址
log.info("Redis master running at " + master + ", starting Sentinel listeners...");
//启动对每一个sentinels的监听
for (String sentinel : sentinels) {
final HostAndPort hap = toHostAndPort(Arrays.asList(sentinel.split(":")));
MasterListener masterListener = new MasterListener(masterName, hap.getHost(),
hap.getPort());
masterListeners.add(masterListener);
masterListener.start();
}
return master;
}
能够看到initSentinels方法的参数有一个masterName，就是咱们所须要查找的master的名字。
一开始，遍历多个sentinels，一个一个链接到sentinel，去询问关于masterName的消息，能够看到是经过jedis.sentinelGetMasterAddrByName()方法去链接sentinel，并询问当前的master的地址。点进这个方法去看看，源代码是这样写的：ui
/**
* <pre>
* redis 127.0.0.1:26381> sentinel get-master-addr-by-name mymaster
* 1) "127.0.0.1"
* 2) "6379"
* </pre>
* @param masterName
* @return two elements list of strings : host and port.
*/
public List<String> sentinelGetMasterAddrByName(String masterName) {
client.sentinel(Protocol.SENTINEL_GET_MASTER_ADDR_BY_NAME, masterName);
final List<Object> reply = client.getObjectMultiBulkReply();
return BuilderFactory.STRING_LIST.build(reply);
}
调用的是与Jedis绑定的client去发送一个"get-master-addr-by-name"命令。
回到initSentinels方法中，若是没有询问到master的地址，那就询问下一个sentinel。若是询问到了master的地址，那么将再也不遍历sentinel集合，直接break退出循环遍历。
若是循环结束后，master的值为null,那么有两种可能：
-
一种是全部的sentinel实例都不可用了
-
另一种是，sentinel实例有可用的，可是没有监控名字为masterName的Redis。
若是master为null，程序会抛出异常，再也不往下走了。若是master不为null呢，继续往下走。
能够从代码中看到，为每一个sentinel都启动了一个监听者MasterListener。MasterListener自己是一个线程，它会去订阅sentinel上关于master节点地址改变的消息。
接下来先分析构造方法中的另一个方法：initPool。以后再看MasterListener的实现。
initPool方法
initPool的实现源代码以下所示：
private void initPool(HostAndPort master) {
if (!master.equals(currentHostMaster)) {
currentHostMaster = master;
if (factory == null) {
factory = new JedisFactory(master.getHost(), master.getPort(), timeout,
password, database);
initPool(poolConfig, factory);
} else {
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);
}
}
能够看到，做为参数传进来的master会与实例变量currentHostMaster做比较，看看是不是相同的，为何要做这个比较呢，由于前文中提到的MasterListener会在发现master地址改变之后，去调用initPool方法。
若是是第一次调用initPool方法(构造函数中调用)，那么会初始化Jedis实例建立工厂，若是不是第一次调用(MasterListener中调用)，那么只对已经初始化的工厂进行从新设置。
从以上也能够看出为何currentHostMaster和factory这两个变量为何要声明为volatile，它们会在多线程环境下被访问和修改，所以必须保证可见性。
第一次调用时，会调用initPool(poolConfig, factory)方法。
看看这个方法的源代码：
public void initPool(final GenericObjectPoolConfig poolConfig,
PooledObjectFactory<T> factory) {
if (this.internalPool != null) {
try {
closeInternalPool();
} catch (Exception e) {
}
}
this.internalPool = new GenericObjectPool<T>(factory, poolConfig);
}
基本上只干了一件事：初始化内部对象池。
MasterListener监听者线程
直接看它的run方法实现吧：
public void run() {
running.set(true);
while (running.get()) {
j = new Jedis(host, port);
try {
//订阅sentinel上关于master地址改变的消息
j.subscribe(new JedisPubSub() {
@Override
public void onMessage(String channel, String message) {
log.fine("Sentinel " + host + ":" + port + " published: "
+ message + ".");
String[] switchMasterMsg = message.split(" ");
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.severe("Lost connection to Sentinel at " + host + ":" + port
+ ". Sleeping 5000ms and retrying.");
try {
Thread.sleep(subscribeRetryWaitTimeMillis);
} catch (InterruptedException e1) {
e1.printStackTrace();
}
} else {
log.fine("Unsubscribing from Sentinel at " + host + ":" + port);
}
}
}
}
能够看到它依然委托了Jedis去与sentinel打交道，订阅了关于master地址变换的消息，当master地址变换时，就会再调用一次initPool方法，从新设置对象池相关的设置。
尾声
Jedis的JedisSentinelPool的实现仅仅适用于单个master-slave。 如今有了更多的需求，既须要sentinel提供的自动主备切换机制，又须要客户端可以作数据分片(Sharding)，相似于memcached用一致性哈希进行数据分片。 接下来可能会本身在现有Jedis上实现一个支持一致性哈希分片的ShardedJedisSentinelPool。
- 1. Jedis的JedisSentinelPool源代码分析
- 2. Jedis的JedisSentinelPool源代码、Sharded源代码分析
- 3. Spring源码-监听事件Applica**Listener和Applica**Event源码分析
- 4. react源码-事件监听
- 5. Jedis的Sharded源代码分析
- 6. Jedis源码分析
- 7. 源码分析ElasticJob任务运行事件监听器
- 8. （转）spring boot实战(第三篇)事件监听源码分析
- 9. zookeeper（10）源码分析-事件监听Watcher（3）
- 10. 代码添加EventTrigger监听事件
- 更多相关文章...
- • Markdown 代码 - Markdown 教程
- • Eclipse 代码模板 - Eclipse 教程
- • 互联网组织的未来：剖析GitHub员工的任性之源
- • IntelliJ IDEA安装代码格式化插件
-
每一个你不满意的现在，都有一个你没有努力的曾经。