private static final String MASTER_NAME = "mymaster";
protected static final HostAndPort sentinel1 = HostAndPorts.getSentinelServers().get(1);
protected static final HostAndPort sentinel2 = HostAndPorts.getSentinelServers().get(3);
@Before
public void setUp() throws Exception {
sentinels.clear();
sentinels.add(sentinel1.toString());
sentinels.add(sentinel2.toString());
}
@Test
public void repeatedSentinelPoolInitialization() {
for (int i = 0; i < 20; ++i) {
GenericObjectPoolConfig<Jedis> config = new GenericObjectPoolConfig<>();
JedisSentinelPool pool = new JedisSentinelPool(MASTER_NAME, sentinels, config, 1000,
"foobared", 2);
pool.getResource().close();
pool.destroy();
}
}
可以看到首先是创建了sentinel 的HostAndPort 对象，然后创建了连接池
public JedisSentinelPool(String masterName, Set<String> sentinels,
final GenericObjectPoolConfig<Jedis> poolConfig, int timeout, final String password,
final int database) {
this(masterName, sentinels, poolConfig, timeout, timeout, null, password, database);
}
...
public JedisSentinelPool(String masterName, Set<String> sentinels,
final GenericObjectPoolConfig<Jedis> poolConfig,
final int connectionTimeout, final int soTimeout, final int infiniteSoTimeout,
final String user, final String password, final int database, final String clientName,
final int sentinelConnectionTimeout, final int sentinelSoTimeout, final String sentinelUser,
final String sentinelPassword, final String sentinelClientName) {
this(masterName, parseHostAndPorts(sentinels), poolConfig,
DefaultJedisClientConfig.builder().connectionTimeoutMillis(connectionTimeout)
.socketTimeoutMillis(soTimeout).blockingSocketTimeoutMillis(infiniteSoTimeout)
.user(user).password(password).database(database).clientName(clientName).build(),
DefaultJedisClientConfig.builder().connectionTimeoutMillis(sentinelConnectionTimeout)
.socketTimeoutMillis(sentinelSoTimeout).user(sentinelUser).password(sentinelPassword)
.clientName(sentinelClientName).build()
);
}
...
public JedisSentinelPool(String masterName, Set<HostAndPort> sentinels,
final GenericObjectPoolConfig<Jedis> poolConfig, final JedisClientConfig masterClientConfig,
final JedisClientConfig sentinelClientConfig) {
this(masterName, sentinels, poolConfig, new JedisFactory(masterClientConfig), sentinelClientConfig);
}
public JedisSentinelPool(String masterName, Set<HostAndPort> sentinels,
final GenericObjectPoolConfig<Jedis> poolConfig, final JedisFactory factory,
final JedisClientConfig sentinelClientConfig) {
super(poolConfig, factory);
this.factory = factory;
this.sentinelClientConfig = sentinelClientConfig;
HostAndPort master = initSentinels(sentinels, masterName);
initMaster(master);
}
这里执行了两个重要方法
initSentinels 和 initMaster
private HostAndPort initSentinels(Set<HostAndPort> sentinels, final String masterName) {
HostAndPort master = null;
boolean sentinelAvailable = false;
LOG.info("Trying to find master from available Sentinels...");
for (HostAndPort sentinel : sentinels) {
LOG.debug("Connecting to Sentinel {}", sentinel);
//连接sentinel 节点
try (Jedis jedis = new Jedis(sentinel, sentinelClientConfig)) {
// 向sentinel发送命令 sentinel get-master-addr-by-name mymaster
List<String> masterAddr = jedis.sentinelGetMasterAddrByName(masterName);
// connected to sentinel...
sentinelAvailable = true;
if (masterAddr == null || masterAddr.size() != 2) {
LOG.warn("Can not get master addr, master name: {}. Sentinel: {}", masterName, sentinel);
continue;
}
master = toHostAndPort(masterAddr);
LOG.debug("Found Redis master at {}", master);
break;
} catch (JedisException e) {
// resolves #1036, it should handle JedisException there's another chance
// of raising JedisDataException
LOG.warn(
"Cannot get master address from sentinel running @ {}. Reason: {}. Trying next one.", sentinel, e);
}
}
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 {}, starting Sentinel listeners...", master);
for (HostAndPort sentinel : sentinels) {
MasterListener masterListener = new MasterListener(masterName, sentinel.getHost(), sentinel.getPort());
// whether MasterListener threads are alive or not, process can be stopped
masterListener.setDaemon(true);
masterListeners.add(masterListener);
masterListener.start();
}
return master;
}
jedis.sentinelGetMasterAddrByName(masterName);,即向sentinel发送命令sentinel get-master-addr-by-name mymaster, 用来获取master节点的地址，并将地址返回
initMaster(master); private void initMaster(HostAndPort master) {
synchronized (initPoolLock) {
if (!master.equals(currentHostMaster)) {
currentHostMaster = master;
// 这里是容易忽略但非常关键的一步
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
super.clear();
LOG.info("Created JedisSentinelPool to master at {}", master);
}
}
}
它是由 构造方法中的 new JedisFactory(masterClientConfig) 构造出来，在单元测试中我们得知masterClientConfig里面的属性都是空值
getResouce() @Override
public Jedis getResource() {
while (true) {
// 关键一步
Jedis jedis = super.getResource();
// 这里没啥大用，容易误导
jedis.setDataSource(this);
// get a reference because it can change concurrently
final HostAndPort master = currentHostMaster;
final HostAndPort connection = jedis.getClient().getHostAndPort();
if (master.equals(connection)) {
// connected to the correct master
return jedis;
} else {
returnBrokenResource(jedis);
}
}
}
super.getResource()， 父类是Pool， 而Pool 的对象一般是由Factory 构建出来 public JedisSentinelPool(String masterName, Set<HostAndPort> sentinels,
final GenericObjectPoolConfig<Jedis> poolConfig, final JedisClientConfig masterClientConfig,
final JedisClientConfig sentinelClientConfig) {
this(masterName, sentinels, poolConfig, new JedisFactory(masterClientConfig), sentinelClientConfig);
}
public JedisSentinelPool(String masterName, Set<HostAndPort> sentinels,
final JedisFactory factory, final JedisClientConfig sentinelClientConfig) {
super(factory);
this.factory = factory;
this.sentinelClientConfig = sentinelClientConfig;
HostAndPort master = initSentinels(sentinels, masterName);
initMaster(master);
}
由此可知 factory 是 new JedisFactory(masterClientConfig), 并且由 父类子类都引用到，并且在 initMaster 方法中调用factory.setHostAndPort(currentHostMaster); 更新了master的地址。
public T getResource() {
try {
return super.borrowObject();
} catch (JedisException je) {
throw je;
} catch (Exception e) {
throw new JedisException("Could not get a resource from the pool", e);
}
}
这里borrowObject 时，实际是调用工厂的方法干活，直接看工厂类JedisFactory
@Override
public PooledObject<Jedis> makeObject() throws Exception {
Jedis jedis = null;
try {
jedis = new Jedis(jedisSocketFactory, clientConfig);
return new DefaultPooledObject<>(jedis);
} catch (JedisException je) {
logger.debug("Error while makeObject", je);
throw je;
}
}
protected JedisFactory(final URI uri, final int connectionTimeout, final int soTimeout,
final int infiniteSoTimeout, final String clientName, final SSLSocketFactory sslSocketFactory,
final SSLParameters sslParameters, final HostnameVerifier hostnameVerifier) {
if (!JedisURIHelper.isValid(uri)) {
throw new InvalidURIException(String.format(
"Cannot open Redis connection due invalid URI. %s", uri.toString()));
}
this.clientConfig = DefaultJedisClientConfig.builder().connectionTimeoutMillis(connectionTimeout)
.socketTimeoutMillis(soTimeout).blockingSocketTimeoutMillis(infiniteSoTimeout)
.user(JedisURIHelper.getUser(uri)).password(JedisURIHelper.getPassword(uri))
.database(JedisURIHelper.getDBIndex(uri)).clientName(clientName)
.protocol(JedisURIHelper.getRedisProtocol(uri))
.ssl(JedisURIHelper.isRedisSSLScheme(uri)).sslSocketFactory(sslSocketFactory)
.sslParameters(sslParameters).hostnameVerifier(hostnameVerifier).build();
this.jedisSocketFactory = new DefaultJedisSocketFactory(new HostAndPort(uri.getHost(), uri.getPort()), this.clientConfig);
}
void setHostAndPort(final HostAndPort hostAndPort) {
if (!(jedisSocketFactory instanceof DefaultJedisSocketFactory)) {
throw new IllegalStateException("setHostAndPort method has limited capability.");
}
((DefaultJedisSocketFactory) jedisSocketFactory).updateHostAndPort(hostAndPort);
}
public Jedis(final JedisSocketFactory jedisSocketFactory, final JedisClientConfig clientConfig) {
connection = new Connection(jedisSocketFactory, clientConfig);
RedisProtocol proto = clientConfig.getRedisProtocol();
if (proto != null) commandObjects.setProtocol(proto);
}
这里直接构造出Connectrion 对象， 并传入socketFactory
public Connection(final JedisSocketFactory socketFactory, JedisClientConfig clientConfig) {
this.socketFactory = socketFactory;
this.soTimeout = clientConfig.getSocketTimeoutMillis();
this.infiniteSoTimeout = clientConfig.getBlockingSocketTimeoutMillis();
initializeFromClientConfig(clientConfig);
}
在这个构造方法中执行关键方法initializeFromClientConfig
private void initializeFromClientConfig(final JedisClientConfig config) {
try {
connect();
......
}
connect()
public void connect() throws JedisConnectionException {
if (!isConnected()) {
try {
socket = socketFactory.createSocket();
soTimeout = socket.getSoTimeout(); //?
outputStream = new RedisOutputStream(socket.getOutputStream());
inputStream = new RedisInputStream(socket.getInputStream());
broken = false; // unset broken status when connection is (re)initialized
} catch (JedisConnectionException jce) {
setBroken();
throw jce;
} catch (IOException ioe) {
setBroken();
throw new JedisConnectionException("Failed to create input/output stream", ioe);
} finally {
if (broken) {
IOUtils.closeQuietly(socket);
}
}
}
}