Redis集群是沒法執行批量操作命令的，如mget，pipeline等。這是因為redis將集群劃分為16383個哈希槽，不同的key會劃分到不同的槽中。但是，Jedis客戶端提供了計算key的slot方法，已經slot和節點之間的映射關系，通過這兩個數據，就可以計算出每個key所在的節點，然后使用pipeline獲取數據。具體代碼如下：
@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>();
//執行
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中獲取結果
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中獲取結果
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，根據這一條繼承鏈，可以獲取到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) {
//計算哈希槽
int crc = JedisClusterCRC16.getSlot(key);
//通過哈希槽獲取節點的連接
currentJedisPool = cache.getSlotPool(crc);
//由于JedisPool作為value保存在JedisClusterInfoCache中的一個map對象中，每個節點的
//JedisPool在map的初始化階段就是確定的和唯一的，所以獲取到的每個節點的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);
}
}
到此這篇關于redis 集群批量操作實現的文章就介紹到這了,更多相關redis 集群批量操作內容請搜索腳本之家以前的文章或繼續瀏覽下面的相關文章希望大家以后多多支持腳本之家！
