在Redis集群中使用pipeline批量插入的实现方法

由于项目中需要使用批量插入功能, 所以在网上查找到了Redis 批量插入可以使用pipeline来高效的插入, 示例代码如下:

String key = "key";

Jedis jedis = new Jedis("xx.xx.xx.xx");

Pipeline p = jedis.pipelined();

List String myData = .... //要插入的数据列表

for(String data: myData){

 p.hset(key, data);

p.sync();

jedis.close();

但实际上遇到的问题是,项目上所用到的Redis是集群,初始化的时候使用的类是JedisCluster而不是Jedis. 去查了JedisCluster的文档, 并没有发现提供有像Jedis一样的获取Pipeline对象的 pipelined()方法.

Google了一下, 发现了解决方案.

Redis集群规范有说: Redis 集群的键空间被分割为 16384 个槽（slot）， 集群的最大节点数量也是 16384 个。每个主节点都负责处理 16384 个哈希槽的其中一部分。当我们说一个集群处于“稳定”（stable）状态时， 指的是集群没有在执行重配置（reconfiguration）操作， 每个哈希槽都只由一个节点进行处理。

所以我们可以根据要插入的key知道这个key所对应的槽的号码, 再通过这个槽的号码从集群中找到对应Jedis. 具体实现如下

//初始化得到了jedis cluster, 如何获取HostAndPort集合代码就不写了

Set HostAndPort nodes = .....

JedisCluster jedisCluster = new JedisCluster(nodes);


 private static TreeMap Long, String getSlotHostMap(String anyHostAndPortStr) {

 TreeMap Long, String tree = new TreeMap Long, String 

 String parts[] = anyHostAndPortStr.split(":");

 HostAndPort anyHostAndPort = new HostAndPort(parts[0], Integer.parseInt(parts[1]));

 try{

 Jedis jedis = new Jedis(anyHostAndPort.getHost(), anyHostAndPort.getPort());

 List Object list = jedis.clusterSlots();

 for (Object object : list) {

 List Object list1 = (List Object ) object;

 List Object master = (List Object ) list1.get(2);

 String hostAndPort = new String((byte[]) master.get(0)) + ":" + master.get(1);

 tree.put((Long) list1.get(0), hostAndPort);

 tree.put((Long) list1.get(1), hostAndPort);

 jedis.close();

 }catch(Exception e){

 return tree;

上面这几步可以在初始化的时候就完成. 不需要每次都调用, 把nodeMap和slotHostMap都定义为静态变量.

//获取槽号

int slot = JedisClusterCRC16.getSlot(key); 

//获取到对应的Jedis对象

Map.Entry Long, String entry = slotHostMap.lowerEntry(Long.valueOf(slot));

Jedis jedis = nodeMap.get(entry.getValue()).getResource();

建议上面这步操作可以封装成一个静态方法, 比如命名为public static Jedis getJedisByKey(String key) 之类的. 意思就是在集群中, 通过key获取到这个key所对应的Jedis对象.

这样再通过上面的jedis.pipelined();来就可以进行批量插入了.

注:这个方法是从Google上搜来的, 直到目前我使用起来还没发现什么问题. 如果哪位大神发现有什么不对的地方欢迎提出来.

本篇文章到此结束，如果您有相关技术方面疑问可以联系我们技术人员远程解决，感谢大家支持本站！

我想要获取技术服务或软件
服务范围：MySQL、ORACLE、SQLSERVER、MongoDB、PostgreSQL 、程序问题
服务方式：远程服务、电话支持、现场服务，沟通指定方式服务
技术标签：数据恢复、安装配置、数据迁移、集群容灾、异常处理、其它问题

本站部分文章参考或来源于网络，如有侵权请联系站长。
数据库远程运维 在Redis集群中使用pipeline批量插入的实现方法