背景
線上有一臺(tái)應(yīng)用使用單機(jī)redis作為緩存，在一次架構(gòu)巡檢時(shí)發(fā)現(xiàn)此單機(jī)使用redis在業(yè)務(wù)高峰期，接收請(qǐng)求的qps達(dá)到6+w/s，考慮到后期業(yè)務(wù)的增長(zhǎng)，應(yīng)用的穩(wěn)定性，需要調(diào)整架構(gòu)設(shè)計(jì)，解決此性能問(wèn)題。
方案
該應(yīng)用業(yè)務(wù)特點(diǎn)為，業(yè)務(wù)高峰期并發(fā)請(qǐng)求高，緩存內(nèi)存占用量較低（大約為5G）。目前應(yīng)用部署在云平臺(tái)上，若申請(qǐng)redis集群最小容量為256G，存在嚴(yán)重資源浪費(fèi)。決定新增一臺(tái)redis，在jedis客戶端使用ShardedJedisPool根據(jù)key做hash分片路由到兩臺(tái)單機(jī)redis上，降低單機(jī)redis的qps的方案。
具體實(shí)現(xiàn)
1.加入maven依賴
<dependency>
<groupId>redis.clients</groupId>
<artifactId>jedis</artifactId>
<version>2.10.2</version>
</dependency>
2.代碼實(shí)現(xiàn)
public interface ShardedJedisAction<T> {
T doAction(ShardedJedis paramShardedJedis);
}
public interface ShardedJedisClient {
<T> T execute(ShardedJedisAction<T> action);
}
public class ShardedJedisClientImpl implements ShardedJedisClient{
private static final Logger logger = LoggerFactory.getLogger(ShardedJedisClientImpl.class);
private static final ScheduledExecutorService shardedJedisPoolMonitorExecutor =
Executors.newSingleThreadScheduledExecutor(new ThreadFactoryBuilder()
.setNameFormat("ShardedJedisPoolMonitor-%d").setDaemon(false).build());
private ShardedJedisPool pool;
public void init(){
shardedJedisPoolMonitorExecutor.scheduleAtFixedRate(new Runnable() {
@Override
public void run() {
logger.info("ShardedJedisPoolMonitor enable,NumActive:{},NumIdle:{},NumWaiters:{}",pool.getNumActive(),pool.getNumIdle(),pool.getNumWaiters());
}
},60,15, TimeUnit.SECONDS);
}
@Override
public <T> T execute(ShardedJedisAction<T> action) {
T result = null;
try {
result = executeAction(action);
} catch (Exception e) {
logger.error("ShardedJedisClient execute error:{}",e);
}
return result;
}
private <T> T executeAction(ShardedJedisAction<T> action) {
ShardedJedis shardedJedis = null;
try {
shardedJedis = pool.getResource();
Object object = action.doAction(shardedJedis);
return (T)object;
} catch (JedisConnectionException jex) {
if (shardedJedis != null) {
try {
pool.returnBrokenResource(shardedJedis);
} catch (Exception ex) {
logger.warn("Can not return broken resource.", ex);
}
shardedJedis = null;
}
throw jex;
}
finally{
if (shardedJedis != null) {
try {
pool.returnResource(shardedJedis);
} catch (Exception ex) {
logger.warn("Can not return resource.", ex);
}
}
}
}
public ShardedJedisPool getPool() {
return pool;
}
public void setPool(ShardedJedisPool pool) {
this.pool = pool;
}
public void destroy() {
shardedJedisPoolMonitorExecutor.shutdown();
pool.destroy();
}
}
public class ShardedJedisTemplate {
private static final Logger logger = LoggerFactory.getLogger(ShardedJedisTemplate.class);
private ShardedJedisClient shardedClient;
private RedisSerializer keySerializer;
private RedisSerializer valueSerializer;
/**
* 設(shè)置單個(gè)值
* @param key
* @param value
* @return
*/
public void set(final String key,final Object value){
final byte[] rawKey = rawKey(key);
final byte[] rawValue = rawValue(value);
shardedClient.execute(new ShardedJedisAction<String>() {
@Override
public String doAction(ShardedJedis shardedJedis) {
logger.debug("key:{},value:{},action:set,redis ShardInfo host:{},port{}", key, JSON.toJSONString(value), shardedJedis.getShardInfo(rawKey).getHost(), shardedJedis.getShardInfo(rawKey).getPort());
return shardedJedis.set(rawKey, rawValue);
}
});
}
/**
* 獲取單個(gè)值
*
* @param key
* @return
*/
public <T> T get(String key) {
final byte[] rawKey = rawKey(key);
byte[] result = shardedClient.execute(new ShardedJedisAction<byte[]>() {
@Override
public byte[] doAction(ShardedJedis shardedJedis) {
byte[] rawValue = shardedJedis.get(rawKey);
logger.debug("key:{},value:{},action:get,redis ShardInfo host:{},port{}",key, JSON.toJSONString(rawValue),shardedJedis.getShardInfo(rawKey).getHost(),shardedJedis.getShardInfo(rawKey).getPort());
return rawValue;
}
});
return (T) deserializeValue(result);
}
/**
* key轉(zhuǎn)字節(jié)
*
* @param key
* @return
*/
private byte[] rawKey(String key) {
Assert.notNull(key, "non null key required");
if (keySerializer == null) {
return key.getBytes();
}
return keySerializer.serialize(key);
}
/**
* value轉(zhuǎn)字節(jié)
*
* @param value
* @return
*/
private byte[] rawValue(Object value) {
if (valueSerializer == null && value instanceof byte[]) {
return (byte[]) value;
}
return valueSerializer.serialize(value);
}
private Object deserializeValue(byte[] value) {
if (valueSerializer == null) {
return value;
}
return valueSerializer.deserialize(value);
}
public ShardedJedisClient getShardedClient() {
return shardedClient;
}
public void setShardedClient(ShardedJedisClient shardedClient) {
this.shardedClient = shardedClient;
}
public RedisSerializer getKeySerializer() {
return keySerializer;
}
public void setKeySerializer(RedisSerializer keySerializer) {
this.keySerializer = keySerializer;
}
public RedisSerializer getValueSerializer() {
return valueSerializer;
}
public void setValueSerializer(RedisSerializer valueSerializer) {
this.valueSerializer = valueSerializer;
}
}
@Slf4j
@Configuration
public class ShardedJedisConfiguration {
@Value("#{'${redis.uri.list}'.split(',')}")
private List<String> uriList ;
@Value("{redis.pool.minIdle")
private int minIdle;
@Value("${redis.pool.maxIdle}")
private int maxIdle;
@Value("${redis.pool.maxTotal}")
private int maxTotal;
@Value("${redis.pool.maxWaitMillis}")
private int maxWaitMillis;
@Value("${redis.pool.testOnBorrow}")
private Boolean testOnBorrow;
@Value("${redis.pool.timeBetweenEvictionRunsMillis}")
private int timeBetweenEvictionRunsMillis;
@Value("${redis.pool.minEvictableIdleTimeMillis}")
private int minEvictableIdleTimeMillis;
@Bean
public JedisPoolConfig jedisPoolConfig() {
JedisPoolConfig jedisPoolConfig = new JedisPoolConfig();
jedisPoolConfig.setMinIdle(minIdle);
jedisPoolConfig.setMaxIdle(maxIdle);
jedisPoolConfig.setMaxTotal(maxTotal);
jedisPoolConfig.setMaxWaitMillis(maxWaitMillis);
jedisPoolConfig.setTestOnBorrow(testOnBorrow);
jedisPoolConfig.setTimeBetweenEvictionRunsMillis(timeBetweenEvictionRunsMillis);
jedisPoolConfig.setMinEvictableIdleTimeMillis(minEvictableIdleTimeMillis);
return jedisPoolConfig;
}
@Bean
public ShardedJedisPool shardedJedisPool(@Qualifier("jedisPoolConfig") JedisPoolConfig jedisPoolConfig){
List<JedisShardInfo> jedisShardInfos = uriList.stream().map(uri->new JedisShardInfo(uri)).collect(Collectors.toList());
return new ShardedJedisPool(jedisPoolConfig, jedisShardInfos);
}
@Bean(name = "shardedJedisClient")
public ShardedJedisClient shardedJedisClient(@Qualifier("shardedJedisPool") ShardedJedisPool shardedJedisPool){
ShardedJedisClientImpl shardedJedisClient = new ShardedJedisClientImpl();
shardedJedisClient.setPool(shardedJedisPool);
return shardedJedisClient;
}
@Bean(name = "shardedRedisTemplate")
public ShardedJedisTemplate shardedRedisTemplate(@Qualifier("shardedJedisClient") ShardedJedisClient shardedJedisClient) {
ShardedJedisTemplate shardedJedisTemplate = new ShardedJedisTemplate();
shardedJedisTemplate.setShardedClient(shardedJedisClient);
shardedJedisTemplate.setKeySerializer(new StringRedisSerializer());
shardedJedisTemplate.setValueSerializer(new JdkSerializationRedisSerializer());
return shardedJedisTemplate;
}
}
踩坑經(jīng)歷
jedis的2.10.2和3.x版本在釋放jedis連接池的寫法不一樣。若maven依賴jedis版本2.10.2，但代碼使用官方3.x的寫法，在高并發(fā)的請(qǐng)求下會(huì)出現(xiàn)連接池不能釋放的問(wèn)題。具體可以參考官方demo：https://github.com/redis/jedis/
實(shí)踐結(jié)果
目前應(yīng)用已穩(wěn)定運(yùn)行半年
