Redis中的批量操作Pipeline

桃扇骨 2022-10-30 13:21 156阅读 0赞

# 单机 #

使用正常和pipeline的简单对比

public class PiplineTest { 
        private static int count = 10000;
     
        public static void main(String[] args){ 
            useNormal();
            usePipeline();
        }
     
        public static void usePipeline(){ 
            ShardedJedis jedis = getShardedJedis();
            ShardedJedisPipeline pipeline = jedis.pipelined();
            long begin = System.currentTimeMillis();
            for(int i = 0;i<count;i++){ 
                pipeline.set("key_"+i,"value_"+i);
            }
            pipeline.sync();
            jedis.close();
            System.out.println("usePipeline total time:" + (System.currentTimeMillis() - begin));
        }
     
        public static void useNormal(){ 
            ShardedJedis jedis = getShardedJedis();
            long begin = System.currentTimeMillis();
            for(int i = 0;i<count;i++){ 
                jedis.set("key_"+i,"value_"+i);
            }
            jedis.close();
            System.out.println("useNormal total time:" + (System.currentTimeMillis() - begin));
        }
     
        public static ShardedJedis getShardedJedis(){ 
            JedisPoolConfig poolConfig = new JedisPoolConfig();
            poolConfig.setMaxTotal(2);
            poolConfig.setMaxIdle(1);
            poolConfig.setMaxWaitMillis(2000);
            poolConfig.setTestOnBorrow(false);
            poolConfig.setTestOnReturn(false);
            JedisShardInfo info1 = new JedisShardInfo("127.0.0.1",6379);
            JedisShardInfo info2 = new JedisShardInfo("127.0.0.1",6379);
            ShardedJedisPool pool = new ShardedJedisPool(poolConfig, Arrays.asList(info1,info2));
            return pool.getResource();
        }
    }

输出结果：

useNormal total time:772
    usePipeline total time:112

# 集群 #

参考文献：  
https://blog.csdn.net/Vi\_NSN/article/details/78704587  
https://www.cnblogs.com/drwong/p/4825752.html  
建议两篇看看，且对比下

批量插入：

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);
    
    
    
    Map<String, JedisPool> nodeMap = jedisCluster.getClusterNodes();
    
    String anyHost = nodeMap.keySet().iterator().next();
    
    //getSlotHostMap方法在下面有
    
    TreeMap<Long, String> slotHostMap = getSlotHostMap(anyHost);

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上搜来的, 直到目前我使用起来还没发现什么问题. 如果哪位大神发现有什么不对的地方欢迎提出来