Redis集群是沒法執(zhí)行批量操作命令的，如mget，pipeline等。這是因?yàn)閞edis將集群劃分為16383個(gè)哈希槽，不同的key會劃分到不同的槽中。但是，Jedis客戶端提供了計(jì)算key的slot方法，已經(jīng)slot和節(jié)點(diǎn)之間的映射關(guān)系，通過這兩個(gè)數(shù)據(jù)，就可以計(jì)算出每個(gè)key所在的節(jié)點(diǎn)，然后使用pipeline獲取數(shù)據(jù)。具體代碼如下：
@Configuration
public class JedisClusterConfig {
@Value("${spring.redis.cluster.nodes}")
private String clusterNodes;
@Value("${spring.redis.cache.commandTimeout}")
private Integer commandTimeout;
@Bean
public JedisCluster getJedisCluster() {
String[] serverArray = clusterNodes.split(",");
SetHostAndPort> nodes = new HashSet>();
for (String ipPort : serverArray) {
String[] ipPortPair = ipPort.split(":");
nodes.add(new HostAndPort(ipPortPair[0].trim(), Integer.valueOf(ipPortPair[1].trim())));
}
return new JedisCluster(nodes, commandTimeout);
}
}
@Component
public class JedisClusterUtil {
@Autowired
private JedisCluster jedisCluster;
@Resource(name = "redisTemplate4Json")
protected RedisTemplateString, Object> redisTemplate;
/**
* ZSet批量查詢
* @param keys
* @return
*/
public ListObject> batchZRange(ListString> keys) {
ListObject> resList = new ArrayList>();
if (keys == null || keys.size() == 0) {
return resList;
}
if (keys.size() == 1) {
BoundZSetOperationsString, Object> operations = redisTemplate.boundZSetOps(keys.get(0));
SetObject> set = operations.reverseRange(0, 0);
resList.add(set.iterator().next());
return resList;
}
MapJedisPool, ListString>> jedisPoolMap = getJedisPool(keys);
ListString> keyList;
JedisPool currentJedisPool = null;
Pipeline currentPipeline;
ListObject> res = new ArrayList>();
MapString, Object> resultMap = new HashMap>();
//執(zhí)行
for (Map.EntryJedisPool, ListString>> entry : jedisPoolMap.entrySet()) {
Jedis jedis = null;
try {
currentJedisPool = entry.getKey();
keyList = entry.getValue();
//獲取pipeline
jedis = currentJedisPool.getResource();
currentPipeline = jedis.pipelined();
for (String key : keyList) {
currentPipeline.zrevrange(key, 0, 0);
}
//從pipeline中獲取結(jié)果
res = currentPipeline.syncAndReturnAll();
currentPipeline.close();
for (int i = 0; i keyList.size(); i++) {
if (null == res.get(i)) {
resultMap.put(keyList.get(i), null);
} else {
SetObject> set = (SetObject>) res.get(i);
if (null == set || set.isEmpty()) {
resultMap.put(keyList.get(i), null);
} else {
byte[] byteStr = set.iterator().next().toString().getBytes();
Object obj = redisTemplate.getDefaultSerializer().deserialize(byteStr);
resultMap.put(keyList.get(i), obj);
}
}
}
} catch (Exception e) {
e.printStackTrace();
} finally {
returnResource(jedis, currentJedisPool);
}
}
resList = sortList(keys, resultMap);
return resList;
}
/**
* Value批量查詢
* @param keys
* @return
*/
public ListObject> batchGet(ListString> keys){
ListObject> resList = new ArrayList>();
if (keys == null || keys.size() == 0) {
return resList;
}
if (keys.size() == 1) {
BoundValueOperationsString, Object> operations = redisTemplate.boundValueOps(keys.get(0));
resList.add(operations.get());
return resList;
}
MapJedisPool, ListString>> jedisPoolMap = getJedisPool(keys);
ListString> keyList;
JedisPool currentJedisPool = null;
Pipeline currentPipeline;
ListObject> res = new ArrayList>();
MapString, Object> resultMap = new HashMap>();
for (Map.EntryJedisPool, ListString>> entry : jedisPoolMap.entrySet()) {
Jedis jedis = null;
try {
currentJedisPool = entry.getKey();
keyList = entry.getValue();
//獲取pipeline
jedis = currentJedisPool.getResource();
currentPipeline = jedis.pipelined();
for (String key : keyList) {
currentPipeline.get(key);
}
//從pipeline中獲取結(jié)果
res = currentPipeline.syncAndReturnAll();
currentPipeline.close();
for (int i = 0; i keyList.size(); i++) {
if (null == res.get(i)) {
resultMap.put(keyList.get(i), null);
} else {
byte[] byteStr = keyList.get(i).toString().getBytes();
Object obj = redisTemplate.getDefaultSerializer().deserialize(byteStr);
resultMap.put(keyList.get(i), obj);
}
}
} catch (Exception e) {
e.printStackTrace();
} finally {
returnResource(jedis, currentJedisPool);
}
}
resList = sortList(keys, resultMap);
return resList;
}
private MapJedisPool, ListString>> getJedisPool(ListString> keys){
//JedisCluster繼承了BinaryJedisCluster
//BinaryJedisCluster的JedisClusterConnectionHandler屬性
//里面有JedisClusterInfoCache，根據(jù)這一條繼承鏈，可以獲取到JedisClusterInfoCache
//從而獲取slot和JedisPool直接的映射
MetaObject metaObject = SystemMetaObject.forObject(jedisCluster);
JedisClusterInfoCache cache = (JedisClusterInfoCache) metaObject.getValue("connectionHandler.cache");
//保存地址+端口和命令的映射
MapJedisPool, ListString>> jedisPoolMap = new HashMap>();
JedisPool currentJedisPool = null;
ListString> keyList;
for (String key : keys) {
//計(jì)算哈希槽
int crc = JedisClusterCRC16.getSlot(key);
//通過哈希槽獲取節(jié)點(diǎn)的連接
currentJedisPool = cache.getSlotPool(crc);
//由于JedisPool作為value保存在JedisClusterInfoCache中的一個(gè)map對象中，每個(gè)節(jié)點(diǎn)的
//JedisPool在map的初始化階段就是確定的和唯一的，所以獲取到的每個(gè)節(jié)點(diǎn)的JedisPool都是一樣
//的，可以作為map的key
if (jedisPoolMap.containsKey(currentJedisPool)) {
jedisPoolMap.get(currentJedisPool).add(key);
} else {
keyList = new ArrayList>();
keyList.add(key);
jedisPoolMap.put(currentJedisPool, keyList);
}
}
return jedisPoolMap;
}
private ListObject> sortList(ListString> keys, MapString, Object> params) {
ListObject> resultList = new ArrayList>();
IteratorString> it = keys.iterator();
while (it.hasNext()) {
String key = it.next();
resultList.add(params.get(key));
}
return resultList;
}
/**
* 釋放jedis資源
*
* @param jedis
*/
public void returnResource(Jedis jedis, JedisPool jedisPool) {
if (jedis != null jedisPool != null) {
jedisPool.returnResource(jedis);
}
}
到此這篇關(guān)于redis 集群批量操作實(shí)現(xiàn)的文章就介紹到這了,更多相關(guān)redis 集群批量操作內(nèi)容請搜索腳本之家以前的文章或繼續(xù)瀏覽下面的相關(guān)文章希望大家以后多多支持腳本之家！
