扩展Redis的Jedis客户端,哨兵模式读请求走Slave集群
Redis哨兵模式,由Sentinel节点和Redis节点组成,哨兵节点负责监控Redis的健康状况,负责协调Redis主从复制的关系。
本文不详细讨论Redis哨兵模式,关于哨兵的详细介绍可以参考(https://blog.csdn.net/u010297957/article/details/55050098)
在使用哨兵模式以后,客户端不能直接连接到Redis集群,而是连接到哨兵集群,通过哨兵节点获取Redis主节点(Master)的信息,再进行连接,下面给出一小段代码。
-
JedisPoolConfig jedisPoolConfig = new JedisPoolConfig();
-
jedisPoolConfig.setMaxTotal(10);
-
jedisPoolConfig.setMaxIdle(5);
-
jedisPoolConfig.setMinIdle(5);
-
-
Set<String> sentinels = new HashSet<>(Arrays.asList(
-
"192.168.80.112:26379",
-
"192.168.80.113:26379",
-
"192.168.80.114:26379"
-
));
-
GenericObjectPoolConfig poolConfig = new GenericObjectPoolConfig();
-
poolConfig.setMaxTotal(10);
-
poolConfig.setMaxIdle(5);
-
poolConfig.setMinIdle(5);
-
JedisSentinelPool pool = new JedisSentinelPool("mymaster", sentinels, jedisPoolConfig);
可以看到,客户端只配置了哨兵集群的IP地址,通过哨兵获取redis主节点信息,再与其进行连接,下面给出关键代码的源码分析,下面代码片段讲述了如何获取主节点信息。
-
private HostAndPort initSentinels(Set<String> sentinels, final String masterName) {
-
//主节点ip与port对象
-
HostAndPort master = null;
-
//是否可以连接到哨兵节点
-
boolean sentinelAvailable = false;
-
-
log.info("Trying to find master from available Sentinels...");
-
-
//循环所有的哨兵节点,依次进行连接,逻辑如下:
-
//先连接第一个,如果连接成功,能够获取主节点信息,方法返回,否则连接第二个,第三个,第N个。
-
//如果全部都失败,则抛出异常,RedisPool初始化失败
-
for (String sentinel : sentinels) {
-
-
//哨兵的ip和port
-
final HostAndPort hap = HostAndPort.parseString(sentinel);
-
-
log.info("Connecting to Sentinel " + hap);
-
-
Jedis jedis = null;
-
try {
-
//与哨兵节点进行连接,这里可能会出错,比如哨兵挂了。。
-
jedis = new Jedis(hap.getHost(), hap.getPort());
-
-
//查询主节点信息
-
List<String> masterAddr = jedis.sentinelGetMasterAddrByName(masterName);
-
-
//设置可以连接到哨兵
-
sentinelAvailable = true;
-
-
//如果主节点信息获取不到,则继续循环,换一下哨兵继续上面逻辑
-
if (masterAddr == null || masterAddr.size() != 2) {
-
log.warn("Can not get master addr, master name: " + masterName + ". Sentinel: " + hap
-
+ ".");
-
continue;
-
}
-
-
//获取到主节点信息
-
master = toHostAndPort(masterAddr);
-
log.info("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 @ " + hap + ". Reason: " + e
-
+ ". Trying next one.");
-
} finally {
-
if (jedis != null) {
-
jedis.close();
-
}
-
}
-
}
-
-
//如果全部哨兵都获取不到主节点信息则抛出异常
-
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...");
-
-
//下面的逻辑是上面可以拿到主节点信息时才会执行
-
//循环哨兵,建立订阅消息,监听节点切换消息,如果redis集群节点发生变动,这里会收到通知
-
for (String sentinel : sentinels) {
-
final HostAndPort hap = HostAndPort.parseString(sentinel);
-
JedisSentinelSlavePool.MasterListener masterListener = new JedisSentinelSlavePool.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;
-
}
下面的代码判断,分析了客户端如何感知到redis主从节点关系发生变化,原理是通过订阅哨兵的频道获取的,当又新的主节点出现,则清空原有连接池,根据新的主节点重新创建连接对象。
-
-
-
running.set(true);
-
-
//死循环
-
while (running.get()) {
-
-
//与哨兵进行连接
-
j = new Jedis(host, port);
-
-
try {
-
// double check that it is not being shutdown
-
if (!running.get()) {
-
break;
-
}
-
-
//订阅频道监听节点切换消息
-
j.subscribe(new JedisPubSub() {
-
@Override
-
public void onMessage(String channel, String message) {
-
log.info("Sentinel " + host + ":" + port + " published: " + message + ".");
-
-
String[] switchMasterMsg = message.split(" ");
-
-
if (switchMasterMsg.length > 3) {
-
-
if (masterName.equals(switchMasterMsg[0])) {
-
//toHostAndPort(Arrays.asList(switchMasterMsg[3], switchMasterMsg[4]))
-
//这里获取了新的主节点的ip与端口
-
//调用initPool清空原来的连接池,这样一来,当需要获取jedis时,池会根据新的主接线创建对象
-
initPool(toHostAndPort(Arrays.asList(switchMasterMsg[3], switchMasterMsg[4])));
-
} else {
-
log.info("Ignoring message on +switch-master for master name "
-
+ switchMasterMsg[0] + ", our master name is " + masterName);
-
}
-
-
} else {
-
log.warn("Invalid message received on Sentinel " + host + ":" + port
-
+ " on channel +switch-master: " + message);
-
}
-
}
-
}, "+switch-master");
-
-
} catch (JedisConnectionException e) {
-
-
if (running.get()) {
-
log.info("Lost connection to Sentinel at " + host + ":" + port
-
+ ". Sleeping 5000ms and retrying.", e);
-
try {
-
//如果连接哨兵异常,则等待若干时间后无限重试
-
Thread.sleep(subscribeRetryWaitTimeMillis);
-
} catch (InterruptedException e1) {
-
log.info( "Sleep interrupted: ", e1);
-
}
-
} else {
-
log.info("Unsubscribing from Sentinel at " + host + ":" + port);
-
}
-
} finally {
-
j.close();
-
}
-
}
-
}
由此我们知道,Redis的客户端,在哨兵模式下的实现,读写都是走Master,那么缺点是显而易见的,那就是若干个Slave完全变成了热备,没有系统分担压力,接下来我们扩展它,让它支持可以在Slave节点读取数据,这样我们的程序,在写入数据时走Master,在读取数据时走Slave,大大提高了系统的性能。
第一步,我们重写这个类 JedisSentinelSlavePool extends Pool<Jedis>
所有代码都拷贝JedisSentinelPool,只修改了下面代码,创建了JedisSlaveFactory,传入了哨兵集群信息,和哨兵的名字。
-
private void initPool(HostAndPort master) {
-
if (!master.equals(currentHostMaster)) {
-
currentHostMaster = master;
-
if (factory == null) {
-
factory = new JedisSlaveFactory(sentinels, masterName, connectionTimeout,
-
soTimeout, password, database, clientName, false, null, null, null);
-
initPool(poolConfig, factory);
-
} else {
-
internalPool.clear();
-
}
-
-
log.info("Created JedisPool to master at " + master);
-
}
-
}
第二步,创建JedisSlaveFactory。
makeObject这个方法,是Redis连接池获取底层连接的地方,我么只需要在这里,创建一个连接到Slave节点的对象即可,
思路就是通过哨兵集群,获取到可用的slave节点信息,然后随机选取一个创建对象,达到负载均衡的效果。
-
package com.framework.core.redis;
-
-
import lombok.extern.slf4j.Slf4j;
-
import org.apache.commons.pool2.PooledObject;
-
import org.apache.commons.pool2.PooledObjectFactory;
-
import org.apache.commons.pool2.impl.DefaultPooledObject;
-
import redis.clients.jedis.BinaryJedis;
-
import redis.clients.jedis.HostAndPort;
-
import redis.clients.jedis.Jedis;
-
import redis.clients.jedis.exceptions.JedisConnectionException;
-
import redis.clients.jedis.exceptions.JedisException;
-
-
import javax.net.ssl.HostnameVerifier;
-
import javax.net.ssl.SSLParameters;
-
import javax.net.ssl.SSLSocketFactory;
-
import java.util.*;
-
-
@Slf4j
-
public class JedisSlaveFactory implements PooledObjectFactory<Jedis> {
-
-
private final Set<String> sentinels;
-
private final String masterName;
-
private final int connectionTimeout;
-
private final int soTimeout;
-
private final String password;
-
private final int database;
-
private final String clientName;
-
private final boolean ssl;
-
private final SSLSocketFactory sslSocketFactory;
-
private SSLParameters sslParameters;
-
private HostnameVerifier hostnameVerifier;
-
private Random random;
-
-
public JedisSlaveFactory(final Set<String> sentinels, final String masterName, final int connectionTimeout,
-
final int soTimeout, final String password, final int database, final String clientName,
-
final boolean ssl, final SSLSocketFactory sslSocketFactory, final SSLParameters sslParameters,
-
final HostnameVerifier hostnameVerifier) {
-
this.sentinels = sentinels;
-
this.masterName = masterName;
-
this.connectionTimeout = connectionTimeout;
-
this.soTimeout = soTimeout;
-
this.password = password;
-
this.database = database;
-
this.clientName = clientName;
-
this.ssl = ssl;
-
this.sslSocketFactory = sslSocketFactory;
-
this.sslParameters = sslParameters;
-
this.hostnameVerifier = hostnameVerifier;
-
this.random = new Random();
-
}
-
-
@Override
-
public void activateObject(PooledObject<Jedis> pooledJedis) throws Exception {
-
final BinaryJedis jedis = pooledJedis.getObject();
-
if (jedis.getDB() != database) {
-
jedis.select(database);
-
}
-
}
-
-
/**
-
* 销毁redis底层连接
-
*/
-
@Override
-
public void destroyObject(PooledObject<Jedis> pooledJedis){
-
log.debug("destroyObject =" + pooledJedis.getObject());
-
final BinaryJedis jedis = pooledJedis.getObject();
-
if (jedis.isConnected()) {
-
try {
-
jedis.quit();
-
jedis.disconnect();
-
} catch (Exception e) {
-
}
-
}
-
}
-
-
/**
-
* 创建Redis底层连接对象,返回池化对象.
-
*/
-
@Override
-
public PooledObject<Jedis> makeObject() {
-
List<HostAndPort> slaves = this.getAlivedSlaves();
-
-
//在slave节点中随机选取一个节点进行连接
-
int index = slaves.size() == 1 ? 0 : random.nextInt(slaves.size());
-
final HostAndPort hostAndPort = slaves.get(index);
-
-
log.debug("Create jedis instance from slaves=[" + slaves + "] , choose=[" + hostAndPort + "]");
-
-
//创建redis客户端
-
final Jedis jedis = new Jedis(hostAndPort.getHost(), hostAndPort.getPort(), connectionTimeout,
-
soTimeout, ssl, sslSocketFactory, sslParameters, hostnameVerifier);
-
-
//测试连接,设置密码,数据库.
-
try {
-
jedis.connect();
-
if (null != this.password) {
-
jedis.auth(this.password);
-
}
-
if (database != 0) {
-
jedis.select(database);
-
}
-
if (clientName != null) {
-
jedis.clientSetname(clientName);
-
}
-
} catch (JedisException je) {
-
jedis.close();
-
throw je;
-
}
-
-
return new DefaultPooledObject<Jedis>(jedis);
-
}
-
-
-
/**
-
* 获取可用的RedisSlave节点信息
-
*/
-
private List<HostAndPort> getAlivedSlaves() {
-
log.debug("Get alived salves start...");
-
-
List<HostAndPort> alivedSalaves = new ArrayList<>();
-
boolean sentinelAvailable = false;
-
-
//循环哨兵,建立连接获取slave节点信息
-
//当某个哨兵连接失败,会忽略异常连接下一个哨兵
-
for (String sentinel : sentinels) {
-
final HostAndPort hap = HostAndPort.parseString(sentinel);
-
-
log.debug("Connecting to Sentinel " + hap);
-
-
Jedis jedis = null;
-
try {
-
jedis = new Jedis(hap.getHost(), hap.getPort());
-
-
List<Map<String, String>> slavesInfo = jedis.sentinelSlaves(masterName);
-
-
//可以连接到哨兵
-
sentinelAvailable = true;
-
-
//没有查询到slave信息,循环下一个哨兵
-
if (slavesInfo == null || slavesInfo.size() == 0) {
-
log.warn("Cannot get slavesInfo, master name: " + masterName + ". Sentinel: " + hap
-
+ ". Trying next one.");
-
continue;
-
}
-
-
//获取可用的Slave信息
-
for (Map<String, String> slave : slavesInfo) {
-
if(slave.get("flags").equals("slave")) {
-
String host = slave.get("ip");
-
int port = Integer.valueOf(slave.get("port"));
-
HostAndPort hostAndPort = new HostAndPort(host, port);
-
-
log.info("Found alived redis slave:[" + hostAndPort + "]");
-
-
alivedSalaves.add(hostAndPort);
-
}
-
}
-
-
log.debug("Get alived salves end...");
-
break;
-
} catch (JedisException e) {
-
//当前哨兵连接失败,忽略错误连接下一个哨兵
-
log.warn("Cannot get slavesInfo from sentinel running @ " + hap + ". Reason: " + e
-
+ ". Trying next one.");
-
} finally {
-
if (jedis != null) {
-
jedis.close();
-
}
-
}
-
}
-
-
//没有可用的slave节点信息
-
if (alivedSalaves.isEmpty()) {
-
if (sentinelAvailable) {
-
throw new JedisException("Can connect to sentinel, but " + masterName
-
+ " cannot find any redis slave");
-
} else {
-
throw new JedisConnectionException("All sentinels down");
-
}
-
}
-
-
return alivedSalaves;
-
}
-
-
@Override
-
public void passivateObject(PooledObject<Jedis> pooledJedis) {
-
}
-
-
/**
-
* 检查jedis客户端是否有效
-
* @param pooledJedis 池中对象
-
* @return true有效 false无效
-
*/
-
@Override
-
public boolean validateObject(PooledObject<Jedis> pooledJedis) {
-
final BinaryJedis jedis = pooledJedis.getObject();
-
try {
-
//是否TCP连接 && 是否ping通 && 是否slave角色
-
boolean result = jedis.isConnected()
-
&& jedis.ping().equals("PONG")
-
&& jedis.info("Replication").contains("role:slave");
-
-
log.debug("ValidateObject Jedis=["+jedis+"] host=[ " + jedis.getClient().getHost() +
-
"] port=[" + jedis.getClient().getPort() +"] return=[" + result + "]");
-
return result;
-
} catch (final Exception e) {
-
log.warn("ValidateObject error jedis client cannot use", e);
-
return false;
-
}
-
}
-
-
}
使用的时候跟原来一样,创建slave连接池就可以了。
-
JedisPoolConfig jedisPoolConfig = new JedisPoolConfig();
-
//连接池中最大对象数量
-
jedisPoolConfig.setMaxTotal(100);
-
//最大能够保持idel状态的对象数
-
jedisPoolConfig.setMaxIdle(1);
-
//最小能够保持idel状态的对象数
-
jedisPoolConfig.setMinIdle(1);
-
//当池内没有可用资源,最大等待时长
-
jedisPoolConfig.setMaxWaitMillis(3000);
-
//表示有一个idle object evitor线程对object进行扫描,调用validateObject方法.
-
jedisPoolConfig.setTestWhileIdle(true);
-
//evitor线程对object进行扫描的时间间隔
-
jedisPoolConfig.setTimeBetweenEvictionRunsMillis(30000);
-
//表示对象的空闲时间,如果超过这个时间对象没有被使用则变为idel状态
-
//然后才能被idle object evitor扫描并驱逐;
-
//这一项只有在timeBetweenEvictionRunsMillis大于0时和setTestWhileIdle=true时才有意义
-
//-1 表示对象不会变成idel状态
-
jedisPoolConfig.setMinEvictableIdleTimeMillis(60000);
-
//表示idle object evitor每次扫描的最多的对象数;
-
jedisPoolConfig.setNumTestsPerEvictionRun(10);
-
-
//在从池中获取对象时调用validateObject方法检查
-
jedisPoolConfig.setTestOnBorrow(false);
-
//在把对象放回池中时调用validateObject方法检查
-
jedisPoolConfig.setTestOnReturn(false);
-
-
Set<String> sentinels = new HashSet<>(Arrays.asList(
-
"192.168.80.112:26379",
-
"192.168.80.113:26379",
-
"192.168.80.114:26379"
-
));
-
-
JedisSentinelSlavePool pool = new JedisSentinelSlavePool("mymaster", sentinels, jedisPoolConfig);
与Spring集成,分别创建不同的对象即可,在程序中查询接口可以先走slave进行查询,查询不到在查询master, master也没有则写入缓存,返回数据,下载在查询slave就同步过去啦,这样一来redis的性能会大幅度的提升。
-
@Primary
-
@Bean(name = "redisTemplateMaster")
-
public RedisTemplate<Object, Object> redisTemplateMaster() {
-
RedisTemplate<Object, Object> template = new RedisTemplate<>();
-
template.setConnectionFactory(redisMasterConnectionFactory());
-
template.setKeySerializer(new StringRedisSerializer());
-
template.setValueSerializer(new GenericJackson2JsonRedisSerializer());
-
return template;
-
}
-
-
@Bean(name = "redisTemplateSlave")
-
public RedisTemplate<Object, Object> redisTemplateSlave() {
-
RedisTemplate<Object, Object> template = new RedisTemplate<>();
-
template.setConnectionFactory(redisSlaveConnectionFactory());
-
template.setKeySerializer(new StringRedisSerializer());
-
template.setValueSerializer(new GenericJackson2JsonRedisSerializer());
-
return template;
-
}